RocketMQ生产者的发送消息源码

2023-08-26 05:24:06

RocketMQ 生产者发送消息机制剖析

导言

RocketMQ 作为一款分布式消息队列系统，提供卓越的消息传递服务。其中，生产者模块肩负着将消息发送至服务器的重任。RocketMQ 的生产者支持同步、异步和单向三种消息发送模式，以及批量消息发送功能。本文将深入剖析这些机制，帮助你更全面地理解 RocketMQ 生产者组件。

同步发送消息

同步发送模式最简单直观，也是最常用的模式。其工作流程如下：

创建一个 Producer 对象
调用 Producer.send() 方法发送消息
Producer.send() 方法将消息发送至 Broker
Broker 收到消息后将其写入磁盘
生产者收到 Broker 响应，确认消息发送成功

由于同步等待 Broker 响应，这种模式提供了最可靠的消息发送保障。

异步发送消息

异步发送模式与同步模式的主要区别在于，它不会等待 Broker 响应。其工作流程如下：

创建一个 Producer 对象
调用 Producer.sendAsync() 方法发送消息
Producer.sendAsync() 方法将消息发送至 Broker
Broker 收到消息后将其写入磁盘
生产者不会收到 Broker 响应，无法获知消息是否发送成功

虽然异步模式降低了发送延迟，但它牺牲了消息发送的可靠性。

单向发送消息

单向发送模式是异步模式的极端版本。其工作流程与异步模式类似，但 Broker 不会将消息写入磁盘。这意味着单向模式既不能保证消息可靠性，也无法获得消息发送状态反馈。

批量发送消息

批量发送消息功能可以提高生产者的吞吐量。其工作流程如下：

创建一个 Producer 对象
调用 Producer.sendBatch() 方法发送消息
Producer.sendBatch() 方法将消息批量发送至 Broker
Broker 收到消息后将其写入磁盘
生产者收到 Broker 响应，确认消息发送成功

批量发送通过减少网络交互次数来提升性能，适合发送体积较大的消息。

代码示例

// 创建 Producer 对象
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.start();

// 同步发送消息
Message message = new Message("TopicTest", "TagA", "Hello RocketMQ".getBytes());
SendResult result = producer.send(message);
System.out.println("同步发送消息：" + result.getMsgId());

// 异步发送消息
producer.sendAsync(message, new SendCallback() {
    @Override
    public void onSuccess(SendResult result) {
        System.out.println("异步发送消息成功：" + result.getMsgId());
    }

    @Override
    public void onException(Throwable e) {
        System.out.println("异步发送消息失败：" + e.getMessage());
    }
});

// 单向发送消息
producer.sendOneway(message);
System.out.println("单向发送消息：" + message.getBody());

// 批量发送消息
List<Message> messages = new ArrayList<>();
messages.add(new Message("TopicTest", "TagA", "Hello RocketMQ 1".getBytes()));
messages.add(new Message("TopicTest", "TagA", "Hello RocketMQ 2".getBytes()));
SendResult result = producer.send(messages);
System.out.println("批量发送消息：" + result.getMsgId());

// 关闭 Producer
producer.shutdown();