跳到主要内容

RocketMQ 生产者类型

RocketMQ 是一个分布式消息中间件，广泛应用于大规模分布式系统中。在 RocketMQ 中，生产者（Producer）是负责发送消息的组件。根据发送消息的方式和需求，RocketMQ 提供了三种主要的生产者类型：同步生产者、异步生产者和单向生产者。本文将详细介绍这三种生产者类型，并通过代码示例帮助您理解它们的使用场景。

1. 同步生产者（Sync Producer）​

同步生产者是最常用的生产者类型。它会在发送消息后等待 Broker 的响应，确保消息成功发送到 Broker 后再继续执行后续代码。这种方式适合对消息发送的可靠性要求较高的场景。

代码示例​

java
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class SyncProducer {
    public static void main(String[] args) throws Exception {
        // 实例化生产者
        DefaultMQProducer producer = new DefaultMQProducer("SyncProducerGroup");
        // 设置NameServer地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        for (int i = 0; i < 10; i++) {
            // 创建消息，指定Topic、Tag和消息体
            Message msg = new Message("TopicTest", "TagA", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            // 发送消息并获取发送结果
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }

        // 关闭生产者
        producer.shutdown();
    }
}

输出示例​

SendResult [sendStatus=SEND_OK, msgId=7F0000010A6418B4AAC29F3A7BB80000, offsetMsgId=7F00000100002A9F0000000000000000, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=3], queueOffset=0]
备注

同步生产者适合对消息发送的可靠性要求较高的场景，但可能会因为等待 Broker 的响应而影响性能。

2. 异步生产者（Async Producer）​

异步生产者在发送消息后不会等待 Broker 的响应，而是通过回调函数处理发送结果。这种方式适合对性能要求较高，且能够容忍少量消息丢失的场景。

代码示例​

java
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class AsyncProducer {
    public static void main(String[] args) throws Exception {
        // 实例化生产者
        DefaultMQProducer producer = new DefaultMQProducer("AsyncProducerGroup");
        // 设置NameServer地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        for (int i = 0; i < 10; i++) {
            // 创建消息，指定Topic、Tag和消息体
            Message msg = new Message("TopicTest", "TagA", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            // 异步发送消息，并注册回调函数
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.printf("Message sent successfully: %s%n", sendResult);
                }

                @Override
                public void onException(Throwable e) {
                    System.err.printf("Message sent failed: %s%n", e);
                }
            });
        }

        // 关闭生产者
        producer.shutdown();
    }
}

输出示例​

Message sent successfully: SendResult [sendStatus=SEND_OK, msgId=7F0000010A6418B4AAC29F3A7BB80000, offsetMsgId=7F00000100002A9F0000000000000000, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=3], queueOffset=0]
提示

异步生产者适合对性能要求较高的场景，但需要确保回调函数能够正确处理发送结果。

3. 单向生产者（One-way Producer）​

单向生产者只负责发送消息，不关心消息是否成功到达 Broker。这种方式适合对消息发送的可靠性要求较低，但对性能要求极高的场景。

代码示例​

java
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class OnewayProducer {
    public static void main(String[] args) throws Exception {
        // 实例化生产者
        DefaultMQProducer producer = new DefaultMQProducer("OnewayProducerGroup");
        // 设置NameServer地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        for (int i = 0; i < 10; i++) {
            // 创建消息，指定Topic、Tag和消息体
            Message msg = new Message("TopicTest", "TagA", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            // 单向发送消息
            producer.sendOneway(msg);
            System.out.printf("Message sent: %s%n", msg);
        }

        // 关闭生产者
        producer.shutdown();
    }
}

输出示例​

Message sent: Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=0}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 48]}
警告

单向生产者适合对消息发送的可靠性要求较低的场景，但无法保证消息一定到达 Broker。

实际应用场景​

  • 同步生产者：适用于金融交易、订单处理等对消息可靠性要求极高的场景。
  • 异步生产者：适用于日志收集、实时监控等对性能要求较高的场景。
  • 单向生产者：适用于广告推送、通知消息等对消息可靠性要求较低的场景。

总结​

RocketMQ 提供了三种生产者类型，分别适用于不同的场景。同步生产者适合对消息可靠性要求较高的场景，异步生产者适合对性能要求较高的场景，而单向生产者则适合对消息可靠性要求较低的场景。根据实际需求选择合适的生产者类型，可以显著提高系统的性能和可靠性。

附加资源​

练习​

  1. 尝试使用同步生产者发送消息，并观察发送结果。
  2. 使用异步生产者发送消息，并实现一个回调函数处理发送结果。
  3. 比较同步生产者和异步生产者的性能差异，并分析原因。