揭秘RocketMQ消费者端的运行核心流程（Pull模式-下）

2023-06-16 01:45:38

RocketMQ消费者端的Pull模式运作流程详解

拉取模式下的消息获取

在RocketMQ的Pull模式下，消费者主动从Broker拉取消息。这与Push模式形成对比，在Push模式下，Broker会主动将消息推送到消费者。Pull模式适用于消息量较小、消费者消费速度较慢的情况，因为在这种情况下，消费者可以根据自己的实际情况主动控制拉取消息的频率和数量。

RocketMQ的消费者端使用pullBlockIfNotFound方法从Broker拉取消息。该方法的工作流程如下：

获取消息队列： 消费者首先根据消费组、主题和队列ID获取对应的消息队列。
设置最大消息数： 消费者设置要拉取的最大消息数。
设置Offset： 消费者设置要从消息队列中拉取消息的开始位置（Offset）。
向Broker发送请求： 消费者向Broker发送拉取消息请求，其中包含了消费组、主题、队列ID、最大消息数和Offset等信息。
Broker处理请求： Broker根据请求中的信息查找对应的消息队列，并从该队列中拉取指定数量的消息。如果队列中没有足够的消息，Broker会等待一段时间，直到有足够的消息可供拉取。
消费者接收消息： 消费者收到消息后，根据自己的业务逻辑对消息进行处理。
Ack消息： 处理完消息后，消费者向Broker发送Ack消息，表示已成功消费该消息。Ack消息的目的是告诉Broker该消息已被消费，以便Broker可以将其从消息队列中删除。
继续消费： 消费者继续拉取和处理消息，直至停止消费或消息队列中没有更多消息可供拉取。

代码示例

public void pullMessage() {
    try {
        // 获取消息队列
        MessageQueue messageQueue = new MessageQueue("TopicTest", "broker-a", 0);
        
        // 设置最大消息数
        int maxMessageNum = 32;
        
        // 设置Offset
        long offset = 0;
        
        // 向Broker发送拉取消息请求
        PullMessageRequest pullMessageRequest = new PullMessageRequest();
        pullMessageRequest.setConsumerGroup("consumerGroup1");
        pullMessageRequest.setMessageQueue(messageQueue);
        pullMessageRequest.setMaxMsgNums(maxMessageNum);
        pullMessageRequest.setQueueOffset(offset);
        PullMessageResponse pullMessageResponse = rocketMQClient.pullMessage(pullMessageRequest);
        
        // 处理消息
        for (MessageExt message : pullMessageResponse.getMessages()) {
            System.out.println("收到消息：" + message.getMsgId());
        }
        
        // Ack消息
        rocketMQClient.ackMessage(pullMessageResponse);
    } catch (Exception e) {
        e.printStackTrace();
    }
}