RocketMQ高级特性三-消费者分类

前言

概述

区别

PullConsumer

定义与概述

原理机制

使用场景

优缺点

Java 代码示例

SimpleConsumer

定义与概述

原理机制

使用场景

优缺点

Java 代码示例

PushConsumer

定义与概述

原理机制

使用场景

优缺点

Java 代码示例

总结

前言

RocketMQ中的消费者分类主要包括三种类型：PullConsumer、SimpleConsumer、和PushConsumer。每种消费者类型都有其特定的使用场景、原理机制以及优缺点。

注：生产环境中相同的 ConsumerGroup 下严禁混用 PullConsumer 和其他两种消费者，否则会导致消息消费异常。

概述

如上图所示， Apache RocketMQ 的消费者处理消息时主要经过以下阶段：消息获取--->消息处理--->消费状态提交。

针对以上几个阶段，Apache RocketMQ 提供了不同的消费者类型： PushConsumer 、SimpleConsumer 和 PullConsumer。这几种类型的消费者通过不同的实现方式和接口可满足您在不同业务场景下的消费需求。具体差异如下：

在实际使用场景中，PullConsumer 仅推荐在流处理框架中集成使用，大多数消息收发场景使用 PushConsumer 和 SimpleConsumer 就可以满足需求。

若您的业务场景发生变更，或您当前使用的消费者类型不适合当前业务，您可以选择在 PushConsumer 和SimpleConsumer 之间变更消费者类型。变更消费者类型不影响当前Apache RocketMQ 资源的使用和业务处理。

区别

对比项	PushConsumer	SimpleConsumer	PullConsumer
接口方式	使用监听器回调接口返回消费结果，消费者仅允许在监听器范围内处理消费逻辑。	业务方自行实现消息处理，并主动调用接口返回消费结果。	业务方自行按队列拉取消息，并可选择性地提交消费结果
消费并发度管理	由SDK管理消费并发度。	由业务方消费逻辑自行管理消费线程。	由业务方消费逻辑自行管理消费线程。
负载均衡粒度	5.0 SDK是消息粒度，更均衡，早期版本是队列维度	消息粒度，更均衡	队列粒度，吞吐攒批性能更好，但容易不均衡
接口灵活度	高度封装，不够灵活。	原子接口，可灵活自定义。	原子接口，可灵活自定义。
适用场景	适用于无自定义流程的业务消息开发场景。	适用于需要高度自定义业务流程的业务开发场景。	仅推荐在流处理框架场景下集成使用

PullConsumer

定义与概述

PullConsumer是一种传统的消息拉取模式，在这种模式下，消费者需要主动从Broker中拉取消息，而不是由Broker主动推送消息。这种模式提供了更大的灵活性，允许消费者根据自身的处理能力和业务逻辑，自主控制消息的拉取和消费节奏。

原理机制

在PushConsumer类型中，消息的实时处理能力是基于SDK内部的典型Reactor线程模型实现的。如下图所示，SDK内置了一个长轮询线程，先将消息异步拉取到SDK内置的缓存队列中，再分别提交到消费线程中，触发监听器执行本地消费逻辑。

消息拉取：消费者调用pull方法，从Broker的消息队列中主动拉取消息。拉取的过程通常是循环进行的，消费者可以在每次拉取时指定要拉取的消息数量和位置。
偏移量管理：消费者需要自行管理消息消费的偏移量（offset），确保每次拉取的消息都是未消费过的，防止消息重复消费。
流控管理：消费者可以根据自身的处理能力和当前系统的负载情况，动态调整消息的拉取频率和数量，避免处理能力不足导致的消息积压。

使用场景

批处理：适合需要批量处理消息的场景，例如数据分析、日志处理等。
自定义处理逻辑：在需要对消息进行复杂的过滤、分组或排序处理时，PullConsumer提供了更大的灵活性。

PushConsumer的消费监听器执行结果分为以下三种情况：

返回消费成功：以Java SDK为例，返回ConsumeResult.SUCCESS，表示该消息处理成功，服务端按照消费结果更新消费进度。
返回消费失败：以Java SDK为例，返回ConsumeResult.FAILURE，表示该消息处理失败，需要根据消费重试逻辑判断是否进行重试消费。
出现非预期失败：例如抛异常等行为，该结果按照消费失败处理，需要根据消费重试逻辑判断是否进行重试消费。

PushConsumer 消费消息时，若消息处理逻辑出现预期之外的阻塞导致消息处理一直无法执行成功，SDK会按照消费超时处理强制提交消费失败结果，并按照消费重试逻辑进行处理。消息超时，请参见PushConsumer消费重试策略。

注：出现消费超时情况时，SDK虽然提交消费失败结果，但是当前消费线程可能仍然无法响应中断，还会继续处理消息。

优缺点

优点：
- 灵活性高：消费者可以完全控制消息拉取的时机和频率，适合对消费策略要求较高的场景。
- 流控能力强：通过手动控制拉取频率，消费者能够避免消息积压和系统过载。
缺点：
- 实现复杂：消费者需要手动管理偏移量、处理消息的幂等性问题，并实现流控逻辑。
- 实时性较低：由于消息是被动拉取的，实时性相对较低。

Java 代码示例

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.Set;

public class PullConsumerExample {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("ConsumerGroupName");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.start();

        Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest");
        for (MessageQueue mq : mqs) {
            long offset = getMessageQueueOffset(mq);
            PullResult pullResult = consumer.pullBlockIfNotFound(mq, "*", offset, 32);
            processPullResult(pullResult);
            putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
        }

        consumer.shutdown();
    }

    private static long getMessageQueueOffset(MessageQueue mq) {
        // 获取消费队列的偏移量
        return 0;
    }

    private static void putMessageQueueOffset(MessageQueue mq, long offset) {
        // 更新消费队列的偏移量
    }

    private static void processPullResult(PullResult pullResult) {
        // 处理拉取的消息
        switch (pullResult.getPullStatus()) {
            case FOUND:
                // 处理消息
                break;
            case NO_NEW_MSG:
                // 没有新消息
                break;
            default:
                break;
        }
    }
}

SimpleConsumer

定义与概述

SimpleConsumer是RocketMQ 4.8.0版本之后引入的一种消费模式，它是PullConsumer的简化版本，适用于需要拉取消息但不想管理复杂消费逻辑的场景。SimpleConsumer简化了消息拉取和流控的实现，提供了更为直观和易用的API。

原理机制

简化的拉取模型：SimpleConsumer内置了部分PullConsumer的逻辑，如偏移量管理和拉取频率控制，使开发者可以更专注于业务逻辑的实现。
异步拉取：SimpleConsumer支持异步拉取消息，进一步简化了使用流程，并减少了开发者在流控管理上的负担。
偏移量自动管理：SimpleConsumer可以自动管理消息的偏移量，无需手动维护，降低了重复消费的风险。

使用场景

轻量级消费：适合对消费量不大或对消费逻辑要求较简单的场景，例如小型任务处理和轻量级的数据收集。
异步任务处理：适合需要异步处理消息的场景，通过异步拉取简化了业务逻辑的实现。

优缺点

优点：
- 简单易用：相较于PullConsumer，SimpleConsumer大大简化了开发难度，降低了使用门槛。
- 自动管理：自动偏移量管理和异步拉取降低了消息丢失和重复消费的风险。
缺点：
- 灵活性不足：由于自动管理逻辑的加入，SimpleConsumer在一些复杂场景下可能不如PullConsumer灵活。
- 适用场景有限：由于其简化的设计，SimpleConsumer更适合处理较为简单的消费逻辑。

Java 代码示例

import org.apache.rocketmq.client.consumer.SimpleConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.List;

public class SimpleConsumerExample {
    public static void main(String[] args) throws MQClientException {
        SimpleConsumer consumer = new SimpleConsumer("ConsumerGroupName", "localhost:9876");

        MessageQueue mq = new MessageQueue("TopicTest", "BrokerName", 0);
        long offset = consumer.searchOffset(mq, System.currentTimeMillis());

        List<Message> messages = consumer.pull(mq, "*", offset, 10);
        for (Message message : messages) {
            System.out.printf("Received Message: %s%n", new String(message.getBody()));
        }

        consumer.shutdown();
    }
}

PushConsumer

定义与概述

PushConsumer是RocketMQ中最常用的消费模式，Broker主动将消息推送给消费者，消费者只需关注如何处理接收到的消息。PushConsumer是事件驱动的消息消费模式，适用于需要实时处理消息的场景。

原理机制

消息推送：Broker会自动将新到达的消息推送到消费者，消费者只需实现相应的处理逻辑。
并发消费：PushConsumer支持多线程并发消费，通过配置线程池可以提高消息处理的并发性。
消费模式：支持两种消费模式：并发消费（ConsumeConcurrently）和顺序消费（ConsumeOrderly）。并发消费不保证消息的顺序性，而顺序消费保证同一个队列内的消息按顺序消费。

使用场景

实时处理：适合需要实时处理消息的场景，如金融交易、监控告警、实时数据分析等。
并发消费：适合需要高并发处理的场景，通过配置多线程提高消费吞吐量。

优缺点

优点：
- 实时性高：消息被推送到消费者后可以立即处理，适用于需要实时响应的业务场景。
- 易于实现：PushConsumer实现简单，开发者只需关注业务逻辑的实现，而无需关心消息拉取和偏移量管理。
缺点：
- 流控难度大：推模式下流控难度较大，可能出现消息积压的情况，尤其是在消费处理速度跟不上消息推送速度时。
- 并发限制：虽然支持多线程并发消费，但对于顺序消费模式，仍然可能存在并发处理的局限性。

Java 代码示例

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

public class PushConsumerExample {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("TopicTest", "*");

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                System.out.printf("Received Message: %s%n", new String(msg.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.start();
    }
}