RocketMQ 基础

Ref: https://www.cnblogs.com/javazhiyin/p/13327925.html

RocketMQ 角色及其作用

NameServer

NameServer 是一个非常简单的 Topic 路由注册中心，其角色类似 Dubbo 中的 zookeeper，支持 Broker 的动态注册与发现。主要包括两个功能：

• Broker 管理：NameServer 接受 Broker 集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查 Broker 是否存活；
• 路由信息管理：每个 NameServer 将保存关于 Broker 集群的整个路由信息和用于客户端查询的队列信息。然后 Producer 和 Conumser 通过 NameServer 就可以知道整个 Broker 集群的路由信息，从而进行消息的投递和消费。

RocketMQ clients(Producer/Consumer) will query the queue routing info from NameServer, but how do clients find NameServer address?
There are four methods to feed NameServer address list to clients:

• Programmatic Way, like producer.setNamesrvAddr("ip:port").
• Java Options, use rocketmq.namesrv.addr.
• Environment Variable, use NAMESRV_ADDR.

• HTTP Endpoint.

  Broker

  Broker 主要负责消息的存储、投递和查询以及服务高可用保证，为了实现这些功能，Broker 包含了以下几个重要子模块：

• Remoting Module：整个 Broker 的实体，负责处理来自 clients 的请求。

• Client Manager：负责管理客户端 (Producer/Consumer) ，维护 Consumer 的 Topic 订阅信息。
• Store Service：提供方便简单的 API 接口处理消息存储到物理硬盘和查询功能。
• HA Service：高可用服务，提供 Master Broker 和 Slave Broker 之间的数据同步功能。
• Index Service：根据特定的 Message key 对投递到 Broker 的消息进行索引服务，以提供消息的快速查询。

Broker 中的消息被消费后不会立即删除。每条消息都会持久化到 CommitLog 中，每个 Consumer 连接到 Broker 后会维持消费进度信息，当有消息消费后只是当前 Consumer 的消费进度（CommitLog 的 offset）更新了。

消息堆积与消息删除

4.6 版本默认 48 小时后会删除不再使用的 CommitLog 文件

• 检查这个文件最后访问时间
• 判断是否大于过期时间

• 指定时间删除，默认凌晨 4 点

  消费模式

  消费模型由 Consumer 决定，消费维度为 Topic。

• 集群消费

1. 一条消息只会被同 Group 中的一个 Consumer 消费
   2. 多个 Group 同时消费一个 Topic 时，每个 Group 都会有一个 Consumer 消费到数据

• 广播消费

消息将对一个 Consumer Group 下的各个 Consumer 实例都消费一遍（即使这些 Consumer 属于同一个 Consumer Group）。

消息拉取

RocketMQ 没有真正意义的 push，都是 pull，虽然有 push 类，但实际底层实现采用的是长轮询机制，即拉取方式。
broker 端属性 longPollingEnable 标记是否开启长轮询，默认开启。
源码如下：

// {@link org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage()}
// 看到没，这是一只披着羊皮的狼，名字叫 PushConsumerImpl，实际干的却是 pull 的活。
// 拉取消息，结果放到 pullCallback 里
this.pullAPIWrapper.pullKernelImpl(pullCallback);

为什么要主动拉取消息而不使用事件监听方式?
事件驱动方式是建立好长连接，由事件（发送数据）的方式来实时推送。
如果 broker 主动推送消息的话有可能 push 速度快，消费速度慢的情况，那么就会造成消息在 consumer 端堆积过多，同时又不能被其他 consumer 消费的情况。
而 pull 的方式可以根据当前自身情况来 pull，不会造成过多的压力而造成瓶颈。所以采取了 pull 的方式。

broker 如何处理拉取请求？

Consumer 首次请求 Broker

• Broker 中是否有符合条件的消息
• 有 ->
  • 响应 Consumer
  • 等待下次 Consumer 的请求

• 没有

  • DefaultMessageStore#ReputMessageService#run 方法
  • PullRequestHoldService 来 Hold 连接，每隔 5s 执行一次检查 pullRequestTable 有没有消息，有的话立即推送
  • 每隔 1ms 检查 commitLog 中是否有新消息，有的话写入到 pullRequestTable
  • 当有新消息的时候返回请求
  • 挂起 consumer 的请求，即不断开连接，也不返回数据
  • 使用 consumer 的 offset

    负载均衡

    通过 Topic 在多 Broker 中分布式存储实现。

    producer 端

    发送端指定 message queue 发送消息到相应的 broker，来达到写入时的负载均衡

• 提升写入吞吐量，当多个 producer 同时向一个 broker 写入数据的时候，性能会下降

• 消息分布在多 broker 中，为负载消费做准备

默认策略是随机选择：

• producer 维护一个 index
• 每次取节点会自增
• index 向所有 broker 个数取余
• 自带容错策略

其他实现：

• SelectMessageQueueByHash
  • hash 的是传入的 args
• SelectMessageQueueByRandom
• SelectMessageQueueByMachineRoom 没有实现

也可以自定义实现 MessageQueueSelector 接口中的 select 方法

MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);

consumer 端

采用的是平均分配算法来进行负载均衡。
其他负载均衡算法

• 【默认】平均分配策略 (AllocateMessageQueueAveragely)
• 环形分配策略 (AllocateMessageQueueAveragelyByCircle)
• 手动配置分配策略 (AllocateMessageQueueByConfig)
• 机房分配策略 (AllocateMessageQueueByMachineRoom)
• 一致性哈希分配策略 (AllocateMessageQueueConsistentHash)
• 靠近机房策略 (AllocateMachineRoomNearby)

当消费负载均衡 consumer 和 queue 不对等的时候：
Consumer 和 queue 会优先平均分配，如果 Consumer 少于 queue 的个数，则会存在部分 Consumer 消费多个 queue 的情况，如果 Consumer 等于 queue 的个数，那就是一个 Consumer 消费一个 queue，如果 Consumer 个数大于 queue 的个数，那么会有部分 Consumer 空余出来，白白的浪费了。

重复消费

影响消息正常发送和消费的重要原因是网络的不确定性。

引起重复消费的原因

• ACK

正常情况下在 consumer 真正消费完消息后应该发送 ack，通知 broker 该消息已正常消费，从 queue 中剔除。当 ack 因为网络原因无法发送到 broker，broker 会认为此条消息没有被消费，此后会开启消息重投机制把消息再次投递到 consumer

• 消费模式

在 CLUSTERING 模式下，消息在 broker 中会保证相同 group 的 consumer 消费一次，但是针对不同 group 的 consumer 会推送多次

解决方案

• 数据库表（幂等）

处理消息前，使用消息主键在表中带有约束的字段中 insert

• Map

单机时可以使用 map ConcurrentHashMap -> putIfAbsent
guava cache

• Redis

分布式锁搞起来。

消息的顺序消费

首先多个 queue 只能保证单个 queue 里的顺序，queue 是典型的 FIFO，天然顺序。多个 queue 同时消费是无法绝对保证消息的有序性的。所以总结如下：
同一 topic，同一个 queue，发消息的时候一个线程去发送消息，消费的时候一个线程去消费一个 queue 里的消息。

保证消息发到同一个 queue

Rocket MQ 给我们提供了 MessageQueueSelector 接口，可以自己重写里面的接口，实现自己的算法，举个最简单的例子：判断i % 2 == 0，那就都放到 queue1 里，否则放到 queue2 里。

for (int i = 0; i < 5; i++) {
    Message message = new Message("orderTopic", ("hello!" + i).getBytes());
    producer.send(
        // 要发的那条消息
        message,
        // queue 选择器 ，向 topic中的哪个queue去写消息
        new MessageQueueSelector() {
            // 手动 选择一个queue
            @Override
            public MessageQueue select(
                // 当前topic 里面包含的所有queue
                List<MessageQueue> mqs,
                // 具体要发的那条消息
                Message msg,
                // 对应到 send（） 里的 args，也就是2000前面的那个0
                Object arg) {
                // 向固定的一个queue里写消息，比如这里就是向第一个queue里写消息
                if (Integer.parseInt(arg.toString()) % 2 == 0) {
                    return mqs.get(0);
                } else {
                    return mqs.get(1);
                }
            }
        },
        // 自定义参数：0
        i, 
        // 2000代表2000毫秒超时时间
        2000);
}

保证消息不丢失

首先在如下三个部分都可能会出现丢失消息的情况：

• Producer 端
• Broker 端

• Consumer 端

  Producer 端如何保证消息不丢失

• 采取 send () 同步发消息，发送结果是同步感知的。

• 发送失败后可以重试，设置重试次数（默认 3 次）。

producer.setRetryTimesWhenSendFailed(10);

• 集群部署，比如发送失败了的原因可能是当前 Broker 宕机了，重试的时候会发送到其他 Broker 上。

  Broker 端如何保证消息不丢失

• 修改刷盘策略为同步刷盘。默认情况下是异步刷盘的。

flushDiskType = SYNC_FLUSH

• 集群部署，主从模式，高可用（同步复制将 Master Broker 中的消息同步到 Slave Broker）。

  Consumer 端如何保证消息不丢失

• 完全消费正常后再进行手动 ack 确认。

• 消息失败重试。

  消息堆积如何处理

  下游消费系统如果宕机了，导致几百万条消息在消息中间件里积压，此时怎么处理？ 你们线上是否遇到过消息积压的生产故障？如果没遇到过，你考虑一下如何应对？

首先要找到是什么原因导致的消息堆积，是 Producer 太多了，Consumer 太少了导致的还是说其他情况，总之先定位问题。
然后看下消息消费速度是否正常，正常的话，可以通过上线更多 consumer 临时解决消息堆积问题。

如果 Consumer 和 Queue 不对等，上线了多台也在短时间内无法消费完堆积的消息怎么办？

• 准备一个临时的 topic
• queue 的数量是堆积的几倍
• queue 分布到多 Broker 中
• 上线一台 Consumer 做消息的搬运工，把原来 Topic 中的消息挪到新的 Topic 里，不做业务逻辑处理，只是挪过去
• 上线 N 台 Consumer 同时消费临时 Topic 中的数据
• 改 bug

• 恢复原来的 Consumer，继续消费之前的 Topic

  堆积时间过长消息会不会超时

  RocketMQ 中的消息只会在 commitLog 被删除的时候才会消失，不会超时。也就是说未被消费的消息不会存在超时删除这情况。

  堆积的消息会不会进死信队列

  Dead-letter queues
  不会，消息在消费失败后会进入重试队列（% RETRY%+ConsumerGroup），18 次后（默认 18 次，网上所有文章都说是 16 次，无一例外。但是我没搞懂为啥是 16 次，这不是 18 个时间吗 ？）才会进入死信队列（% DLQ%+ConsumerGroup）。
  源码如下：

  public class MessageStoreConfig {
    // 每隔如下时间会进行重试，到最后一次时间重试失败的话就进入死信队列了。
     private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
  }

  分布式事务机制

  分布式系统中的事务可以使用 TCC（Try、Confirm、Cancel）、2pc 来解决分布式系统中的消息原子性。
  RocketMQ 4.3 + 提供分布事务功能，通过 RocketMQ 事务消息能达到分布式事务的最终一致。
  RocketMQ 实现方式：

• Half Message：预处理消息，当 broker 收到此类消息后，会存储到 RMQ_SYS_TRANS_HALF_TOPIC 的消息消费队列中。

• 检查事务状态：Broker 会开启一个定时任务，消费 RMQ_SYS_TRANS_HALF_TOPIC 队列中的消息，每次执行任务会向消息发送者确认事务执行状态（提交、回滚、未知），如果是未知，Broker 会定时回调，重新检查。
• 超时：如果超过回查次数，默认回滚消息。

也就是消息并未真正进入 Topic 的 queue，而是用了临时 queue 来放所谓的 half message，等提交事务后才会真正的将 half message 转移到 topic 下的 queue。

源码概览

里面比较典型的设计模式有单例、工厂、策略、门面模式。
单例工厂无处不在；
策略模式比如发消息和消费消息的时候 queue 的负载均衡就是 N 个策略算法类，有随机、hash 等，这也是能够快速扩容，天然支持集群的必要原因之一；
持久化做的也比较完善，采取的 CommitLog 来落盘，同步异步两种方式。

性能优化

开发

• 同一 group 下，多机部署，并行消费
• 单个 Consumer 提高消费线程个数

• 批量消费

  • 消息批量拉取
  • 业务逻辑批量处理

    运维

• 网卡调优

• jvm 调优
• 多线程与 cpu 调优

• Page Cache

  保证数据高容错性

• 在不开启容错的情况下，轮询队列进行发送，如果失败了，重试的时候过滤失败的 Broker

• 如果开启了容错策略，会通过 RocketMQ 的预测机制来预测一个 Broker 是否可用
• 如果上次失败的 Broker 可用那么还是会选择该 Broker 的队列
• 如果上述情况失败，则随机选择一个进行发送
• 在发送消息的时候会记录一下调用的时间与是否报错，根据该时间去预测 broker 的可用时间。其实就是 send 消息的时候 queue 的选择。源码在如下：

  org.apache.rocketmq.client.latency.MQFaultStrategy#selectOneMessageQueue()

  Broker 宕机处理

  Broker 主从架构以及多副本策略。Master 收到消息后会同步给 Slave，这样一条消息就不止一份了，Master 宕机了还有 Slave 中的消息可用，保证了 MQ 的可靠性和高可用性。
  而且 Rocket MQ4.5.0 开始就支持了 Dlegder 模式，基于 raft 的，做到了真正意义的 HA（High Availability）。

  Broker 信息注册到 NameServer

  Broker 会向所有的 NameServer 上注册自己的信息，而不是某一个，是每一个，全部！