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

消息队列解耦项目异步化

分布式消息队列

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

主要特点：

异步通信：发送方和接收方可以在不同的时间工作，不必直接交互。
负载均衡：通过将消息分发到多个消费者，可以有效利用系统资源。
消息持久化：许多消息队列系统支持将消息存储在磁盘上，以防数据丢失。
顺序处理：某些队列支持按顺序处理消息，确保消息的处理顺序。
容错性：分布式架构增强了系统的容错能力，可以在部分组件故障时继续工作。

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

Apache Kafka：高吞吐量、可扩展的消息队列，常用于大数据处理。
RabbitMQ：支持多种消息协议，易于使用，适合复杂的路由场景。
ActiveMQ：功能丰富，支持多种编程语言和消息协议。

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

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

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

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

然后异步更新数据库即可

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

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

先引入依赖 amqp的客户端

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

配置一下

  rabbitmq:host: localhostport: 5672password: guestusername: 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 {@Resourceprivate 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 {@Resourceprivate 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);}}}