初识RabbitMQ - 实践
初识RabbitMQ
- RabbitMQ安装和使用（docker，centos 7 ）
- RabbitMQ的介绍
- RabbitMQ的作用
- RabbitMQ的基础知识
- 架构
- 交换机类型和路由规则
- SpringBoot整合RabbitMQ
- 发送和接收消息（测试版）
- 发送消息的api
- 消息堆积问题
- 消息可靠性问题
- 生产者确认机制
- MQ消息可靠性
- 消费者确认机制
RabbitMQ安装和使用（docker，centos 7 ）
docker run \
-e RABBITMQ_DEFAULT_USER=root \
-e RABBITMQ_DEFAULT_PASS=214913 \ - 指定用户名和密码，用于登录RabbitMQ
-v mq-plugins:/plugins \ - 设置数据卷
--name mq - 为容器指定名称为"mq"
--hostname mq - 设置容器的主机名为"mq"，RabbitMQ推荐显式设置主机名以确保正确运行
-p 15672:15672 - 端口映射，将容器的15672端口（RabbitMQ管理界面）映射到宿主机的15672端口
-p 5672:5672 - 端口映射，将容器的5672端口（RabbitMQ AMQP协议端口）映射到宿主机的5672端口
--network cc\ - 不存在不会创建，会报错
-d \ - 以守护进程方式运行容器（后台运行）
rabbitmq:3.8-management - 使用RabbitMQ 3.8版本镜像，并且包含管理插件RabbitMQ管理界面的地址：ip:15672
这个貌似不需要开放虚拟机防火墙端口也可以访问（没试过），但是api端口需要开放。下面就是RabbitMQ的管理端页面，这里可以自己摸索一下。
RabbitMQ的介绍
RabbitMQ的作用
在一些场景下，比如用户支付，支付之后还需要修改订单状态，加积分，发通知，等一系列操作，如果我是同步调用，会引发一些问题。
- 性能低，支付业务逻辑处理完之后，还需要在进行其他业务，消耗时间。
- 耦合度高，如果加积分的代码修改了，可能支付业务这一块代码也需要修改
- 拓展性差，支付完之后，老板说在给客户来个京豆，优惠卷啥的，又要修改代码
如何解决呢，异步调用。
异步调用的模型结构：
- 发布者
- 消息代理（broker）,也就是RabbitMQ的扮演的角色
- 消费者
具体流程就是：用户完成支付之后，支付的业务代码发送一条消息到RabbitMQ,消费者监听并处理RabbitMQ中的消息。
Broker的选择不只RabbitMQ一种，还有kafka,RocketMQ等。可以了解。
RabbitMQ的基础知识
架构
- 虚拟主机（virtualHost） :起到数据隔离的作用，就像java的模块，每一个虚拟主机都有自己的交换机和队列
- 交换机（exchange）:生产者发送的消息由交换机决定投递到哪个队列
- 队列（queue）:生产者投递的消息会暂存在消息队列中，等待消费者处理
- 生产者：发送消息的一方
- 消费者：消费消息的一方
交换机类型和路由规则
- Fanout Exchange （广播模式）:会把接收到的消息广播到每一个跟其绑定的queue。
- Direct Exchange （定向路由）: 把消息按照规则路由到指定的Queue，可以根据这种特性实现类似Fanout的效果。
○ 每一个Queue都与Exchange 设置一个BindingKey
○ 发布者发布消息时，指定消息的RoutingKey
○ Exchange将消息路由到BingdingKey和消息的RoutingKey一致的队列 - Topic交换机：和上一个类似。
a. routingKey可以是多个单词的列表，可以以 . 来分割
b. BingdingKey 可以使用通配符： # ：代指0个或者是多个单词 * ： 代指一个单词。
SpringBoot整合RabbitMQ
1.导入依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>2.yml文件配置
rabbitmq:
port: 5672
virtual-host: your-VirtualHost # 虚拟主机
username: your_username # 用户名
password: your_password # 密码
listener: # 监听器配置
simple:
prefetch: 1 # 每次只处理一个消息
publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型
publisher-returns: true # 开启publisher return机制3.使用注解来声明交换机和队列
@RabbitListener(
// 绑定队列到交换机
bindings = @QueueBinding(
// 队列配置 ，durable = true 表示持久化
value = @Queue(
value = "petlife.board.queue",
name = "petlife.board.queue",
arguments = @Argument(name = "x-queue-mode", value = "lazy"),
// x-queue-mode: 队列模式，lazy 懒加载，表示队列中的消息会存储在磁盘中，而不是内存中
//value: 指定队列在 RabbitMQ 中的实际名称
//name: 指定队列在 Spring 容器中的 Bean 名称
durable = "true"),
// 交换机配置，
exchange = @Exchange(
value = "petlife.board.exchange",
// 交换机名称
type = ExchangeTypes.TOPIC),
// 交换机类型，topic 交换机
key = "board.#"))
// 路由键，匹配 board. 开头的路由键
public void listener(String message) {
//执行具体的业务逻辑
System.out.println("监听到消息：" + message);
}4.声明消息转换器
通过源码可以看到，默认是SimpleMessageConverter ,支持的是java原生序列化，不是Json序列化。
@SpringBootApplication
@ComponentScan(basePackages = {"com.petlife"})
public class PetLifeApplication {
public static void main(String[] args) {
SpringApplication.run(PetLifeApplication.class, args);
}
@Bean
public MessageConverter messageConverter(){
// 1.定义消息转换器 : springboot默认使用JDK序列化，这里使用Jackson2JsonMessageConverter
Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
// 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
jackson2JsonMessageConverter.setCreateMessageIds(true);
return jackson2JsonMessageConverter;
}
}发送和接收消息（测试版）
直接往队列中发送消息,当前虚拟主机中要有这个队列，不然会报错。
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class TestApplicationTests {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSimpleQueue() {
// 队列名称
String queueName = "simple.queue";
// 消息
String message = "hello, spring amqp!";
// 发送消息
rabbitTemplate.convertAndSend(queueName, message);
}
}import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class TestListener {
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(String message) {
System.out.println("TestListener.listenSimpleQueue: " + message);
}
}运行结果
发送消息的api
@Service
public class SimpleMessageService {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 1. 简单发送 - 使用默认交换机
*/
public void sendSimpleMessage(String queueName, Object message) {
rabbitTemplate.convertAndSend(queueName, message);
}
/**
* 2. 发送到指定交换机和路由键
*/
public void sendToExchange(String exchange, String routingKey, Object message) {
rabbitTemplate.convertAndSend(exchange, routingKey, message);
}
/**
* 3. 发送消息并设置消息属性
*/
public void sendWithProperties(String exchange, String routingKey, Object message) {
MessageProperties properties = new MessageProperties();
properties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
properties.setHeader("custom-header", "value");
properties.setExpiration("10000"); // 10秒过期
Message msg = MessageBuilder.withBody(JsonUtils.toJsonBytes(message))
.andProperties(properties)
.build();
rabbitTemplate.send(exchange, routingKey, msg);
}
/**
* 发送消息并处理确认和返回
*/
public void sendWithCallback(String exchange, String routingKey, Object message) {
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
// 异步处理确认结果
correlationData.getFuture().addCallback(
result -> {
if (result != null && result.isAck()) {
log.info("消息发送成功, ID: {}", correlationData.getId());
} else {
log.error("消息发送失败, ID: {}", correlationData.getId());
}
},
ex -> log.error("消息发送异常, ID: {}", correlationData.getId(), ex)
);
rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
}
}消息堆积问题
默认情况下，RabbitMQ会将消息一次轮询投递给绑定在队列上的每一个消费者，但这并没有考虑到消费者是否已经处理完消息，可能会出现消息堆积。
测试：
@SpringBootTest
class ListenerApplicationTests {
@Resource
private RabbitTemplate rabbitTemplate;
@Test
public void testSimpleQueue() {
// 队列名称
String queueName = "simple.queue";
// 消息
String message = "hello, spring amqp!";
for (int i = 0; i < 10; i++) {
rabbitTemplate.convertAndSend(queueName, message + i);
}
}
}@Component
public class TestListener {
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(String message) throws InterruptedException {
Thread.sleep(1000);
log.info("Simple queue1: {}", message);
}
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue2(String message) {
log.info("Simple queue2: {}", message);
}
}运行结果：
解决办法：
在yml配置中添加
listener: # 监听器配置
simple:
prefetch: 1 # 每次只处理一个消息开启后同样的代码再次运行
对于消息堆积问题，可以使用work模型
- 多个消费者绑定同一个队列可以加快消息处理速度
- 同一条消息只会被同一个消息处理
- 通过yml配置来控制消费者预取的消息数量，处理完一条在处理下一条，能者多劳。
消息可靠性问题
- MQ没有成功的发送消息，
- MQ发送消息了，但是还有没接收到，MQ挂了；
- 消息收到了，但是处理过程中出了异常。
这些都可能导致业务出现错误。
对于这些，RabbitMQ有哪些相应的机制来解决呢。
生产者确认机制
1，生产者重连
网络波动导致客户端连接MQ失败，可以配置失败重连机制。spingAMQP重试机制是阻塞式的，会影响业务性能。
connection-timeout: 1s # MQ超时连接时间
template:
retry:
enabled: true # 开启重试
initial-interval: 1000ms # 初始等待时间
max-attempts: 3 # 重试次数
multiplier: 2 # 下次等待时间 = initial - interval * multiplier2.生产者确认机制
发布者发送消息到RabbitMQ,怎么知道是否发送到了呢，他有两种机制，一个是Publisher Return ，一个是publisher-confirm。但是，比较消耗性能，所以一般不建议开启。开启之后，会根据消息的状态返回一个回执.● 消息投递到了MQ但是路由失败了，会通过 PR 返回路由异常原因，然后返回ACK，告知投递成功（就是发送到了交换机，但是没有队列接收，通常是routingkey或者是绑定配置出现了问题）
● 临时消息投递到了MQ，并且入队成功，返回ACK，告知通敌成功
● 持久消息投递到了MQ，并且入队成功完成持久化，返回ACK，告知投递成功
● 其他情况返回NACK，告知投递失败
其中ack和nack属于Publisher Confirm机制，ack是投递成功；nack是投递失败。而return则属于Publisher Return机制。
默认两种机制都是关闭状态，需要通过配置文件来开启。
MQ消息可靠性
默认情况下MQ的数据都是在内存存储的临时数据，重启后就会消失。为了保证数据的可靠性，必须配置数据持久化。
Spring amqp 默认情况下创建的交换机是持久的。（注解是如此，可以看源码，使用bean创建，没怎么注意）
1.交换机持久化
可以在控制台设置
之前使用注解来声明交换机和队列时，里面的参数都有讲解，这里就不说了。
在这里插入图片描述
2.队列持久化
队列也是一样的
3.消息持久化
消费者确认机制
当消费者处理消息结束后，应该向RabbitMQ发送一个回执，告知RabbitMQ自己消息处理状态。
- ack：成功处理消息，RabbitMQ从队列中删除该消息
- nack：消息处理失败，RabbitMQ需要再次投递消息
- reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息
…未完待续
