浅析RocketMQ

SpringBoot引入RocketMQ

快速构建单机RocketMQ

https://www.haveyb.com/article/3079 参考这篇文章，快速构建单机RocketMQ

项目引入jar包和配置

        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.3.1</version>
        </dependency>

rocketmq:
  consumer:
    group: oneCoupon_merchant_admin_consumer_group
    # 一次拉取消息最大值，注意是拉取消息的最大值而非消费最大值
    pull-batch-size: 10
  name-server: xxx:9876
  producer:
    # 发送同一类消息的设置为同一个group，保证唯一
    group: oneCoupon_merchant_admin_producer_group
    # 发送消息超时时间，默认3000
    sendMessageTimeout: 10000
    # 发送消息失败重试次数，默认2
    retryTimesWhenSendFailed: 2
    # 异步消息重试此处，默认2
    retryTimesWhenSendAsyncFailed: 2
    # 消息最大长度，默认1024 * 1024 * 4(默认4M)
    maxMessageSize: 4096
    # 压缩消息阈值，默认4k(1024 * 4)
    compressMessageBodyThreshold: 4096
    # 是否在内部发送失败时重试另一个broker，默认false
    retryNextServer: false

在console中添加消费者

示例

生产者

@Component
@RequiredArgsConstructor
@Slf4j
public class ShortLinkStatsSaveProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    public void send(String topic ,Map<String, String> producerMap) {
        log.info("send message to rocketMQ, topic: {}, producerMap: {}", topic, producerMap);
        rocketMQTemplate.syncSend(topic, producerMap);
    }
}

消费者

@Component
@RocketMQMessageListener(consumerGroup = "saaslink_consumer_group", topic = RedisKeyConstant.SHORT_LINK_STATS_STREAM_TOPIC_KEY)
@Slf4j
public class ShortLinkStatsSaveConsumer implements RocketMQListener<MessageExt> {
@Override
    public void onMessage(MessageExt msgExt) {
        String msgId = msgExt.getMsgId();
        // 使用redis实现幂等
        if (messageQueueIdempotentHandler.isMessageBeingConsumed(msgId.toString())) {
            // 判断当前的这个消息流程是否执行完成
            if (messageQueueIdempotentHandler.isAccomplish(msgId.toString())) {
                return;
            }
            throw new ServiceException("消息未完成流程，需要消息队列重试");
        }
        try {
            byte[] msgExtBody = msgExt.getBody();
            // 转为map
            Map<String, String> producerMap = JSON.parseObject(msgExtBody, Map.class);
            ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);
            // 实际新增的逻辑
            actualSaveShortLinkStats(producerMap.get("fullShortUrl"), producerMap.get("gid"), statsRecord);
        } catch (Throwable ex) {
            // 某某某情况宕机了
            messageQueueIdempotentHandler.delMessageProcessed(msgId.toString());
            log.error("记录短链接监控消费异常", ex);
            throw ex;
        }
        messageQueueIdempotentHandler.setAccomplish(msgId.toString());
    }

}

RocketMQ剖析

RocketMQ架构组成

Producer：消息发布的角色，支持分布式集群方式部署。Producer通过MQ的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程支持快速失败并且低延迟。
Consumer：消息消费的角色，支持分布式集群方式部署。支持以push推，pull拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制，可以满足大多数用户的需求。
NameServer：NameServer是一个非常简单的Topic路由注册中心，其角色类似Dubbo中的zookeeper，支持Broker的动态注册与发现。主要包括两个功能：Broker管理，NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活；路由信息管理，每个NameServer将保存关于Broker集群的整个路由信息和用于客户端查询的队列信息。然后Producer和Conumser通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费。NameServer通常也是集群的方式部署，各实例间相互不进行信息通讯。Broker是向每一台NameServer注册自己的路由信息，所以每一个NameServer实例上面都保存一份完整的路由信息。当某个NameServer因某种原因下线了，Broker仍然可以向其它NameServer同步其路由信息，Producer,Consumer仍然可以动态感知Broker的路由的信息。
BrokerServer：消息中转角色，负责存储消息、转发消息。代理服务器在RocketMQ系统中负责接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备。代理服务器也存储消息相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。

集群工作流程

启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。
Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。
收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。
Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。
Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。

消息存贮

Producer 和 CommitLog：
- Producer 通过Topic、QueueId和Message将消息发送到RocketMQ。消息在到达服务器后，会被写入到CommitLog中，这是RocketMQ存储消息的核心位置。图中，CommitLog部分展示了消息存储的状态。
- 红色表示已写入的消息，空心框表示等待写入的消息。消息按照先后顺序被存储在CommitLog中，并附有相应的commitLogOffset（即在CommitLog中的偏移量）、msgSize（消息大小）和tagsCode（用于消息筛选的标签编码）。
异步构建消费逻辑队列（doDispatch）：
- RocketMQ会异步地将写入到CommitLog中的消息构建成消费逻辑队列。这一步通过doDispatch来完成。doDispatch会将消息的commitLogOffset、msgSize、tagsCode等信息添加到相应的ConsumerQueue中，以支持消费者对消息的消费。
- 这个过程实现了消息的逻辑队列和物理存储（CommitLog）之间的关联。
ConsumerQueue 和消费过程：
- 图中展示了三条ConsumerQueue（ConsumerQueue0、ConsumerQueue1、ConsumerQueue2），每条ConsumerQueue对应着不同的Topic队列。
- 这些ConsumerQueue持有指向CommitLog中消息的偏移量（commitLogOffset）。通过这些偏移量，消费者可以根据ConsumerQueue从CommitLog中找到并消费相应的消息。
- 在每条ConsumerQueue中，minOffset表示消费队列中最小的偏移量，而maxOffset表示消费队列中最大的偏移量。consumerOffset则是当前消费的进度。
- 消息状态标记为三种：未消费（红色块）、已消费（实心框），以及等待分发（空心框）。
消费者（Consumer）与消息消费：
- 消费者通过读取对应的ConsumerQueue中的consumerOffset来消费消息。消费的过程是通过更新consumerOffset来记录消费的进度。
- 当消息被消费时，consumerOffset会更新到下一个未消费的偏移量，同时消费的消息状态在ConsumerQueue中标记为已消费。

消息刷盘

在RocketMQ中，消息刷盘的过程有两种方式：同步刷盘（synchronous flush disk）和异步刷盘（asynchronous flush disk）。
可以通过在broker.conf文件中进行配置，默认是异步刷盘：

flushDiskType = SYNC_FLUSH
# 或者
flushDiskType = ASYNC_FLUSH

同步刷盘 (Synchronous Flush Disk)
- 消息发送：Producer 发送消息给 Broker。
- 存储流程：
  - Broker 接收到消息后，将其先写入到 Java 堆内存中。
  - 然后，消息被转移到虚拟内存。
  - 最终，虚拟内存中的消息被刷入到磁盘。
- 确认 (ACK) 机制：
  - 在同步刷盘模式下，只有当消息成功写入到磁盘后，Broker 才会返回 ACK 给 Producer，表示消息已持久化成功。
  - 这种方式的优点是数据安全性较高，确保数据在返回成功确认之前已经持久化到磁盘中。
  - 缺点是性能较低，因为需要等待刷盘操作完成。
异步刷盘 (Asynchronous Flush Disk)
- 消息发送：Producer 发送消息给 Broker。
- 存储流程：
  - Broker 接收到消息后，同样会先将其写入到 Java 堆内存。
  - 然后，消息被存储到虚拟内存中。
  - 在此模式下，消息写入到虚拟内存后，Broker 就立即返回 ACK 给 Producer。
  - 消息的实际刷盘（写入磁盘）是通过异步多线程方式完成的。
- 确认 (ACK) 机制：
  - 异步刷盘模式在消息写入虚拟内存后，不等待消息写入磁盘的结果，就会直接返回 ACK。
  - 这种方式的优点是性能更高，Producer 可以更快地发送消息，不必等待刷盘完成。
  - 缺点是数据安全性较低，因为在返回 ACK 后，消息还未立即被持久化到磁盘中，存在数据丢失的风险。

RocketMQ功能实现分析

RocketMQ延时消息

rockeketMQ支持18个级别的延时等级，默认值为：“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”

消息主题替换 (SCHEDULE_TOPIC_XXX)
- 当 Producer 发送一条延时消息时，RocketMQ 并不会直接将消息投递到用户指定的实际 Topic。相反，RocketMQ 会将这些消息的 Topic 替换为内部预定义的 SCHEDULE_TOPIC_XXX。
- 这样做的目的是为了将这些延时消息区分开，并通过内部的调度机制来管理这些延时消息。SCHEDULE_TOPIC_XXX 是一个特殊的主题，专用于存放延时消息。
根据延时等级放入对应的队列
- RocketMQ 的延时消息有 18 个延时等级，每个等级对应不同的延时时长（例如 1s、5s、10s 等）。
- 为了管理这些不同延时等级的消息，SCHEDULE_TOPIC_XXX 主题下会有 18 个队列，每个队列对应一个延时等级。例如：
  - SCHEDULE_TOPIC_XXX 下的 Queue0 可能对应 1 秒的延时等级。
  - Queue1 可能对应 5 秒的延时等级，以此类推。
- 当消息到达 RocketMQ 后，Broker 根据消息的 delayLevel 属性，将其放入对应延时等级的队列中。
每个队列创建定时任务进行调度
- RocketMQ 在内部为每个延时队列创建了 定时任务。这些定时任务会不断检查其对应的延时队列，判断其中是否有已经达到投递时间的消息。
- 定时任务的主要作用是定时扫描延时消息的存储情况，并判断是否需要将某些消息进行恢复处理。
- 例如，针对 Queue0（1 秒延时等级），定时任务会每秒扫描一次，查看是否有延时时间到达的消息。如果有，则执行下一步操作。
恢复到期消息重新投递到真实的 Topic
- 当定时任务发现某条消息的延时时间已经到达时，RocketMQ 会将该消息恢复到用户指定的真实 Topic。
- 恢复的过程包括以下几个步骤：
  - 从 SCHEDULE_TOPIC_XXX 的延时队列中移除消息。
  - 修改消息的 Topic 为原始的用户指定 Topic，并根据消息的目标主题、队列信息重新投递到消息队列中。
  - 将恢复后的消息写入 CommitLog，并且根据消息的 QueueId 和原始 Topic 放入到对应的 Consumer Queue 中。
- 这样一来，当延时消息到达指定时间后，它就会重新进入实际的消费逻辑中，消费者可以像处理普通消息一样消费这些到期的延时消息。

RocketMQ 消息重试机制

RocketMQ 在消费者（Consumer）消费消息失败后，提供了一种消息 重试机制，使消息能够重新被消费。重试机制是通过将失败的消息重新投递到特定的重试队列中实现的。下面是 RocketMQ 消息重试的详细分析。

重试机制的实现原理

重试队列 (%RETRY%+consumerGroup)
- 当消费者消费消息失败时，RocketMQ 会将这条消息放入一个 专门用于重试的队列。这个队列的名称为 "%RETRY%+consumerGroup"，其中 consumerGroup 是消费组的名称。
- 需要注意的是，重试队列是针对每个 消费组（Consumer Group） 设置的，而不是针对每个 Topic 设置的。
- 这样做的好处是，可以针对每个消费组实现独立的重试机制和消费状态的跟踪，保证各个消费组的重试消息不会互相干扰。
延时重试级别
- RocketMQ 为重试队列设置了多个 重试级别，并且每个重试级别都有对应的重新投递延时时间。
- 消息消费失败后，并不会立即重新投递，而是会被存入一个内部的延迟队列中（SCHEDULE_TOPIC_XXXX）。然后，RocketMQ 会通过后台定时任务，按照设定的延时级别，将消息重新保存到对应的重试队列（%RETRY%+consumerGroup）中。
- 每次重试的延时时间会逐渐增加，以便给系统一定的时间来恢复消费失败的原因。例如，第一次重试可能是在 10 秒后，第二次重试是在 30 秒后，以此类推。
- 这种机制可以有效避免频繁重试导致的系统负载过大。
重试消息的消费
- 当消息被存入 "%RETRY%+consumerGroup" 队列后，RocketMQ 会按照正常的消费流程再次将消息投递给消费者进行消费。
- 如果消费者成功消费，则重试过程结束；如果仍然消费失败，则会根据设定的重试次数进行下一次重试。

消费失败的策略

RocketMQ 针对消费失败的情况设计了一系列的重试策略，以确保在失败的情况下可以最大程度地保证消息的被消费。具体策略如下：

重试次数：
- RocketMQ 默认会对每条消费失败的消息 重试最多 16 次。
- 如果消息在这 16 次尝试后依然未能被成功消费，RocketMQ 就认为消费者无法成功消费这条消息，需要进行特殊处理。
重试时间间隔递增：
- 消息的重试时间间隔是递增的，通过内部的延迟队列（SCHEDULE_TOPIC_XXXX）完成。
- 这种递增机制是为了给系统足够的时间来解决导致消费失败的问题，避免短时间内反复重试带来的不必要的资源消耗。
- 例如：第一次重试可能是 10 秒后，第二次重试是 30 秒后，第三次重试是 1 分钟后，等等。重试的时间间隔逐步增加，最大化保证消息的成功消费。
死信队列（DLQ, Dead Letter Queue）：
- 如果消息在所有重试次数（例如 16 次）之后依然无法被消费，则 RocketMQ 会将这条消息放入一个特殊的队列，称为 死信队列（Dead Letter Queue）。
- 死信队列的作用是保存那些经过多次尝试后仍无法被正常消费的消息，以便后续分析和处理。
- 死信队列的 Topic 名称为 "%DLQ%+consumerGroup"，每个消费组都有其对应的死信队列。

事务消息

public class TransactionProducer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        TransactionListener transactionListener = new TransactionListenerImpl();
        TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });

        producer.setExecutorService(executorService);
        //事务监听器
        producer.setTransactionListener(transactionListener);
        producer.start();

        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            try {
                Message msg =
                    new Message("TopicTest1234", tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                System.out.printf("%s%n", sendResult);

                Thread.sleep(10);
            } catch (MQClientException | UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }

        for (int i = 0; i < 100000; i++) {
            Thread.sleep(1000);
        }
        producer.shutdown();
    }
}

public interface TransactionListener {
    /**
     * When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction.
     *
     * @param msg Half(prepare) message
     * @param arg Custom business parameter
     * @return Transaction state
     */
    LocalTransactionState executeLocalTransaction(final Message msg, final Object arg);

    /**
     * When no response to prepare(half) message. broker will send check message to check the transaction status, and this
     * method will be invoked to get local transaction status.
     *
     * @param msg Check message
     * @return Transaction state
     */
    LocalTransactionState checkLocalTransaction(final MessageExt msg);
}

RocketMQ 提供了一种 事务消息机制，用于实现分布式事务，确保分布式系统中的不同组件能够一致性地执行一组操作。事务消息允许应用程序在本地事务执行成功后，发送一条确认消息，使得消息系统能够确保其他系统最终一致性地处理这条消息。下面是对 RocketMQ 事务消息机制的详细分析。

事务消息的实现原理

RocketMQ 的事务消息由三个主要步骤组成：预处理消息、本地事务执行、事务状态回查。

预处理消息（Prepare Message）
- 当 Producer 需要发送事务消息时，它首先会发送一条 预处理消息 到 RocketMQ。
- 这条消息会存储在 Broker 中，但是它的状态是 “未确认”，表示事务还在处理中，尚未被正式提交。
- Broker 接收到预处理消息后，会返回一个确认给 Producer，表示预处理消息已成功存储。
执行本地事务（Local Transaction Execution）
- Producer 在收到预处理消息成功的确认后，会开始执行对应的 本地事务。本地事务通常是应用程序的业务逻辑操作，比如数据库写操作等。
- 本地事务的执行结果有两种情况：
  - 成功：说明本地事务执行成功，接下来需要确认消息正式提交。
  - 失败：说明本地事务执行失败，需要回滚该事务消息。
提交或回滚事务消息
- 在本地事务执行完毕后，Producer 会根据执行结果向 Broker 发送提交或回滚消息：
  - 提交事务消息：表示本地事务执行成功，Broker 将预处理消息转换为正式消息，消息状态变为可被消费。
  - 回滚事务消息：表示本地事务执行失败，Broker 会删除该预处理消息，消息将不会被消费者接收到。
- 提交消息或回滚消息是决定消息能否被消费的重要步骤，确保消息的最终一致性。
事务状态回查（Transaction Status Check）
- 如果由于网络或系统故障，Producer 无法发送提交或回滚请求，RocketMQ 的 Broker 会通过 事务状态回查 来确保消息的最终状态。
- Broker 会向 Producer 发送 回查请求，询问该预处理消息对应的本地事务状态。Producer 根据本地事务的执行情况返回状态，可能是：
  - COMMIT：确认提交，Broker 会将消息转换为正式消息。
  - ROLLBACK：回滚消息，Broker 会删除该预处理消息。
  - UNKNOWN：Producer 无法确定事务状态，Broker 会再次进行回查，直到确认结果。
- 回查机制确保了系统的可靠性，即使出现故障，也能够通过回查机制保证消息的最终状态。

简单例子-电商订单系统

当用户下单购买商品时，需要同时执行以下三个操作：

生成订单记录：在订单数据库中插入一个新订单记录。
扣减库存：在库存系统中扣减相应商品的库存数量。
通知物流系统：在物流系统中生成一个新的配送请求

这三个操作涉及订单服务、库存服务、和物流服务，需要保证他们之间的数据一致性。如果订单生成成功，但库存扣减失败，或者通知物流系统失败，这都会导致数据不一致的问题。
为了确保这三个操作在分布式环境中一致成功，可以使用 RocketMQ 的事务消息机制，步骤如下：

发送预处理消息：
- 在用户下单时，订单服务首先发送一条预处理消息（Prepare Message）到 RocketMQ，表示要创建订单的事务开始。
- 这条消息会被存储到 RocketMQ 的 Broker 中，但状态是“未确认”，等待订单服务完成本地事务。
  执行本地事务：
执行本地事务
- 订单服务在发送预处理消息成功后，开始执行本地事务：
- 在订单数据库中插入订单记录。
- 如果订单插入成功，则继续调用库存服务，扣减库存数量。
- 库存扣减成功后，通知物流系统，生成配送请求。

如果以上三个操作都成功，说明本地事务成功执行。

提交或回滚事务消息：
- 如果本地事务执行成功，订单服务向 RocketMQ 发送提交事务消息，确认这条预处理消息可以被正式消费。RocketMQ 会将这条消息投递给订阅了这个 Topic 的消费组，例如物流服务。
- 如果在本地事务执行中出现任何问题，例如库存扣减失败，则订单服务会向 RocketMQ 发送回滚事务消息，RocketMQ 会删除该预处理消息，表示此次下单事务失败，消息不再继续投递。
事务状态回查：
- 如果由于网络异常或其他原因，RocketMQ 在等待订单服务的确认（提交或回滚）时超时未收到确认，RocketMQ 会向订单服务发起事务状态回查。
- RocketMQ 会询问订单服务，之前的预处理消息对应的本地事务到底是成功了还是失败了。
- 订单服务返回相应的状态，如果确认本地事务成功，则提交事务消息；如果失败，则回滚消息。

用户下单 --> 订单服务 --> 发送预处理消息到 RocketMQ --> RocketMQ 存储消息 (Prepare)
  --> 订单服务执行本地事务 (生成订单、扣减库存、通知物流)
    --> 如果成功 --> 订单服务提交事务消息 (Commit) --> RocketMQ 消息正式投递到物流服务
    --> 如果失败 --> 订单服务回滚事务消息 (Rollback) --> RocketMQ 删除预处理消息
    --> 如果未确认 --> RocketMQ 发起事务状态回查 --> 订单服务返回最终状态

标签：事务,队列,Broker,重试,消息,浅析,RocketMQ
From： https://www.cnblogs.com/b1uesk9/p/18452939