当前位置：首页 > article >正文

RocketMQ 中如何实现消息的可靠传递？

article 2025/1/31 4:38:06

引言

作为头部消息队列开源中间件，学习其中的技术方案并且总结可靠性和健壮性，提升我们的架构思维和解决问题的能力。

在 RocketMQ 中实现消息的可靠传递可以从多个方面入手，涵盖生产者、Broker 以及消费者等不同环节。

生产者端

1. 同步发送消息

生产者使用同步发送模式时，会等待 Broker 返回发送结果，确保消息成功发送到 Broker 才会继续后续操作。若发送失败，生产者可以进行重试。

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;

public class SyncProducer {
    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, RemotingException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        Message msg = new Message("TopicTest", "TagA", "Hello RocketMQ".getBytes("UTF-8"));
        try {
            // 同步发送消息
            producer.send(msg);
        } catch (Exception e) {
            // 发送失败，可进行重试等处理
            e.printStackTrace();
        }
        producer.shutdown();
    }
}

2. 重试机制

生产者在发送消息失败时，可配置重试次数。RocketMQ 支持自动重试，当遇到网络抖动、Broker 临时不可用等情况时，会自动尝试重新发送消息。

producer.setRetryTimesWhenSendFailed(3); // 设置发送失败时的重试次数为 3 次

3. 消息幂等性处理

为避免因重试导致消息重复发送，生产者可以为每条消息生成唯一的 ID。Broker 在接收消息时，会根据消息 ID 进行去重处理，确保相同 ID 的消息只被处理一次。

Broker 端

1. 刷盘策略

同步刷盘：当 Broker 收到消息后，会先将消息写入磁盘，再返回响应给生产者。这种策略保证了消息不会因 Broker 异常重启而丢失，但会降低系统的吞吐量。
```
flushDiskType = SYNC_FLUSH
```

异步刷盘：Broker 收到消息后，先将消息写入内存缓冲区，然后立即返回响应给生产者，由专门的线程将消息异步写入磁盘。这种策略性能较高，但在 Broker 异常崩溃时，可能会丢失部分内存中的消息。

2. 主从复制

RocketMQ 支持主从复制架构，主 Broker 接收消息后，会将消息同步复制到从 Broker。当主 Broker 出现故障时，可以切换到从 Broker 继续提供服务，保证消息的可用性。

brokerRole = SYNC_MASTER # 主 Broker 配置为同步主节点
brokerRole = SLAVE # 从 Broker 配置为从节点

消费者端

1. 手动提交消费偏移量

消费者在处理完消息后，手动向 Broker 提交消费偏移量，确保只有在消息处理成功后才更新消费进度。这样，当消费者出现异常时，可以从上次提交的偏移量处继续消费，避免消息丢失。

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.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

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

        // 手动提交消费偏移量
        consumer.setAutoCommit(false);

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        // 处理消息
                        System.out.println(new String(msg.getBody()));
                    } catch (Exception e) {
                        // 处理失败，返回重试
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                // 手动提交消费偏移量
                context.setAckIndex(msgs.size() - 1);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
    }
}