OJ在线评测系统 微服务 用分布式消息队列 RabbitMQ 解耦判题服务和题目服务 手搓交换机和队列 实现项目异步化

时间：2024-10-08 07:00:12

消息队列解耦 项目异步化

分布式消息队列

分布式消息队列是一种用于异步通信的系统，它允许不同的应用程序或服务之间传递消息。消息队列的核心理念是将消息存储在一个队列中，发送方可以将消息发送到队列，而接收方则可以在适当的时候从队列中读取消息。这种机制有助于解耦应用程序，提高系统的可扩展性和可靠性。

主要特点：

  1. 异步通信：发送方和接收方可以在不同的时间工作，不必直接交互。

  2. 负载均衡：通过将消息分发到多个消费者，可以有效利用系统资源。

  3. 消息持久化：许多消息队列系统支持将消息存储在磁盘上，以防数据丢失。

  4. 顺序处理：某些队列支持按顺序处理消息，确保消息的处理顺序。

  5. 容错性：分布式架构增强了系统的容错能力，可以在部分组件故障时继续工作。

常见的分布式消息队列系统：

  • Apache Kafka：高吞吐量、可扩展的消息队列，常用于大数据处理。

  • RabbitMQ：支持多种消息协议，易于使用，适合复杂的路由场景。

  • ActiveMQ：功能丰富，支持多种编程语言和消息协议。

分布式消息队列在微服务架构、事件驱动架构等场景中广泛应用，可以有效提高系统的灵活性和可维护性。

我们就用RabbitMQ去改造项目 解耦判题服务 题目服务

题目服务只需要向消息队列中发信息

判题服务从消息队列中取消息去执行判题

然后异步更新数据库即可

我们的题目服务和判题服务需要引入rabbitMQ

先引入消息队列的Java客户端

先引入依赖 amqp的客户端

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

配置一下

  rabbitmq:
    host: localhost
    port: 5672
    password: guest
    username: guest

我们要创建交换机和队列

先启动生产者 的消息队列

package com.yupi.yuojbackendjudgeservice.rabbitmq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import lombok.extern.slf4j.Slf4j;

/**
 * 用于创建测试程序用到的交换机和队列（只用在程序启动前执行一次）
 */
@Slf4j
public class InitRabbitMq {

    public static void doInit() {
        try {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            Connection connection = factory.newConnection();
            
            // 创建交换机 用于发 收信息
            Channel channel = connection.createChannel();
            String EXCHANGE_NAME = "code_exchange";
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");

            // 创建队列 随机分配一个队列名称
            String queueName = "code_queue";
            channel.queueDeclare(queueName, true, false, false, null);
            channel.queueBind(queueName, EXCHANGE_NAME, "my_routingKey");
            log.info("消息队列启动成功");
        } catch (Exception e) {
            log.error("消息队列启动失败");
        }
    }

    public static void main(String[] args) {
        doInit();
    }
}

在启动类里面可以看见

package com.yupi.yuojbackendjudgeservice;

import com.yupi.yuojbackendjudgeservice.rabbitmq.InitRabbitMq;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
@EnableAspectJAutoProxy(proxyTargetClass = true, exposeProxy = true)
@ComponentScan("com.yupi")
@EnableDiscoveryClient
@EnableFeignClients(basePackages = {"com.yupi.yuojbackendserviceclient.service"})
public class YuojBackendJudgeServiceApplication {

    public static void main(String[] args) {
        // 初始化消息队列，先注释掉，改用 Bean 的方式初始化消息队列（InitRabbitMqBean.java）
//        InitRabbitMq.doInit();
        SpringApplication.run(YuojBackendJudgeServiceApplication.class, args);
    }

}

接下来要把生产者的消息扔到交换机里面

package com.yupi.yuojbackendquestionservice.rabbitmq;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class MyMessageProducer {

    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 发送消息
     * @param exchange
     * @param routingKey
     * @param message
     */
    public void sendMessage(String exchange, String routingKey, String message) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message);
    }

}

写一个接收消息的代码

package com.yupi.yuojbackendjudgeservice.rabbitmq;

import com.rabbitmq.client.Channel;
import com.yupi.yuojbackendjudgeservice.judge.JudgeService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
@Slf4j
public class MyMessageConsumer {

    @Resource
    private JudgeService judgeService;

    // 指定程序监听的消息队列和确认机制
    @SneakyThrows
    @RabbitListener(queues = {"code_queue"}, ackMode = "MANUAL")
    public void receiveMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        log.info("receiveMessage message = {}", message);
        long questionSubmitId = Long.parseLong(message);
        try {
            judgeService.doJudge(questionSubmitId);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            channel.basicNack(deliveryTag, false, false);
        }
    }

}

要传递的消息

是什么

传递的数据

相关文章