RabbitMQ消息中间件基本概念

Posted 2022-04-27 ly甲烷

RabbitMQ的使用场景

1. 异步处理。 提升异步处理能力
   例如： 原本业务是 : 用户注册信息写入数据库（50ms–> 发送邮件通知（50ms）–> 发送短信通知（50ms） 全部完成，用户响应150ms
   如果把后两个发生邮件和短信的任务，开两个线程异步发送， 用户响应就是100ms。
   最佳处理方式是： 数据写入数据库后，直接把后两个服务放入消息队列（5ms），然后直接响应回去，用户响应55ms。消息队列自己慢慢去发短信，发邮件。就算发短信、邮件失败也没关系，因为数据库已经保存了数据
2. 应用解耦
   例如： 原本业务是： 用户下订单，调用远程库存服务减库存。 如果有一天减库存的方式变了，原本需要三个参数，现在需要5个参数。那我们订单系统也得修改源代码，重新上线。
   有了消息队列， 下订单只需要给消息队列写入消息，那个用户下来那个订单，不需要关注库存系统的接口。库存系统订阅这个消息，收到消息去分析怎么做。
3. 流量控制， 流量削峰
   例如： 原本业务是： 秒杀业务， 瞬间百万并发，然后处理业务又比较慢，大量请求阻塞在服务器，导致服务器资源耗尽。
   有了消息队列，请求全放队列里面， 返回根据服务器能力，挨个处理，不会导致服务器宕机。

消息服务概述

• 消息代理：安装了消息中间件的服务器
• 目的地：消息代理接管消息后，保证消息传递到指定目的地

目的地

1. 队列： 点对点消息通信， 消息只有唯一发送者和接受者，但是可以用多个接收者（抢）
2. 主题：发布 / 订阅 消息通信。 如果多个接收者监听这个主题，那么就会在消息到达时同时接收到消息

JMS Java消息服务
基于JVM消息代理的规范， ActiveMQ 、HornetMQ 是JMS实现

AMQP 高级消息队列协议
兼容JMS， RabbitMQ是AMQP的实现

RabbitMQ概念

由Exchange、 Queue、 RoutingKey 三个才能决定一个从EXchange 到 Queue 的唯一的线路。
使用 AmqpAdmin 创建、删除 Exchange、 Queue、 Binding
使用RabbitTemplate 发消息（如果发送的消息是个对象，会使用序列化机制，将对象写出去。也可以把他变成Json发出去）
使用注解接收消息

常用的交换机

1. direct ： 如果路由键完全匹配，消息就投递到相应的队列。
2. fanout ： 如果交换机收到消息，将会广播到所有绑定的队列上。
3. topic ： 使不同源头的消息到达一部分队列。 可以使用通配符，* 匹配一个单词， # 匹配0或1个单词。

RabbitMQ工作模式

1. 简单模式（simple）： 一个生成者，一个消费者，一个队列
2. 工作模式（work）： 多个相互竞争的消费者绑定到一个队列上去
3. 发布订阅（publish / subscribe）： 同时向多个消费者发送消息（类似广播），生产者发送消息到交换机，在到不同的队列，所有订阅消息的消费者都能收到消息。
4. 路由模式（routing）：根据路由键选择性给多个消费者发送消息
5. 主题模式（topic）： 根据路由键匹配规则选择性给多个消费者发送消息的模式

【重点】 RabbitMQ消息确认机制-可靠抵达

生产者 到 消息代理， 交换机到队列、队列到消费者都可能出现意外，必须保证可靠抵达
使用事务消息， 性能下降250倍， 所以引入消息确认机制

1.发送方确认机制

1. 先在配置文件中开启发送方确认模式， 再设置确认回调。
2. 只要消息到达Broker， 就会触发 confirmCallback， 表示服务器成功收到消息。
   【做好日志记录，每个消息是否收到都应在数据库中记录，定期扫描未发送成功的消息进行重发】
3. 在配置文件中开启发送端消息抵达队列确认， 再设置确认回调。
   【如果触发回调，在回调中修改日志状态为发送失败，再次进行重发】
4. 消息成功抵达Broker，不一定能成功投递到队列， 当交换机未能成功将消息投递到队列，就会触发returnCallback

2. 消费端确认机制（保证每个消息被正确消费，才能删除消息）

默认是自动ACK模式。只要消息收到，客户端会自动确认，服务端就会移除这个消息
自动确认的问题：收到很多消息，自动回复给服务器ack， 只有一个消息处理成功，宕机了。发生消息丢失。

1. 在配置文件中设置手动ACK模式
2. channe.basicAck ( deliveryTag ) ： 手动确认， broker移除此消息
3. channe.basicNack( deliveryTag ，false, true) ：否定确认， 可以指定是否丢弃消息，可以批量（批量之前的都否）
4. channe.reject() : 否定确认，不能批量。
   【设置手动ACK， 如果服务器出现宕机或bug没有正确ack，消息会重新入队】

发生方确认机制 + 消费端确认机制保证消息百分百不丢失。

【重点】 消息积压如何解决？

消息积压原因：

1. 消费者宕机一段时间， 而发送者还在源源不断的发送消息， 导致消息都积压在队列中；
2. 消费者消费能力不足， 不足以匹配发送者发送消息的频率；
3. 发送者发送流量过大。

解决方法：

1. 在发送端限制消息的发送；
2. 上线更多的消费者；
3. 上线一个专门处理消息的消费者， 将消息全部拿出来存入数据库， 之后再进行离线处理。

【重点】 如何保证消息不被重复消费？如何保证消息消费时的幂等性？

重复消费的原因
正常情况，消费完毕，回复一个确认消息给消息队列，消息队列就把这个消息删除了，但是如果因为忘了传输等故障，导致确认消息没有返回成功，消息队列就不知道自己消费过该消息，会将消息再次分发给消费者

幂等性：多次调用方法或者接口不会改变业务状态，可以保证重复调用的结果和单次调用的结果一致。
解决方法

1. 消费者的业务逻辑接口应设计成幂等性的。 比如扣库存的时候，带上唯一的订单号和状态标志
2. 使用防重表， 每一个消息都有业务唯一标识，处理过就不再处理了。

【重点】 RabbitMQ如何实现分布式事务？可靠消息最终一致性

<待补充>

谢谢你们的点赞、收藏、关注。让我们一起进步，不断创作更优秀的文章
转载请注明地址：https://blog.csdn.net/weixin_44179010/article/details/124264393