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

RocketMQ 消息消费机制解读

2023-10-094.8k 阅读

RocketMQ 消息消费概述

在分布式系统中，消息队列扮演着至关重要的角色，而 RocketMQ 作为一款高性能、高可靠的消息队列，其消息消费机制更是核心部分。RocketMQ 的消息消费机制设计精妙，旨在满足不同场景下的消息处理需求，确保消息能够被准确、高效地消费。

RocketMQ 采用了基于拉模式（Pull-based）的消息消费方式，与推模式（Push-based）相对。拉模式下，消费者主动从 Broker 拉取消息，这种方式给予了消费者更多的控制权，比如可以自主控制拉取的频率、批量大小等参数，以更好地适应自身的处理能力。而推模式则是 Broker 主动将消息推送给消费者，消费者处于被动接收状态，在一些复杂场景下可能难以灵活处理消息。

消费模式

RocketMQ 支持两种消费模式：集群消费（Clustering）和广播消费（Broadcasting）。

  • 集群消费：多个消费者实例组成一个消费组（Consumer Group）共同消费一组消息。在这种模式下，RocketMQ 会将 Topic 中的消息尽可能均匀地分配给消费组内的各个消费者实例。例如，假设有一个 Topic 包含 10 条消息，消费组内有 2 个消费者实例，那么每个实例可能会分配到 5 条消息进行消费。这种模式适用于对消息处理性能要求较高，且可以并行处理消息的场景，比如订单处理系统，多个消费者可以同时处理不同订单的消息。
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroup");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe("TopicName", "*");
consumer.setMessageModel(MessageModel.CLUSTERING);
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            System.out.println("Consume message: " + new String(msg.getBody()));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});
consumer.start();

上述代码展示了如何使用 Java 客户端设置为集群消费模式，通过 consumer.setMessageModel(MessageModel.CLUSTERING) 来指定。DefaultMQPushConsumer 是 RocketMQ 提供的一个用于推模式消费的类，registerMessageListener 方法注册了一个消息监听器，当有消息到达时，会调用 consumeMessage 方法进行处理。

  • 广播消费：消费组内的每个消费者实例都会接收 Topic 中的全部消息。这意味着如果有 10 条消息，消费组内有 2 个消费者实例，每个实例都会接收到这 10 条消息。广播消费适用于一些需要每个消费者都知晓所有消息的场景，比如系统配置更新通知，每个消费者都需要获取最新的配置信息。
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroup");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe("TopicName", "*");
consumer.setMessageModel(MessageModel.BROADCASTING);
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            System.out.println("Consume message: " + new String(msg.getBody()));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});
consumer.start();

这段代码与集群消费类似，只是通过 consumer.setMessageModel(MessageModel.BROADCASTING) 设置为广播消费模式。

消息队列分配策略

在集群消费模式下，RocketMQ 需要将 Topic 的消息队列合理地分配给消费组内的各个消费者实例，这就涉及到消息队列分配策略。RocketMQ 提供了多种分配策略，常见的有以下几种：

  • 平均分配策略（AllocateMessageQueueAveragely）：这是默认的分配策略。它将消息队列按照消费者实例的数量进行平均分配。例如，有 10 个消息队列和 3 个消费者实例，10 除以 3 得 3 余 1，那么前两个消费者实例会分配到 4 个队列，第三个消费者实例会分配到 2 个队列。这种策略简单直观，能在一定程度上保证负载均衡。
consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragely());

通过上述代码可以设置使用平均分配策略。

  • 平均粘性分配策略（AllocateMessageQueueAveragelySticky）：该策略在平均分配的基础上，增加了粘性特性。当消费者实例数量发生变化时，尽量保持原有队列的分配关系，减少队列的重新分配。比如，一个消费者实例停止工作，其他实例在重新分配队列时，会尽量沿用之前的分配方式，减少对整体消费进度的影响。
consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragelySticky());

设置使用平均粘性分配策略的代码如上。

  • 环形分配策略（AllocateMessageQueueByCircle）：按照环形的方式将消息队列依次分配给消费者实例。这种策略在一些特定场景下，如希望消息队列分配更加均匀且循环有序时会比较有用。
consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueByCircle());

通过此代码设置环形分配策略。

消费进度管理

RocketMQ 对于消费进度的管理非常关键，它确保了消费者在重启或故障恢复后能够继续从上次消费的位置开始处理消息。消费进度由消费组和消息队列共同标识，每个消费组针对每个消息队列都有一个独立的消费进度。 在 RocketMQ 中，消费进度存储在 Broker 端。当消费者成功消费一批消息后，会向 Broker 发送消费进度更新请求。Broker 会将消费进度持久化到本地存储（如文件系统），以保证在 Broker 重启后消费进度不丢失。 在代码层面，对于推模式消费，RocketMQ 客户端会自动处理消费进度的更新。例如在上述集群消费和广播消费的代码示例中，当 consumeMessage 方法返回 ConsumeConcurrentlyStatus.CONSUME_SUCCESS 时，客户端会自动向 Broker 更新消费进度。 对于拉模式消费，开发者需要手动管理消费进度。如下代码示例：

DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("ConsumerGroup");
consumer.start();
MessageQueue mq = new MessageQueue("TopicName", "BrokerName", 0);
long offset = consumer.fetchConsumeOffset(mq, true);
PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
switch (pullResult.getPullStatus()) {
    case FOUND:
        List<MessageExt> msgList = pullResult.getMsgFoundList();
        for (MessageExt msg : msgList) {
            System.out.println("Consume message: " + new String(msg.getBody()));
        }
        offset = pullResult.getNextBeginOffset();
        consumer.updateConsumeOffset(mq, offset);
        break;
    case NO_NEW_MSG:
        break;
    case NO_MATCHED_MSG:
        break;
    case OFFSET_ILLEGAL:
        break;
}
consumer.shutdown();

在这段代码中，fetchConsumeOffset 方法获取当前消息队列的消费进度，pullBlockIfNotFound 方法拉取消息，当成功拉取并消费消息后，通过 updateConsumeOffset 方法更新消费进度。

消费重试机制

在消息消费过程中，可能会因为各种原因导致消费失败，如网络异常、业务逻辑错误等。RocketMQ 提供了消费重试机制来处理这种情况。 当消费者消费消息失败时，RocketMQ 会根据重试策略进行重试。对于集群消费模式，默认情况下，消息会重试 16 次。每次重试的时间间隔会逐渐变长，从 10 秒开始，最长到 2 小时。例如，第一次重试间隔 10 秒，第二次 30 秒，第三次 1 分钟等。

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

通过上述代码可以设置自定义的最大重试次数。 在消费重试过程中，如果消息最终还是无法成功消费，RocketMQ 会将这些消息发送到死信队列（Dead Letter Queue）。死信队列是一个特殊的 Topic，用于存放这些无法成功消费的消息。开发者可以从死信队列中分析这些消息，排查消费失败的原因。

// 死信队列相关配置在 Broker 端进行，这里以示例说明如何查看死信队列中的消息
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("ConsumerGroup");
consumer.start();
MessageQueue mq = new MessageQueue("%DLQ%ConsumerGroup", "BrokerName", 0);
long offset = consumer.fetchConsumeOffset(mq, true);
PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, offset, 32);
switch (pullResult.getPullStatus()) {
    case FOUND:
        List<MessageExt> msgList = pullResult.getMsgFoundList();
        for (MessageExt msg : msgList) {
            System.out.println("Dead letter message: " + new String(msg.getBody()));
        }
        break;
    case NO_NEW_MSG:
        break;
    case NO_MATCHED_MSG:
        break;
    case OFFSET_ILLEGAL:
        break;
}
consumer.shutdown();

上述代码展示了如何从死信队列中拉取消息进行分析。

并发消费与顺序消费

  • 并发消费：RocketMQ 支持并发消费，消费者可以同时处理多个消息，以提高消费效率。在集群消费模式下，默认采用并发消费方式。如前面的集群消费代码示例，MessageListenerConcurrently 接口的 consumeMessage 方法每次会接收一批消息，消费者可以并行处理这批消息中的每一条。
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        ExecutorService executorService = Executors.newFixedThreadPool(msgs.size());
        for (MessageExt msg : msgs) {
            executorService.submit(() -> {
                System.out.println("Consume message: " + new String(msg.getBody()));
                // 处理业务逻辑
            });
        }
        executorService.shutdown();
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

上述代码通过创建线程池，进一步提高并发消费的效率，每个消息在独立的线程中处理。

  • 顺序消费：在某些场景下，消息的顺序性非常重要，比如电商订单的创建、支付、发货等流程，必须按照顺序处理。RocketMQ 支持顺序消费，它通过将同一业务逻辑相关的消息发送到同一个消息队列，消费者按照顺序从该队列中拉取消息进行处理，从而保证消息的顺序性。
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroup");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe("TopicName", "*");
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        for (MessageExt msg : msgs) {
            System.out.println("Consume message: " + new String(msg.getBody()));
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
});
consumer.start();

上述代码通过实现 MessageListenerOrderly 接口来实现顺序消费，consumeMessage 方法中的消息是按照顺序依次处理的。

消息过滤

在实际应用中，消费者可能只对 Topic 中的部分消息感兴趣，这就需要消息过滤功能。RocketMQ 支持两种类型的消息过滤：

  • Tag 过滤：在发送消息时，可以为消息设置 Tag，Tag 可以理解为消息的类别。消费者在订阅 Topic 时，可以指定需要消费的 Tag。例如：
// 发送消息设置 Tag
Message msg = new Message("TopicName", "Tag1", "Hello, RocketMQ!".getBytes(RemotingHelper.DEFAULT_CHARSET));
producer.send(msg);

// 消费者订阅指定 Tag 的消息
consumer.subscribe("TopicName", "Tag1");

在上述代码中，生产者发送消息时设置了 Tag1，消费者通过 subscribe 方法指定只消费 Tag1 的消息。

  • SQL92 表达式过滤：RocketMQ 还支持使用 SQL92 表达式进行更复杂的消息过滤。在发送消息时，可以为消息设置属性，然后在消费者订阅时使用 SQL 表达式根据属性进行过滤。例如：
// 发送消息设置属性
Message msg = new Message("TopicName", "Tag1", "Hello, RocketMQ!".getBytes(RemotingHelper.DEFAULT_CHARSET));
msg.putUserProperty("age", "20");
producer.send(msg);

// 消费者使用 SQL92 表达式过滤消息
consumer.subscribe("TopicName", MessageSelector.bySql("age > 18"));

在这段代码中，生产者为消息设置了 age 属性，消费者通过 MessageSelector.bySql 方法使用 SQL 表达式 age > 18 过滤出符合条件的消息。

消费端高可用

为了保证消息消费的高可用性，RocketMQ 从多个方面进行了设计。 在消费组层面，通过多个消费者实例组成消费组，当某个消费者实例出现故障时，其他实例可以接管其未处理完的消息队列，继续进行消费。例如，在一个有 3 个消费者实例的消费组中，如果其中一个实例宕机，RocketMQ 会重新分配消息队列给另外两个实例，确保消息不会丢失且能继续被消费。 在网络层面，RocketMQ 客户端与 Broker 之间采用了心跳机制。消费者定期向 Broker 发送心跳包，以保持连接的有效性。如果 Broker 在一定时间内没有收到某个消费者的心跳包，会认为该消费者出现故障，从而触发消息队列的重新分配。 同时，RocketMQ 还支持主从架构，Broker 有主节点和从节点。当主节点出现故障时，从节点可以切换为主节点，继续提供消息服务，包括消费服务，保证消息消费的连续性。

与其他组件的集成

RocketMQ 可以与多种其他组件进行集成，以满足复杂的业务需求。 与 Spring Boot 的集成非常方便，通过引入相关的依赖，可以轻松地在 Spring Boot 项目中使用 RocketMQ。例如：

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

在配置文件中配置 RocketMQ 的相关参数：

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: ProducerGroup

然后在代码中使用 @RocketMQMessageListener 注解来定义消费者：

@Component
@RocketMQMessageListener(topic = "TopicName", consumerGroup = "ConsumerGroup")
public class MyConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("Consume message: " + message);
    }
}

这样就完成了 RocketMQ 与 Spring Boot 的集成，借助 Spring Boot 的特性，可以更方便地管理和使用 RocketMQ。

RocketMQ 还可以与大数据处理框架如 Apache Flink 集成。通过 Flink 的 RocketMQ Connector，可以将 RocketMQ 作为数据源或数据 sink，实现实时数据处理。例如，将 RocketMQ 中的消息作为 Flink 流处理的输入：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Properties properties = new Properties();
properties.setProperty(ConsumerConfig.NAME_SERVER_ADDR, "127.0.0.1:9876");
properties.setProperty(ConsumerConfig.GROUP_ID, "ConsumerGroup");
DataStreamSource<String> stream = env.addSource(new FlinkRocketMQConsumer<>(
        "TopicName",
        new SimpleStringSchema(),
        properties
));
stream.print();
env.execute("RocketMQ to Flink");

上述代码展示了如何将 RocketMQ 中的消息引入到 Flink 进行处理，通过 FlinkRocketMQConsumer 实现了从 RocketMQ 到 Flink 的数据传输。

性能优化

在使用 RocketMQ 进行消息消费时，有几个方面可以进行性能优化。 在消费端，可以合理调整批量拉取的大小。如果批量拉取的消息数量过小，会增加拉取的频率，消耗网络资源；如果过大，可能会导致单个批次处理时间过长。可以根据业务场景和消费能力进行测试，找到合适的批量大小。例如，对于处理速度较快的简单业务，可以适当增大批量大小。

consumer.setPullBatchSize(32); // 设置批量拉取大小为 32

另外，合理使用线程池也能提高消费性能。对于并发消费，可以根据系统的资源情况创建合适大小的线程池来处理消息。例如，在 CPU 密集型业务中，线程池大小可以设置为 CPU 核心数的 1 - 2 倍；在 I/O 密集型业务中，可以适当增大线程池大小。

ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);

在 Broker 端，合理配置存储参数也能提升性能。比如，可以调整刷盘策略，异步刷盘可以提高写入性能，但可能会在 Broker 故障时丢失少量未刷盘的消息；同步刷盘则保证了数据的可靠性，但会降低写入性能。需要根据业务对数据可靠性和性能的要求来选择合适的刷盘策略。

<brokerName>broker-a</brokerName>
<brokerId>0</brokerId>
<storePathRootDir>/data/rocketmq/store</storePathRootDir>
<storePathCommitLog>/data/rocketmq/store/commitlog</storePathCommitLog>
<flushDiskType>ASYNC_FLUSH</flushDiskType>

上述 Broker 配置文件片段展示了如何设置刷盘策略为异步刷盘（ASYNC_FLUSH）。

故障排查与监控

在 RocketMQ 消息消费过程中，可能会遇到各种故障，需要有效的排查方法。常见的故障包括消费进度异常、消息丢失、消费失败等。 对于消费进度异常，可以通过 RocketMQ 提供的管理工具（如 RocketMQ Console）查看消费组的消费进度，对比各个消费者实例的进度是否正常。如果某个实例的进度长时间没有更新，可能是该实例出现了问题，如网络连接中断、代码逻辑阻塞等。 消息丢失问题可能由多种原因导致，如 Broker 故障、网络问题、消费未确认等。可以通过查看 Broker 的日志文件，分析是否有消息丢失的记录。同时，在消费端确保消息被成功消费后及时向 Broker 确认，避免消息被重复消费或丢失。 消费失败时，可以查看消费端的日志，分析具体的失败原因，如业务逻辑错误、依赖服务不可用等。结合重试机制，逐步排查问题。 为了及时发现和处理故障，监控 RocketMQ 的运行状态非常重要。可以监控 Broker 的性能指标，如消息堆积量、吞吐量、磁盘使用情况等。在消费端，可以监控消费进度、消费成功率、消费延迟等指标。通过这些监控数据，可以提前发现潜在的问题，及时采取措施进行优化和调整。例如，可以使用 Prometheus 和 Grafana 搭建监控系统，对 RocketMQ 的各项指标进行可视化展示和监控报警。

通过深入理解和掌握 RocketMQ 的消息消费机制，合理运用其各种特性，开发者能够构建出高性能、高可靠的分布式消息处理系统，满足不同业务场景的需求。无论是在电商、金融、物联网等各个领域，RocketMQ 的消息消费机制都能为系统的稳定性和高效性提供有力保障。