消息中间件MQ 基础知识、常见问题的解决方案

迈不过友情╰ 2023-02-12 14:22 9阅读 0赞

### 目录 ###

*   *   *  mq的优缺点
         *  mq常见的使用场景
         *  java消息服务 JMS
         *  主流消息中间件及其特点
         *  消息发送模型
         *  消息发送方式及其使用场景
         *  延迟消息
         *  如何保证消息生成、消费的顺序性
         *  如何保证消息的可靠性传输
         *  如何避免消息的重复消费
         *  消息堆积问题 | 大量消息堆积在broker中，应该如何处理

MQ，全称MessageQueue 消息队列，专门用作消息队列的中间组件叫做消息中间件，主要是用于应用之间通信，消费者、生产者可以使用不同语言编写。

### mq的优缺点 ###

优点 | 作用

*  模块解耦：各模块通过消息中间件来交换数据，无需耦合在一起，使各模块具有良好的扩展性。
 *  流量削峰：减轻下游服务的压力，保护下游服务不被流量冲垮
 *  异步化：可以使用mq实现异步操作，多操作并行，提高效率。只有在业务流程允许异步处理的情况下才能这么做，比如注册、登录流程中的发送邮件、短信。

缺点

*  系统可用性降低：系统引入的外部依赖越多，越容易出问题
 *  维护成本增加
 *  系统复杂度提高：需要考虑消息重复消费、保证可靠投递、保证消息消费的顺序性等问题
 *

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoeV8xODg4MzcwMTE2MQ_size_16_color_FFFFFF_t_70_pic_center]

### mq常见的使用场景 ###

*  秒杀系统存储待处理的订单，抢票系统存储需要抢票的用户，存储待推送的消息
 *  收集处理日志、监控信息
 *  收集处理用户行为数据

### java消息服务 JMS ###

JMS: 全称Java Message Service java消息服务，是sun公司早期推出的java消息标准，旨在为java应用提供统一的消息操作，包括create、send、receive。

JMS是java中的消息中间件接口，定义了java中标准消息传递的API，类似于JDBC，sun只规定标准、规范，由各MQ厂商提供具体实现。

JMS中的常见概念

*  消息message：数据对象
 *  队列queue：存储待消费消息的区域
 *  主题topic：一种支持发送消息给多个订阅者的机制

JMS的两种消息类型|发送模式

*  点对点 Point-to-Point（P2P）：消息只会被一个消费者消费。多个消费者可以订阅同一个topic，对于这个topic的一则消息，只有其中一个消费者消费。
 *  发布/订阅 Publish/Subscribe：消息可以被多个消费者消费。多个消费者都订阅同一个topic，对于这个topic的一则消息，订阅了此topic的消费者都会消费。

### 主流消息中间件及其特点 ###

1、ActiveMQ

*  Apache开源的消息中间件，比较古老，支持多种语言的客户端和协议，基于JMS Provider的实现。
 *  缺点：吞吐量不高，多队列的时候性能下降，存在消息丢失的情况，很少大规模使用。

2、Kafka

*  Apache开源的流处理平台，由Scala和Java编写；
 *  Kafka是一个分布式的消息发布/订阅系统，是一个流处理平台，高吞吐量，可以处理大规模网站中的动作流数据(网页浏览、搜索等用户行为)；
 *  并不是JMS规范的实现，严格意义上不属于MQ，只是提供了类似于JMS的特性，只支持常规的MQ功能；
 *  支持数据持久化，使用副本集机制实现数据冗余，保障数据尽可能不丢失；
 *  缺点：不支持批量、广播消息，不支持事务，定制的话需要掌握Scala

Kafka适合需要对大量数据进行收集处理，但对可靠性要求不高的场景，在大数据开发中用得多，在web开发中常用于处理大规模网站中的流数据，常见的使用场景如下

*  日志的收集处理
 *  收集实时监控信息、运行状态数据
 *  记录用户的访问行踪，比如用户访问的页面、点击的条目、浏览时长、搜索的关键词，发送给订阅者来分析、挖掘用户偏好、消费能力

3、RocketMQ

*  阿里开源的消息中间件，纯java编写；
 *  具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。支持事务，支持零拷贝技术、性能强劲，支持海量消息堆积， 支持指定次数和时间间隔的失败消息重发，支持延迟消息等；
 *  适合大规模的分布式应用，适合电商、互联网金融领域，在阿里内部大规模使用；
 *  缺点：目前大多只在阿里系广泛使用，文档、资料相对较少，社区相对不活跃。

4、RabbitMQ

*  基于AMQP协议，是AMQP的一个开源实现，使用Erlang语言编写
 *  各方面的折中，可靠性、稳定性、性能、吞吐量都不错，支持事务
 *  缺点：定制的话需要掌握Erlang

### 消息发送模型 ###

*  点对点：消息生产者向消息队列中发送了一个消息之后，只被一个消费者消费一次
 *  发布/订阅：消息生产者向频道发送一个消息之后，订阅此频道的消费者都可以接收并消费这条消息

发布/订阅模式和观察者模式的区别

*  在观察者模式中，观察者、主题都知道对方的存在；在发布与订阅模式中，生产者与消费者不知道对方的存在，它们之间通过频道进行通信。
 *  观察者模式是同步的，事件触发时主题会调用观察者的方法，然后等待方法返回；发布/订阅模式是异步的，生产者向频道发送一个消息之后，不关心消费者何时消费消息。

### 消息发送方式及其使用场景 ###

消息发送方式一般分三种

*  SYNC 同步发送：常用于有顺序要求的操作，需要等待消息发送后才能继续进行下一步操作。
 *  ASYNC 异步发送：常用于可以单独进行的操作，比如注册成功发送通知邮件、短信，下单发放积分、返优惠券。异步发送可以提供系统性能，支持更高的并发。
 *  ONEWAY 单向发送：生产者只负责发送消息，发送消息后不需要等待服务器响应，没有回调函数，即只管发送不管服务器是否接收到，可能存在消息丢失的情况，常用于对可靠性要求不高的操作，比如日志收集。

同步、异步发送服务器都会返回发送结果，可靠性高；单向发送不返回发送结果，可靠性低，但性能最高。

### 延迟消息 ###

延迟消息：生产者将消息发送到mq服务器后，但并不期望这条消息立马被投递，而推迟到当前时间点之后的某一个时间点进行投递、被消费者消费。

常见的使用场景

*  买票后发送一条延迟消息，在火车、航班、大巴、影院等指定班次开始前2小时发送消息通知用户，避免用户遗忘。
 *  超时未支付自动关闭订单，创建未支付订单时发送延迟消息，30分钟后投递给消费者消费，消息者先判断订单状态是否是已支付，未支付则关闭订单。

### 如何保证消息生成、消费的顺序性 ###

顺序消息：消息的消费顺序和生产顺序一致，比如

*  依次发送2条消息订单A、订单B，消费时也应该依次消费订单A、订单B。
 *  同一个订单相关的订单创建消息、订单支付消息、订单退款消息、订单物流消息、订单交易成功消息应该是有序的，应该按照顺序进行处理。

顺序消息有2种

*  全局顺序：topic中的所有消息都要有序，所有消息都严格按照FIFO进行发布、消费，因为是串行化处理，性能不行，吞吐量上不去，一般不用这种方式。
 *  局部顺序：根据业务需求保证部分queue|partition中的消息有序即可，平时所说的顺序消息一般是指局部顺序消息。

实现顺序消息

1、顺序发布

*  生产者将相关消息发送到同一个queue|partition中。rabbitmq可以指定exchange的类型、设置routing\_key，kafka可以自定义消息的分区路由规则，常见的比如%取模。
 *  不能使用异步发送方式，异步发送无法严格保证消息的发送顺序。

2、顺序消费

*  不能使用广播模式，同一个queue|partition中的消息只由一个消费者去消费；
 *  消费者不能使用多线程同时消费同一个queue|partition中的消息。

### 如何保证消息的可靠性传输 ###

producer端

*  不采用oneway单向发送，使用同步或者异步方式发送；
 *  使用失败重试机制；
 *  记录好消息投递日志，包括关键字段、投递时间、投递状态、重试次数。

broker端

*  多主多从架构，多机房，同步双写、异步刷盘

同步双写、异步刷盘指的是一个broker将数据写入内存后，同步到另一个broker节点，并异步持久化到硬盘。同步刷盘可靠性更高，但性能差一些，根据需要选择。

consumer端

*  消费完成后手工ack签收；
 *  因为producer端使用了重试机制，消费端需要做好幂等性处理，防止消息重复消费；
 *  记录好消息的消费日志，包括消息的元数据、消息体

### 如何避免消息的重复消费 ###

如果业务要求消息不被重复消费，则需要在消费端实现消息消费的幂等性，保证消息不被重复消费。

核心思路：使用数据库或redis记录已经消费的消息的id，消费之前先查数据库|redis，如果已消费过则不再消费。

*  消息id可以是消息的md5值，如果消息自带的id是全局唯一的，也可以直接使用消息本身的id；
 *  使用数据库会增加数据库压力，使用redis性能更好，但会在内存中保存大量的、长期有效的存储已消费消息id的key

常见三种实现方式

*  消费前先使用redis的setnx()设置消息id，设置成功说明之前没有消费过此条消费，进行消费；否则说明之前已经消费过这条消息。
 *  消费前先使用redis的incr(key)值自增（key是消息id），返回1说明之前没有消费过此条消费，进行消费；否则说明之前已经消费过这条消息。
 *  新建一张专门用于消息去重的数据表，记录已消费消息的id，消息id作为唯一索引（unique），消费时先查去重表中是否有对应记录，没有才进行消费。

### 消息堆积问题 | 大量消息堆积在broker中，应该如何处理 ###

核心思想：临时扩容，尽快消费完堆积的消息

*  检查Consumer是否存在问题，如果有问题则尽快修复，注意需要设置一批次提取的消息数量，防止大量消息涌入，冲垮Consumer；
 *  临时增加queue|partition，根据堆积的消息数量来确定要增加的queue|partition数量；
 *  编写临时分发程序，从旧queue|partition中快速读取消息分发到临时新加的queue|partition中；
 *  增加Consumer节点数量，使用更多的Consumer来消费消息。注意Consumer的数量应该小于等于queue|partition的数量，不然一些Consumer可能分配不到queue|partition；
 *  堆积的消息消费完毕后，将queue|partition、Consumer还原到正常数量。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoeV8xODg4MzcwMTE2MQ_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20210601115318846.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoeV8xODg4MzcwMTE2MQ==,size_16,color_FFFFFF,t_70#pic_center