RocketMq知识总结及消息顺序性

为什么选择RocketMq？

几种MQ的区别：

• 如果业务场景对并发量要求不是太高（十万级、百万级），那这四种消息队列中，RabbitMQ 一定是你的首选。如果是大数据领域的实时计算、日志采集等场景，用 Kafka 是业内标准的，绝对没问题，社区活跃度很高，绝对不会黄，何况几乎是全世界这个领域的事实性规范
  

RocketMq的优点：

1. 延迟消息简单高效
2. 死信队列
3. 完善的事务消息功能

RocketMq的基本概念：

• Producer：消息的发送者；举例：发信者
• Consumer：消息接收者；举例：收信者
• Broker：暂存和传输消息；举例：邮局
• NameServer：管理Broker；举例：各个邮局的管理机构
• Topic：区分消息的种类；一个发送者可以发送消息给一个或者多个Topic；一个消息的接收者可以订阅一个或者多个Topic消息
• Message Queue：相当于是Topic的分区；用于并行发送和接收消息
  

按照发送的特点分：

• 同步发送

1. 同步发送，线程阻塞，投递completes阻塞结束
2. 如果发送失败，会在默认的超时时间3秒内进行重试，最多重试2次
3. 投递completes不代表投递成功，要check SendResult.sendStatus来判断是否投递成功
4. SendResult里面有发送状态的枚举：SendStatus，同步的消息投递有一个状态返回值的
5. retry的实现原理:只有ack的SendStatus=SEND_OK才会停止retry
   注意事项：发送同步消息且Ack为SEND_OK，只代表该消息成功的写入了MQ当中，并不代表该消息成功的被Consumer消费了

• 异步发送

1. 异步调用的话，当前线程一定要等待异步线程回调结束再关闭producer啊，因为是异步的，不会阻塞,提前关闭producer会导致未回调链接就断开了
2. 异步消息不retry，投递失败回调onException()方法，只有同步消息才会retry,源码参考 DefaultMQProducerImpl.class
3. 异步发送一般用于链路耗时较长，对 RT 响应时间较为敏感的业务场景，例如用户视频上传后通知启动转码服务，转码完成后通知推送转码结果等。
4. 单向发送
5. 消息不可靠，性能高，只负责往服务器发送一条消息，不会重试也不关心是否发送成功
6. 此方式发送消息的过程耗时非常短，一般在微秒级别

• 下表概括了三者的特点和主要区别。
  

按照使用功能特点分：(默认使用集群模式，广播模式可以通过消费模式设置)

1. 普通消息（订阅）
   github网址: https://github.com/apache/rocketmq-client-go/blob/v2.1.0/examples/producer/simple/main.go

   普通消息是我们在业务开发中用到的最多的消息类型，生产者需要关注消息发送成功即可，消费者消费到消息即可，不需要保证消息的顺序，所以消息可以大规模并发地发送和消费，吞吐量很高，适合大部分场景。

2. 顺序消息
   顺序消息分为分区顺序消息和全局顺序消息，全局顺序消息比较容易理解，也就是哪条消息先进入，哪条消息就会先被消费，符合我们的FIFO，很多时候全局消息的实现代价很大，所以就出现了分区顺序消息。分区顺序消息的概念可以如下图所示：
   
   我们通过对消息的key，进行hash，相同hash的消息会被分配到同一个分区里面，当然如果要做全局顺序消息，我们的分区只需要一个即可，所以全局顺序消息的代价是比较大的。

3. 延时消息 - 订单超时库存归还
   github网址: https://github.com/apache/rocketmq-client-go/blob/v2.1.0/examples/producer/delay/main.go

   延迟的机制是在 服务端实现的，也就是Broker收到了消息，但是经过一段时间以后才发送
   服务器按照1-N定义了如下级别： “1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”；若要发送定时消息，在应用层初始化Message消息对象之后，调用Message.setDelayTimeLevel(int level)方法来设置延迟级别，按照序列取相应的延迟级别，例如level=2，则延迟为5s

实现原理：

• 发送消息的时候如果消息设置了DelayTimeLevel,那么该消息会被丢到ScheduleMessageService.SCHEDULE_TOPIC这个Topic里面
• 根据DelayTimeLevel选择对应的queue
• 再把真实的topic和queue信息封装起来，set到msg里面
• 然后每个SCHEDULE_TOPIC_XXXX的每个DelayTimeLevelQueue，有定时任务去刷新，是否有待投递的消息
• 每 10s 定时持久化发送进度

4. 事务消息
   github网址: https://github.com/apache/rocketmq-client-go/blob/v2.1.0/examples/producer/transaction/main.go

   消息队列RocketMQ版提供的分布式事务消息适用于所有对数据最终一致性有强需求的场景。本文介绍消息队列RocketMQ版事务消息的概念、优势、典型场景、交互流程以及使用过程中的注意事项。

概念介绍

• 事务消息：消息队列RocketMQ版提供类似X或Open XA的分布式事务功能，通过消息队列RocketMQ版事务消息能达到分布式事务的最终一致。
• 半事务消息：暂不能投递的消息，发送方已经成功地将消息发送到了消息队列RocketMQ版服务端，但是服务端未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”状态，处于该种状态下的消息即半事务消息。
• 消息回查：由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，消息队列RocketMQ版服务端通过扫描发现某条消息长期处于“半事务消息”时，需要主动向消息生产者询问该消息的最终状态（Commit或是Rollback），该询问过程即消息回查。

分布式事务消息的优势

 消息队列RocketMQ版分布式事务消息不仅可以实现应用之间的解耦，
又能保证数据的最终一致性。同时，传统的大事务可以被拆分为小事务，
不仅能提升效率，还不会因为某一个关联应用的不可用导致整体回滚，
从而最大限度保证核心系统的可用性。在极端情况下，如果关联的某一个
应用始终无法处理成功，也只需对当前应用进行补偿或数据订正处理，而
无需对整体业务进行回滚。

消息的有序性

（原文链接: https://blog.csdn.net/weixin_47068446/article/details/129773098

• 消息有序指的是一类消息消费时，能按照发送的顺序来消费。例如：一个订单产生了三条消息分别是订单创建、订单付款、订单完成。消费时要按照这个顺序消费才能有意义，但是同时订单之间是可以并行消费的。RocketMQ可以严格的保证消息有序。

• 顺序消息分为全局顺序消息与部分顺序消息，全局顺序是指某个Topic下的所有消息都要保证顺序；部分顺序消息只要保证每一组消息被顺序消费即可。
  我们不用管不同的订单ID的消息之间的总体消费顺序，只需要保证同样订单ID的消息能按照订单创建、订单付款、订单完成这个顺序消费就可以了。

• 顺序消费实际上有两个核心点，一个是生产者有序存储，另一个是消费者有序消费。

生产者保证消息的有序性

• 先看如何实现生产者有序存储。我们知道RocketMQ中生产者生产的消息会放置在某个队列中，基于队列先进先出的特性天然的可以保证存入队列的消息顺序和拉取的消息顺序是一致的，因此，我们只需要保证一组相同的消息按照给定的顺序存入同一个队列中，就能保证生产者有序存储。

• 普通发送消息的模式下，生产者会采用轮询的方式将消费均匀的分发到不同的队列中，然后被不同的消费者消费，因为一组消息在不同的队列，此时就无法使用 RocketMQ 带来的队列有序特性来保证消息有序性了。

• 这个问题很好解决，因为RocketMQ支持生产者在投放消息的时候自定义投放策略，我们实现一个MessageQueueSelector接口，使用Hash取模法来保证同一个订单在同一个队列中就行了，即通过订单ID%队列数量得到该ID的订单所投放的队列在队列列表中的索引，然后该订单的所有消息都会被投放到这个队列中。

• 生产者发送消息的方法中就有一些添加队列选择器的方法，保证消息发送顺序。比如只有两个队列，那么订单ID为1,2,3的三组消息中，1、3组消息存放于第一个队列，而2组消息存放于第二个队列

• 另外，顺序消息必须使用同步发送的方式才能保证生产者发送的消息有序。实际上，采用队列选择器的方法不能保证消息的严格顺序，我们的目的是将消息发送到同一个队列中，如果某个broker挂了，那么队列就会减少一部分，如果采用取余的方式投递，将可能导致同一个业务中的不同消息被发送到不同的队列中，导致同一个业务的不同消息被存入不同的队列中，短暂的造成部分消息无序。同样的，如果增加了服务器，那么也会造成短暂的造成部分消息无序。

消费者端保证消息消费的有序性

• 生产者有序存储实现了，那么该如何实现消费者有序消费呢？RockerMQ的MessageListener回调函数提供了两种消费模式，有序消费模式MessageListenerOrderly和并发消费模式MessageListenerConcurrently。

• 在消费的时候，还需要保证消费者注册MessageListenerOrderly类型的回调接口实现顺序消费，如果消费者采用Concurrently并行消费，则仍然不能保证消息消费顺序。

• 实际上，每一个消费者的的消费端都是采用线程池实现多线程消费的模式，即消费端是多线程消费。虽然MessageListenerOrderly被称为有序消费模式，但是仍然是使用的线程池去消费消息。

• MessageListenerConcurrently是拉取到新消息之后就提交到线程池去消费，而MessageListenerOrderly则是通过加分布式锁和本地锁保证同时只有一条线程去消费一个队列上的数据。

即顺序消费模式使用3把锁来保证消费的顺序性：

• broker端的分布式锁：
  在负载均衡的处理新分配队列的updateProcessQueueTableInRebalance方法，以及ConsumeMessageOrderlyService服务启动时的start方法中，都会尝试向broker申请当前消费者客户端分配到的messageQueue的分布式锁。

  broker端的分布式锁存储结构为ConcurrentMap<String/* group */, ConcurrentHashMap<MessageQueue, LockEntry>>，该分布式锁保证同一个consumerGroup下同一个messageQueue只会被分配给一个consumerClient。

  获取到的broker端的分布式锁，在client端的表现形式为processQueue. locked属性为true，且该分布式锁在broker端默认60s过期，而在client端默认30s过期，因此ConsumeMessageOrderlyService#start会启动一个定时任务，每过20s向broker申请分布式锁，刷新过期时间。而负载均衡服务也是每20s进行一次负载均衡。

  broker端的分布式锁最先被获取到，如果没有获取到，那么在负载均衡的时候就不会创建processQueue了也不会提交对应的消费请求了。

messageQueue的本地synchronized锁：

• 在执行消费任务的开头，便会获取该messageQueue的本地锁对象objLock，它是一个Object对象，然后通过synchronized实现锁定。
  这个锁的锁对象存储在MessageQueueLock.mqLockTable属性中，结构为ConcurrentMap<MessageQueue, Object>，所以说，一个MessageQueue对应一个锁，不同的MessageQueue有不同的锁。

  因为顺序消费也是通过线程池消费的，所以这个synchronized锁用来保证同一时刻对于同一个队列只有一个线程去消费它。

ProcessQueue的本地consumeLock：

• 在获取到broker端的分布式锁以及messageQueue的本地synchronized锁的之后，在执行真正的消息消费的逻辑messageListener#consumeMessage之前，会获取ProcessQueue的consumeLock，这个本地锁是一个ReentrantLock。

那么这把锁有什么作用呢？

• 在负载均衡时，如果某个队列C被分配给了新的消费者，那么当前客户端消费者需要对该队列进行释放，它会调用removeUnnecessaryMessageQueue方法对该队列C请求broker端分布式锁的解锁。

  而在请求broker分布式锁解锁的时候，一个重要的操作就是首先尝试获取这个messageQueue对应的ProcessQueue的本地consumeLock。只有获取了这个锁，才能尝试请求broker端对该messageQueue的分布式锁解锁。
  如果consumeLock加锁失败，表示当前消息队列正在消息，不能解锁。那么本次就放弃解锁了，移除消息队列失败，只有等待下次重新分配消费队列时，再进行移除。

  如果没有这把锁，假设该消息队列因为负载均衡而被分配给其他客户端B，但是由于客户端A正在对于拉取的一批消费消息进行消费，还没有提交消费点位，如果此时客户端A能够直接请求broker对该messageQueue解锁，这将导致客户端B获取该messageQueue的分布式锁，进而消费消息，而这些没有commit的消息将会发送重复消费。

  所以说这把锁的作用，就是防止在消费消息的过程中，该消息队列因为发生负载均衡而被分配给其他客户端，进而导致的两个客户端重复消费消息的行为。

日常个更新中…