消息队列解耦 项目异步化
分布式消息队列
分布式消息队列是一种用于异步通信的系统，它允许不同的应用程序或服务之间传递消息。消息队列的核心理念是将消息存储在一个队列中，发送方可以将消息发送到队列，而接收方则可以在适当的时候从队列中读取消息。这种机制有助于解耦应用程序，提高系统的可扩展性和可靠性。
主要特点：
-
异步通信：发送方和接收方可以在不同的时间工作，不必直接交互。
-
负载均衡：通过将消息分发到多个消费者，可以有效利用系统资源。
-
消息持久化：许多消息队列系统支持将消息存储在磁盘上，以防数据丢失。
-
顺序处理：某些队列支持按顺序处理消息，确保消息的处理顺序。
-
容错性：分布式架构增强了系统的容错能力，可以在部分组件故障时继续工作。
常见的分布式消息队列系统：
-
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);
}
}
}
要传递的消息
是什么
传递的数据
