MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

RocketMQ 消息生命周期管理

2021-09-052.1k 阅读

RocketMQ 消息生命周期概述

RocketMQ 是一款分布式消息队列，广泛应用于高并发、高性能的场景中。消息在 RocketMQ 中的生命周期涵盖了从生产者发送消息，到消息在服务器端存储、流转，再到消费者接收并处理消息的全过程。理解这个生命周期对于正确使用 RocketMQ，保障系统的稳定性和可靠性至关重要。

消息发送阶段

在 RocketMQ 中，消息发送是消息生命周期的起点。生产者负责将业务数据封装成消息并发送到 RocketMQ 服务器。生产者通过与 NameServer 进行交互，获取 Topic 对应的 Broker 信息，然后选择一个 Broker 进行消息发送。

  1. 同步发送 同步发送是最常见的发送方式，生产者发送消息后，会等待 Broker 返回发送结果。这种方式可靠性高，适合对消息发送成功与否要求严格的场景，比如订单创建消息。以下是 Java 代码示例：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class SyncProducer {
    public static void main(String[] args) throws Exception {
        // 创建生产者实例
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        // 设置 NameServer 地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        for (int i = 0; i < 10; i++) {
            // 创建消息实例
            Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " + i).getBytes("UTF-8") /* 消息体 */);
            // 同步发送消息
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }

        // 关闭生产者
        producer.shutdown();
    }
}

在上述代码中，我们创建了一个 DefaultMQProducer 实例，并设置了其所属的生产者组和 NameServer 地址。然后，我们循环创建并同步发送 10 条消息，每次发送后打印发送结果。

  1. 异步发送 异步发送时，生产者发送消息后不会等待 Broker 的响应，而是通过回调函数来处理发送结果。这种方式适用于对发送性能要求较高，且对消息发送结果处理不那么急切的场景，比如日志消息的发送。示例代码如下：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class AsyncProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("group2");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        for (int i = 0; i < 10; i++) {
            final int index = i;
            Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " + i).getBytes("UTF-8") /* 消息体 */);
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
                }

                @Override
                public void onException(Throwable e) {
                    System.out.printf("%-10d Exception %s %n", index, e);
                }
            });
        }

        Thread.sleep(5000);
        producer.shutdown();
    }
}

这里我们在发送消息时传入了一个 SendCallback 回调函数，在 onSuccess 方法中处理发送成功的逻辑，在 onException 方法中处理发送异常的逻辑。

  1. 单向发送 单向发送即生产者只负责发送消息，不关心发送结果。这种方式性能最高，但可靠性相对较低，适用于一些对消息可靠性要求不高的场景，如监控数据的上报。代码示例如下：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;

public class OnewayProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("group3");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        for (int i = 0; i < 10; i++) {
            Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " + i).getBytes("UTF-8") /* 消息体 */);
            producer.sendOneway(msg);
        }

        producer.shutdown();
    }
}

上述代码中，我们通过 producer.sendOneway(msg) 方法实现单向发送消息。

消息在服务器端的存储与流转

消息发送到 RocketMQ 服务器后，会经历存储和流转等过程。RocketMQ 的存储结构主要由 CommitLog、ConsumeQueue 和 IndexFile 组成。

CommitLog

CommitLog 是 RocketMQ 的核心存储结构，所有主题的消息都顺序存储在 CommitLog 中。这种顺序存储方式大大提高了写性能。每个 CommitLog 文件大小固定为 1G，当一个文件写满后，会创建新的文件继续写入。以下是 CommitLog 的写入流程：

  1. 生产者发送消息到 Broker。
  2. Broker 将消息追加写入 CommitLog 文件。
  3. 写入成功后，更新 CommitLog 的相关元数据，如文件偏移量等。

ConsumeQueue

ConsumeQueue 是消息消费的索引结构，每个 Topic 下的每个队列都有一个对应的 ConsumeQueue。ConsumeQueue 存储了消息在 CommitLog 中的物理偏移量、消息大小和消息 Tag 的哈希值等信息。消费者通过 ConsumeQueue 快速定位到消息在 CommitLog 中的位置，从而进行消息消费。其主要作用如下：

  1. 加速消息的定位和拉取，提高消费效率。
  2. 支持消费者的并发消费，每个 ConsumeQueue 可以被多个消费者并发消费。

IndexFile

IndexFile 是为了支持消息的按照 Key 或时间区间查询而设计的。IndexFile 以哈希表的形式存储消息的 Key 和消息在 CommitLog 中的物理偏移量。当生产者发送消息时，可以指定消息的 Key，这些 Key 会被写入 IndexFile。用户可以通过 RocketMQ 的 API 按照 Key 来查询消息，在排查问题时非常有用。例如，在电商系统中，可以将订单号作为消息 Key，方便查询特定订单相关的消息。

消息消费阶段

消息消费是消息生命周期的终点，消费者从 RocketMQ 服务器拉取消息并进行业务处理。RocketMQ 支持两种消费模式：集群消费和广播消费。

集群消费

在集群消费模式下，一个消费者组内的多个消费者实例共同消费一个 Topic 下的消息。每个消费者实例负责消费 Topic 下部分队列中的消息，以达到负载均衡的目的。例如，一个 Topic 有 4 个队列，一个消费者组内有 2 个消费者实例，那么每个消费者实例会负责消费 2 个队列中的消息。以下是 Java 代码示例：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class ClusterConsumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group4");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅 Topic 和 Tag
        consumer.subscribe("TopicTest", "TagA");

        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                          ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("Consumer Started.");
    }
}

在上述代码中，我们创建了一个 DefaultMQPushConsumer 实例，并设置了其所属的消费者组和 NameServer 地址。通过 consumer.subscribe 方法订阅了 Topic 和 Tag，然后注册了一个 MessageListenerConcurrently 消息监听器来处理接收到的消息。

广播消费

广播消费模式下，一个消费者组内的每个消费者实例都会消费 Topic 下的所有消息。这种模式适用于一些需要所有消费者都处理相同消息的场景，如配置更新消息。代码示例如下：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class BroadcastConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group5");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("TopicTest", "TagA");
        // 设置为广播消费模式
        consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                          ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.println("Consumer Started.");
    }
}

这里通过 consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING) 将消费模式设置为广播消费。

消息的重试与死信队列

在消息消费过程中，可能会出现消息消费失败的情况。RocketMQ 提供了消息重试和死信队列机制来处理这种情况。

消息重试

当消费者消费消息失败时，RocketMQ 会自动进行消息重试。默认情况下，一条消息最多重试 16 次。重试的间隔时间会逐渐变长，从 10 秒开始，最长达到 2 小时。例如，第一次重试间隔 10 秒，第二次 30 秒，第三次 1 分钟等。在集群消费模式下，消息会发送回原队列，等待重试。以下是自定义消息重试次数的代码示例：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class RetryConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group6");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("TopicTest", "TagA");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                          ConsumeConcurrentlyContext context) {
                MessageExt msg = msgs.get(0);
                // 模拟消费失败
                if (msg.getReconsumeTimes() < 3) {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 设置最大重试次数为 3 次
        consumer.setMaxReconsumeTimes(3);

        consumer.start();
        System.out.println("Consumer Started.");
    }
}

在上述代码中，我们通过 consumer.setMaxReconsumeTimes(3) 设置最大重试次数为 3 次，并在消息监听器中根据 msg.getReconsumeTimes() 判断是否需要重试。

死信队列

当消息重试次数达到上限后，RocketMQ 会将该消息发送到死信队列（DLQ，Dead - Letter Queue）。死信队列是一个特殊的 Topic，每个消费者组都有一个对应的死信队列。死信队列中的消息不会再被自动消费，需要人工介入处理。例如，可以分析死信队列中的消息，找出消费失败的原因并进行修复，然后手动将消息重新发送到原 Topic 进行消费。以下是获取死信队列名称的代码示例：

import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;

public class DeadLetterQueueExample {
    public static void main(String[] args) throws Exception {
        DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt();
        defaultMQAdminExt.setNamesrvAddr("localhost:9876");
        defaultMQAdminExt.start();

        TopicRouteData topicRouteData = defaultMQAdminExt.examineTopicRouteInfo("ConsumerGroupName");
        for (TopicConfig topicConfig : topicRouteData.getTopicConfigList()) {
            if (topicConfig.getTopicName().startsWith("%DLQ%")) {
                System.out.println("Dead Letter Queue Name: " + topicConfig.getTopicName());
            }
        }

        defaultMQAdminExt.shutdown();
    }
}

上述代码通过 DefaultMQAdminExt 获取消费者组对应的死信队列名称。

消息生命周期的监控与管理

为了保障消息在 RocketMQ 中的正常生命周期，需要对其进行监控和管理。RocketMQ 提供了多种方式来实现这一目标。

控制台监控

RocketMQ 提供了一个可视化的控制台，通过控制台可以查看 Topic、Broker、Consumer 等相关信息。可以监控消息的发送速率、消费速率、消息堆积情况等指标。例如，在控制台的 Topic 页面，可以看到每个 Topic 的消息入队速率、出队速率以及队列的堆积数量。如果发现某个 Topic 的消息堆积量持续增加，可能意味着消费者处理能力不足，需要及时调整。

自定义监控

除了控制台监控，还可以通过自定义代码来监控消息生命周期。可以利用 RocketMQ 的 API 获取消息队列的元数据信息，如队列数量、消息大小等。同时，可以结合一些开源的监控框架，如 Prometheus 和 Grafana，将 RocketMQ 的监控数据进行可视化展示。以下是一个简单的获取 Topic 队列数量的代码示例：

import org.apache.rocketmq.client.MQClientInstance;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.factory.MQClientManager;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;

import java.util.List;

public class TopicQueueMonitor {
    public static void main(String[] args) throws MQClientException {
        MQClientInstance mqClientInstance = MQClientManager.getInstance().getAndCreateMQClientInstance(null, null);
        TopicRouteData topicRouteData = mqClientInstance.examineTopicRouteInfo("TopicTest");
        List<TopicConfig> topicConfigList = topicRouteData.getTopicConfigList();
        for (TopicConfig topicConfig : topicConfigList) {
            if (topicConfig.getTopicName().equals("TopicTest")) {
                int queueCount = topicConfig.getWriteQueueNums();
                System.out.println("Topic " + topicConfig.getTopicName() + " has " + queueCount + " queues.");
            }
        }
    }
}

上述代码通过 MQClientInstance 获取 TopicTest 的队列数量。

消息清理与过期

RocketMQ 支持设置消息的过期时间（TTL，Time - To - Live）。默认情况下，消息在 CommitLog 中的保存时间为 72 小时。可以通过修改配置文件来调整这个时间。当消息过期后，会被自动清理。此外，也可以手动清理一些不需要的 Topic 及其相关的消息。在清理 Topic 时，需要注意确保该 Topic 不再被使用，以免影响业务。以下是通过控制台删除 Topic 的步骤：

  1. 登录 RocketMQ 控制台。
  2. 在 Topic 管理页面找到要删除的 Topic。
  3. 点击删除按钮，确认删除操作。

通过以上对 RocketMQ 消息生命周期管理的详细介绍，包括消息的发送、在服务器端的存储与流转、消费、重试与死信队列以及监控与管理等方面，希望能帮助开发者更好地理解和使用 RocketMQ，构建出更加稳定、可靠的分布式系统。在实际应用中，需要根据业务场景合理选择消息发送方式、消费模式，并做好消息的监控与管理，以充分发挥 RocketMQ 的优势。