掘金 后端 ( ) • 2024-05-08 20:38
消息队列的拉模式和推模式
队列 模式 消息

消息队列（Message Queue）是一种应用程序之间传递数据和信息的通信方法。消息队列中的推拉模式（Push-Pull Model）是指消息的生产者（Producer）将消息发送到队列，然后消费者（Consumer）可以按照推（Push）或拉（Pull）的方式从队列中获取消息。

简单阐述

RocketMQ 推模式（Push Mode）

推模式是消息中间件主动将消息推送到消费者，消费者无需主动拉取。这种模式适合对消息处理实时性有较高要求的场景。

实现步骤与代码示例：

  1. 消费者订阅：消费者通过指定Topic和Tag来订阅感兴趣的消息。
  2. 消息监听：设置一个消息监听器来异步接收和处理消息。
  3. 确认消息：处理完成后，发送消息确认（ACK）。
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;

@Service
@RocketMQMessageListener(topic = "test-topic", consumerGroup = "my-consumer-group")
public class MyConsumerService implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("Received message: " + message);
        // 处理消息
    }
}

RocketMQ 拉模式（Pull Mode）

拉模式允许消费者按需拉取消息，适合消息消费量不均匀或需要精细控制处理速度的场景。

实现步骤与代码示例：

  1. 初始化消费者：创建一个消费者实例，并配置必要的参数。
  2. 订阅消息：指定要拉取的Topic。
  3. 循环拉取：在循环中不断拉取消息，并更新消息偏移量。
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.client.exception.MQClientException;
import java.util.Set;

public class PullConsumerService {
    private DefaultMQPullConsumer consumer;
    
    public void start() throws MQClientException {
        consumer = new DefaultMQPullConsumer("my-consumer-group");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.start();
        String topic = "test-topic";
        Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues(topic);
        for (MessageQueue mq : mqs) {
            long offset = consumer.fetchConsumeOffset(mq, true);
            while (true) {
                // 拉取消息
                var pullResult = consumer.pull(mq, "*", offset, 32);
                for (MessageExt message : pullResult.getMsgFoundList()) {
                    System.out.println("Received message: " + new String(message.getBody()));
                }
                offset = pullResult.getNextBeginOffset();
                // 更新偏移量
                consumer.updateConsumeOffset(mq, offset);
            }
        }
    }
}

使用场景比较

  • 推模式：适用于需要快速响应的场景，如实时数据处理和实时通知。
  • 拉模式：适合于处理量大或处理速度变化大的场景，允许应用根据处理能力调整拉取速度。

这两种模式在RocketMQ中可以根据不同的业务需求和系统负载灵活选择。推模式因为其低延迟和服务器端推动的特性而适用于实时性较高的应用场景，而拉模式则更适合于那些需要精细控制消费速度和时间的复杂应用场景。

RocketMQ 推模式关键组件讲解

关键组件概览

  1. 消费者（Consumer）：

    • 负责从Broker订阅并接收消息。
    • 需要设置监听器（Message Listener）来异步处理接收到的消息。

  2. Broker：

    • 消息队列服务器，负责消息的存储和转发。
    • 将消息从存储层推送到订阅的消费者。

  3. 监听器（Message Listener）：

    • 由消费者实现的回调接口。
    • 当消息到达时，Broker会触发这个监听器。

  4. 确认机制（Acknowledgement）：

    • 消费者在成功处理消息后必须向Broker发送确认信号（ACK）。
    • 确保消息不会因为未确认而被重复发送。

示例代码解释

1. 创建消费者并设置监听器

在Spring Boot应用中，我们通常使用@RocketMQMessageListener注解来标记监听器并定义消费者的配置，如订阅的topic和消费者组。

import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.stereotype.Service;

@Service
@RocketMQMessageListener(topic = "your-topic", consumerGroup = "your-consumer-group")
public class ProductListener implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("Received message: " + message);
        // 这里可以添加消息处理逻辑
    }
}

2. 消息监听器的工作原理

  • 当消息到达时，RocketMQ的Broker将消息推送到所有订阅了相应Topic的消费者。
  • 消费者的监听器接收到消息后，onMessage方法被调用，开始处理消息。

3. 确认机制（Acknowledgement）

  • 在RocketMQ中，如果使用事务消息或顺序消费，确认机制尤为重要。
  • Spring Boot集成的RocketMQ客户端通常会自动处理消息确认。

手动确认消息示例

如果需要手动确认消息，可以通过使用RocketMQPushConsumerLifecycleListener接口来访问底层的消费者（Consumer）对象，并进行更细致的控制。

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.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.stereotype.Service;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

@Service
@RocketMQMessageListener(topic = "your-topic", consumerGroup = "your-consumer-group")
public class AdvancedProductListener implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
    @Override
    public void onMessage(MessageExt message) {
        System.out.println("Received message: " + new String(message.getBody()));
        // 处理消息逻辑
    }

    @Override
    public void prepareStart(final org.apache.rocketmq.client.consumer.DefaultMQPushConsumer consumer) {
        // 设置消息监听器来手动确认消息
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            // 手动处理消息
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; // 确认消息
        });
    }
}

在上述示例中，prepareStart方法允许我们访问和修改DefaultMQPushConsumer的配置，包括如何处理消息确认。这种高级用法允许在处理消息时进行复杂的逻辑判断或错误处理。

RocketMQ 拉模式关键组件讲解

RocketMQ的拉模式（Pull Mode）提供了一种方式，允许消费者按自己的需求和节奏主动从Broker拉取消息。这种模式适合于对消息处理有精细控制需求的场景，例如在处理速度或系统资源有限制的情况下。

关键组件

  1. 消费者（Consumer）：

    • 在拉模式中，消费者主动请求消息，而不是被动接收。
    • 消费者需要管理消息偏移量（Offset），以便知道从哪里开始拉取新消息。

  2. 消息队列（Message Queue）：

    • Broker端的队列，存储发布到特定主题的消息。
    • 每个队列在物理上可能分布在不同的Broker上。

  3. 偏移量（Offset）：

    • 消费者需要跟踪每个队列的偏移量，以确保消息能够被顺序消费，且不会丢失或重复。

  4. Broker：

    • 消息中间件的服务器组件，负责消息的存储和管理。
    • 消费者通过发送请求到Broker来拉取消息。

示例代码解释

1. 配置消费者

首先，你需要创建一个服务类，用于初始化消费者，并定期拉取消息：

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Set;

@Service
public class PullConsumerService {

    private DefaultMQPullConsumer consumer;

    @PostConstruct
    public void init() throws MQClientException {
        consumer = new DefaultMQPullConsumer("pullConsumerGroup");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.start();

        new Thread(() -> {
            try {
                // 指定要拉取的Topic
                Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest");
                for (MessageQueue mq : mqs) {
                    long offset = consumer.fetchConsumeOffset(mq, true);
                    while (true) {
                        // 从Broker拉取消息
                        var pullResult = consumer.pullBlockIfNotFound(mq, "*", offset, 32);
                        for (MessageExt message : pullResult.getMsgFoundList()) {
                            System.out.printf("Message Body: %s%n", new String(message.getBody(), RemotingHelper.DEFAULT_CHARSET));
                        }
                        offset = pullResult.getNextBeginOffset();
                        // 更新偏移量
                        consumer.updateConsumeOffset(mq, offset);
                        // 模拟处理延迟
                        Thread.sleep(1000);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
    }
}

2. 代码解释

  • 初始化消费者：创建DefaultMQPullConsumer实例，并设置必要的参数，如消费者组名和服务器地址。
  • 订阅消息队列：通过fetchSubscribeMessageQueues方法获取特定主题的所有消息队列。
  • 拉取消息：使用pullBlockIfNotFound方法从指定的队列拉取消息。这个方法在指定的队列上阻塞，直到找到新消息。
  • 处理消息：拉取到的消息可以在循环中被处理。
  • 更新偏移量：每次拉取后更新消费偏移量，确保下次可以拉取到新消息。

注意

  • 偏移量管理：确保偏移量准确无误是实现有效的消息消费的关键。错误的偏移量可能导致消息丢失或重复。

  • 错误处理：在生产环境中，添加适当的错误处理逻辑至关重要，例如处理网络异常和响应超时。

通过上述示例和说明，你可以看到RocketMQ拉模式提供了较高的灵活性，允许消费者根据实际处理能力和需求来控制消息的拉取。这使得它适用于需要精细控制消费行为的复杂应用场景。

简单案例

在 Spring Boot 应用中使用RocketMQ的推模式

1. 添加RocketMQ依赖

首先，确保你的Spring Boot项目中包含了RocketMQ的客户端依赖。以下是Maven配置示例：

<dependencies>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.2.0</version> <!-- 使用适合你项目的版本 -->
    </dependency>
</dependencies>

2. 配置RocketMQ

在application.properties或application.yml中添加RocketMQ的配置，如服务器地址和消费者组信息：

rocketmq.name-server=127.0.0.1:9876
rocketmq.consumer.group=my-consumer-group

3. 创建消息监听器

在Spring Boot应用中，你可以通过定义一个服务并使用@RocketMQMessageListener注解来创建一个消息监听器。这个监听器会自动订阅指定的Topic和消费者组，并处理接收到的消息。

package com.example.demo.service;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

@Service
@RocketMQMessageListener(topic = "test-topic", consumerGroup = "my-consumer-group")
public class MyConsumerService implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("Received message: " + message);
        // 这里可以添加消息处理逻辑
    }
}

4. 详细解释

  • 消费者组：consumerGroup属性定义了消息的消费者组，这是消费者的逻辑分组，用于消息的负载均衡和故障转移。
  • 主题订阅：topic属性指定了此消费者要订阅的消息主题。
  • 消息处理：onMessage方法是消息到达时调用的回调方法。这里你可以实现具体的业务逻辑处理接收到的消息。

通过以上步骤和示例代码，你可以在Spring Boot应用中实现RocketMQ的推模式。这种模式适用于需要低延迟和高吞吐量的实时消息处理场景。

在 Spring Boot 应用中使用 RocketMQ 的拉模式

1. 添加依赖

首先，确保在你的 Spring Boot 项目中添加了 RocketMQ 的依赖。如果使用 Maven，可以在 pom.xml 文件中添加以下依赖：

<dependencies>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.2.0</version> <!-- 使用适合你项目的版本 -->
    </dependency>
</dependencies>

2. 配置 RocketMQ

在 application.properties 或 application.yml 中添加 RocketMQ 的配置，例如：

rocketmq.name-server=127.0.0.1:9876
rocketmq.consumer.group=my-consumer-group

3. 创建消费者服务

在 Spring Boot 应用中创建一个服务来实现消息的拉取。我们需要手动控制如何从服务器拉取消息：

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.store.OffsetStore;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

@Service
public class RocketMQPullConsumerService {
    @Value("${rocketmq.name-server}")
    private String nameServer;

    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;

    private final Map<MessageQueue, Long> offsetTable = new HashMap<>();

    @PostConstruct
    public void init() throws MQClientException {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(consumerGroup);
        consumer.setNamesrvAddr(nameServer);
        consumer.start();

        // Assume we are subscribing to a topic
        String topic = "yourTopicName";
        Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues(topic);

        for (MessageQueue mq : mqs) {
            System.out.printf("Consume from the queue: %s%n", mq);
            SINGLE_MQ: while (true) {
                try {
                    PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
                    System.out.printf("%s%n", pullResult);

                    List<MessageExt> foundList = pullResult.getMsgFoundList();
                    if (foundList != null) {
                        for (MessageExt msg : foundList) {
                            // Process your message here
                            System.out.printf("%s%n", new String(msg.getBody()));
                        }
                    }

                    putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
                    switch (pullResult.getPullStatus()) {
                        case FOUND:
                            break;
                        case NO_MATCHED_MSG:
                            break;
                        case NO_NEW_MSG:
                            break SINGLE_MQ;
                        case OFFSET_ILLEGAL:
                            break;
                        default:
                            break;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

        consumer.shutdown();
    }

    private long getMessageQueueOffset(MessageQueue mq) {
        Long offset = offsetTable.get(mq);
        if (offset != null)
            return offset;
        return 0;
    }

    private void putMessageQueueOffset(MessageQueue mq, long offset) {
        offsetTable.put(mq, offset);
    }
}

4. 解析代码

  • 初始化消费者：创建一个 DefaultMQPullConsumer，设置其名字服务器地址和消费者组。
  • 订阅主题：获取特定主题的所有消息队列。
  • 拉取循环：对每个消息队列，尝试拉取消息。这个例子使用 pullBlockIfNotFound 方法，在没有消息时阻塞。
  • 消息处理：处理每个拉取到的消息。
  • 偏移管理：手动管理每个队列的消费偏移量。
相关内容
RocketMQ 源码探究 -- 延迟队列实现
如何实现任意时刻的延迟队列 -- 以 QMQ 为例
🐳 一文学会 《RabbitMQ》 的死信、延时队列
一文读懂半连接队列
🐳 一文学会 消息队列 《RabbitMQ》 的消息可靠性
Volcano社区新版本发布！7大功能全面增强队列能力与调度稳定性
表面简单实则暗藏玄机的面试题：Java数组适合做队列吗？
深入解析消息队列中的死信队列
一次消息队列异常堆积的排查 (crossoverjie.top)
Java 面试宝典：怎么使用 Redis 实现一个延时队列？