楔子：既然开了车，加了油，那就带上好心情上路吧。川藏318公路的豁然也好，全美50号公路的孤独也罢，是奇美，是旷野，是路上的人与风景。

1. 在过去的周末

也许周末是个可以用来闲聊这个故事：

1）在一家人看电视的时候，宝宝他妈说给宝宝放动画片吧，宝宝爸说放哪个呢？宝宝妈问宝宝喜欢看哪个？宝宝说看佩奇，然后宝宝妈跟宝宝爸说放小猪佩奇；等等，这就同步消息，她是在收到接收方返回响应之后再发下一个消息。

2）在看电视的时候，宝宝妈说想吃西瓜，然后又说想吃葡萄，然后又说想吃瓜子，还想吃冰激凌... 于是，宝宝爸就开始去端西瓜，拿瓜子，洗葡萄，再去买冰激凌；等等，这就是异步消息，她并不会等宝宝爸说不或先拿来一样，再发第二第三次请求，即异步消息不需要等待接收方发回响应，接着发送下个消息，而且接收方也不需要一定按照先后顺序完成。

3）在宝宝看小猪佩奇的时候，一般叫他的话他会一动不动，比如宝宝妈喊宝宝他不应，叫他坐在沙发上看也不应，问他那个是谁也不回，简直充耳不闻；等等，这就是单向消息，其特点就是他不会回应你，也就是说只发送请求不等待应答。

2. 一个默认生产者

DefaultMQProducer 是一个普通模式默认的消息生产者，可以支持发送普通消息和顺序消息。

当然还有像 TransactionMQProducer 这样的事务模式下的消息生产者，这里不做为分析对象。

2.1. 源码分析

public class DefaultMQProducer extends ClientConfig implements MQProducer {

    ......

    public DefaultMQProducer() {
        // 默认构造一个叫 DEFAULT_PRODUCER 的生产者组
        this("DEFAULT_PRODUCER", (RPCHook)null);
    }

    public DefaultMQProducer(String producerGroup, RPCHook rpcHook) {
        // 创建 Topic 时的 topicKey，在测试时可指定 Broker 自增模式
        this.createTopicKey = "AUTO_CREATE_TOPIC_KEY";
        // 默认每个 Topic 中默认有4个 Queue 来存储消息
        this.defaultTopicQueueNums = 4;
        // 默认发送超时时长 3000ms
        this.sendMsgTimeout = 3000;
        // 默认情况下，当消息体字节数超过4k时，消息会被压缩(Consumer收到消息会自动解压缩)
        this.compressMsgBodyOverHowmuch = 4096;
        // 同步发送消息时，消息发送失败后的最大重试次数
        // RocketMQ 在消息重试机制上有很好的支持，但是重试可能会引起重复消息的问题，这需要在逻辑上进行幂等处理
        this.retryTimesWhenSendFailed = 2;
        // 异步发送时的最大重试次数，类似 retryTimesWhenSendFailed
        this.retryTimesWhenSendAsyncFailed = 2;
        // 如果消息发送成功，但是返回 SendResult != SendStatus.SEND_OK，是否尝试发往别的 Broker
        this.retryAnotherBrokerWhenNotStoreOK = false;
        // 默认最大消息长度:4M，当消息长度超过限制时，RocketMQ 会自动抛出异常
        this.maxMessageSize = 4194304;
        // 生产者组
        this.producerGroup = producerGroup;
        // 构建一个默认生产者
        this.defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
    }

    ......

2.2 实现示例

往下看，详见三种普通消息的实现和代码分析。有点小长，但很简单。

3. 两种消费方式

在 RocketMQ 中，消费者 Consumer 分为两类：MQPullConsumer（DefaultMQPullConsumer 为其实现类）和 MQPushConsumer（DefaultMQPushConsumer 为其实现类）。但二者其本质都是 pull 模式，即 Consumer轮询从 Broker 拉取消息。

3.1. 拉取式消费（Pull Consumer）

在 pull 方式中，需要应用自己实现拉取消息的过程，首先通过消费的 Topic 拿到 MessageQueue 集合，并遍历MessageQueue 集合，然后针对每个 MessageQueue 批量拉取消息。取完一次后，记录 MessageQueue 下一次要取的起始 offset，取完后再换下一个 MessageQueue。Pull 方式中 Consumer 与 Broker 建立的是短连接。

3.2. 推动式消费（Push Consumer）

在 push 方式中，Consumer 把轮询的过程封装了。当应用注册 MessageListener 后，Broker 接收到消息时，会自动回调MessageListener 的 consumeMessage() 方法，在 Consumer 端执行消费。对于应用来说，这个过程好像是消息自动推送过来的。Push 方式中 Consumer 与 Broker 建立了长连接。

4. 三类普通消息

使用 RocketMQ 发送三种类型的消息：同步消息、异步消息和单向消息。其中前两种消息是可靠的，因为会有发送是否成功的应答。

前置依赖：开发前，我们需要加入相关 pom 依赖：

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.3.0</version>
</dependency>

4.1 可靠同步发送

**原理简解：**同步发送是指消息发送方发出数据后，会在收到接收方发回响应之后才发下一个数据包的通讯方式。

**应用场景：**这种可靠同步方式发送应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等。

4.1.1 源码与示例

同步消息生产者（Producer）

package com.meiwei.service.mq.tcp.producer;

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

import java.util.Date;

/**
 * 可靠同步发送 - 生产者
 * <p>
 * 原理
 * 同步发送是指消息发送方发出数据后，会在收到接收方发回响应之后才发下一个数据包的通讯方式。
 * <p>
 * 应用场景
 * 此种方式应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等。
 */
public class SimpleSyncMqProducer {

    // Topic 为 Message 所属的一级分类，就像学校里面的初中、高中
    // Topic 名称长度不得超过 64 字符长度限制，否则会导致无法发送或者订阅
    private static final String MQ_CONFIG_TOPIC = "TOPIC_MEIWEI_SMS_NOTICE_TEST";

    // Tag 为 Message 所属的二级分类，比如初中可分为初一、初二、初三；高中可分为高一、高二、高三
    private static final String MQ_CONFIG_TAG_PUSH = "PID_MEIWEI_SMS_SYNC";

    public static void main(String[] args) throws Exception {
        // 声明并实例化一个 producer 生产者来产生消息
        // 需要一个 producer group 名字作为构造方法的参数
        DefaultMQProducer producer = new DefaultMQProducer("meiwei-producer-simple-sync");

        // 指定 NameServer 地址列表，多个nameServer地址用半角分号隔开。此处应改为实际 NameServer 地址
        // NameServer 的地址必须有，但也可以通过启动参数指定、环境变量指定的方式设置，不一定要写死在代码里
        producer.setNamesrvAddr("127.0.0.1:9876");

        // 在发送MQ消息前，必须调用 start 方法来启动 Producer，只需调用一次即可
        producer.start();

        // 循环发送MQ测试消息
        String content = "";
        for (int i = 0; i < 5; i++) {
            content = "【MQ测试消息】可靠同步发送 " + i;

            // Message Body 可以是任何二进制形式的数据，消息队列不做任何干预，需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
            Message message = new Message(MQ_CONFIG_TOPIC, MQ_CONFIG_TAG_PUSH, content.getBytes(RemotingHelper.DEFAULT_CHARSET));

            // 发送消息。send 方法默认使用的是同步发送方式，有返回结果
            // 发送消息到一个 Broker
            SendResult sendResult = producer.send(message);

            // 同步发送消息，只要不抛异常就是成功
            if (sendResult != null) {
                // 消息发送成功
                System.out.printf("Send MQ message success! Topic: %s, Tag: %s, MsgId: %s, Message: %s %n",
                        message.getTopic(), message.getTags(), sendResult.getMsgId(), new String(message.getBody()));
            } else {
                System.out.println(new Date() + " Send MQ message failed! Topic: " + message.getTopic());
            }
        }

        // 在发送完消息之后，销毁 Producer 对象。如果不销毁也没有问题
        producer.shutdown();
    }
}

同步消息消费者（Consumer）【Push消费方式】

package com.meiwei.service.mq.tcp.consumer;

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.common.message.MessageExt;

import java.util.List;

/**
 * 可靠同步发送 - 消费者（Push模式）
 */
public class SimpleMqPushSyncConsumer {
    // Topic 为 Message 所属的一级分类，就像学校里面的初中、高中
    // Topic 名称长度不得超过 64 字符长度限制，否则会导致无法发送或者订阅
    // Message 所属的 Topic 一级分类，须要与提供者的频道保持一致才能消费到消息内容
    private static final String MQ_CONFIG_TOPIC = "TOPIC_MEIWEI_SMS_NOTICE_TEST";
    private static final String MQ_CONFIG_TAG_PUSH = "PID_MEIWEI_SMS_SYNC";

    public static void main(String[] args) throws Exception {
        // 声明并初始化一个 consumer
        // 需要一个 consumer group 名字作为构造方法的参数
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("meiwei-consumer-simple-sync-push");

        // 同样也要设置 NameServer 地址，须要与提供者的地址列表保持一致
        consumer.setNamesrvAddr("127.0.0.1:9876");

        // 设置 consumer 所订阅的 Topic 和 Tag，*代表全部的 Tag
        consumer.subscribe(MQ_CONFIG_TOPIC, MQ_CONFIG_TAG_PUSH);

        // 设置一个Listener，主要进行消息的逻辑处理
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                list.forEach(msg->{
                    System.out.printf("Thread: %s, Topic: %s, Tags: %s, MsgId: %s, Message: %s %n",
                            Thread.currentThread().getName(),
                            msg.getTopic(),
                            msg.getTags(),
                            msg.getMsgId(),
                            new String(msg.getBody()));
                });

                // 返回消费状态
                // CONSUME_SUCCESS 消费成功
                // RECONSUME_LATER 消费失败，需要稍后重新消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 调用 start() 方法启动 consumer
        consumer.start();
        System.out.println("Simple Consumer Started.");
    }
}

同步消息消费者（Consumer）【Pull消费方式】

package com.meiwei.service.mq.tcp.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * 可靠同步发送 - 消费者（Pull模式）
 */
public class SimpleMqPullSyncConsumer {

    // 记录每个 MessageQueue 的消费位点 offset，可以持久化到 DB 或缓存 Redis，这里作为演示就保存在程序中
    private static final Map<MessageQueue, Long> OFFSE_TABLE = new HashMap<MessageQueue, Long>();

    // Message 所属的 Topic 一级分类，须要与提供者的频道保持一致才能消费到消息内容
    private static final String MQ_CONFIG_TOPIC = "TOPIC_MEIWEI_SMS_NOTICE_TEST";
    private static final String MQ_CONFIG_TAG_PUSH = "PID_MEIWEI_SMS_SYNC";

    public static void main(String[] args) throws Exception {
        // 声明并初始化一个 consumer
        // 需要一个 consumer group 名字作为构造方法的参数
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("meiwei-consumer-simple-sync-pull");
        // 同样也要设置 NameServer 地址，须要与提供者的地址列表保持一致
        consumer.setNamesrvAddr("127.0.0.1:9876");

        // 调用 start() 方法启动 consumer
        consumer.start();
        System.out.println("Simple Consumer Started.");

        // 获取该MessageQueue的消费位点
        Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues(MQ_CONFIG_TOPIC);

        // 遍历MessageQueue，获取Message
        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);

                    // 记录offset
                    putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
                    switch (pullResult.getPullStatus()) {
                        // 拉取到消息
                        case FOUND:
                            break;
                        // 没有匹配的消息
                        case NO_MATCHED_MSG:
                            break;
                        // 暂时没有新消息
                        case NO_NEW_MSG:
                            break SINGLE_MQ;
                        // offset非法
                        case OFFSET_ILLEGAL:
                            break;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

    }

    private static long getMessageQueueOffset(MessageQueue mq) {
        Long offset = OFFSE_TABLE.get(mq);
        if (offset != null)
            return offset;

        return 0;
    }

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

4.1.2 测试及结果

同步消息生产者（Producer）发送结果：

同步消息消费者（Consumer）消费结果：

4.1.3 源码分析

public class DefaultMQProducerImpl implements MQProducerInner {

    ......

    // 默认使用的是同步发送方式
    public SendResult send(Message msg, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        return this.sendDefaultImpl(msg, CommunicationMode.SYNC, (SendCallback)null, timeout);
    }

    ......

public enum CommunicationMode {
    SYNC,
    ASYNC,
    ONEWAY;

    private CommunicationMode() {
    }
}

可能看到，发送消息的 send 方法默认使用的是同步发送方式，有返回结果。

4.2. 可靠异步发送

**原理简解：**异步发送是指发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式。消息队列 MQ 的异步发送，需要用户实现异步发送回调接口（SendCallback）。消息发送方在发送了一条消息后，不需要等待服务器响应即可返回，进行第二条消息发送。发送方通过回调接口接收服务器响应，并对响应结果进行处理。

**应用场景：**异步发送一般用于链路耗时较长，对 RT 响应时间较为敏感的业务场景，即发送端不能容忍长时间地等待 Broker 的响应。例如用户视频上传后通知启动转码服务，转码完成后通知推送转码结果等。

4.2.1 源码与示例

异步消息生产者（Producer）

package com.meiwei.service.mq.tcp.producer;

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;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 可靠异步发送 - 生产者
 * <p>
 * 原理
 * 异步发送是指发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式。
 * 消息队列 MQ 的异步发送，需要用户实现异步发送回调接口（SendCallback）。
 * 消息发送方在发送了一条消息后，不需要等待服务器响应即可返回，进行第二条消息发送。
 * 发送方通过回调接口接收服务器响应，并对响应结果进行处理。
 */
public class SimpleAsyncMqProducer {

    // Topic 为 Message 所属的一级分类，就像学校里面的初中、高中
    // Topic 名称长度不得超过 64 字符长度限制，否则会导致无法发送或者订阅
    private static final String MQ_CONFIG_TOPIC = "TOPIC_MEIWEI_SMS_NOTICE_TEST";

    // Tag 为 Message 所属的二级分类，比如初中可分为初一、初二、初三；高中可分为高一、高二、高三
    private static final String MQ_CONFIG_TAG_ASYNC = "PID_MEIWEI_SMS_ASYNC";

    public static void main(String[] args) throws Exception {
        // 声明并实例化一个 producer 生产者来产生消息
        // 需要一个 producer group 名字作为构造方法的参数
        DefaultMQProducer producer = new DefaultMQProducer("meiwei-producer-simple-async");

        // 指定 NameServer 地址列表，多个nameServer地址用半角分号隔开。此处应改为实际 NameServer 地址
        // NameServer 的地址必须有，但也可以通过启动参数指定、环境变量指定的方式设置，不一定要写死在代码里
        producer.setNamesrvAddr("127.0.0.1:9876");

        // 在发送MQ消息前，必须调用 start 方法来启动 Producer，只需调用一次即可
        producer.start();

        // 设置重试次数，默认情况下是2次重试
        producer.setRetryTimesWhenSendFailed(0);

        int msgCount = 3;
        // 实例化一个倒计数器，count 指定计数个数
        final CountDownLatch countDownLatch = new CountDownLatch(msgCount);

        // 循环发送MQ测试消息
        String content = "";
        for (int i = 0; i < 5; i++) {
            // 配置容灾机制，防止当前消息异常时阻断发送流程
            try {
                final int index = i;
                content = "【MQ测试消息】可靠异步发送 " + index;

                // Message Body 可以是任何二进制形式的数据，消息队列不做任何干预，需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
                Message message = new Message(MQ_CONFIG_TOPIC, MQ_CONFIG_TAG_ASYNC, content.getBytes(RemotingHelper.DEFAULT_CHARSET));
                producer.send(message, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        // 计数减一
                        countDownLatch.countDown();
                        // 消息发送成功
                        System.out.printf("Send MQ message success! Topic: %s, Tag: %s, MsgId: %s, Message: %s %n",
                                message.getTopic(), message.getTags(), sendResult.getMsgId(), new String(message.getBody()));
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        // 计数减一
                        countDownLatch.countDown();
                        // 消息发送失败
                        System.out.printf("%-10d Exception %s %n", index, throwable);
                        throwable.printStackTrace();
                    }
                });
            } catch (Exception e) {
                // 消息发送失败
                System.out.printf("%-10d Exception %s %n", i, e);
                e.printStackTrace();
            }
        }

        // 等待，当计数减到0时，所有线程并行执行
        countDownLatch.await(5, TimeUnit.SECONDS);

        // 在发送完消息之后，销毁 Producer 对象。如果不销毁也没有问题
        producer.shutdown();
    }
}

异步消息消费者（Consumer）【Push消费方式】

package com.meiwei.service.mq.tcp.consumer;

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.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 可靠异步发送 - 消费者（Push模式）
 */
public class SimpleMqPushAsyncConsumer {
    // Topic 为 Message 所属的一级分类，就像学校里面的初中、高中
    // Topic 名称长度不得超过 64 字符长度限制，否则会导致无法发送或者订阅
    // Message 所属的 Topic 一级分类，须要与提供者的频道保持一致才能消费到消息内容
    private static final String MQ_CONFIG_TOPIC = "TOPIC_MEIWEI_SMS_NOTICE_TEST";
    private static final String MQ_CONFIG_TAG_PUSH = "PID_MEIWEI_SMS_ASYNC";

    public static void main(String[] args) throws Exception {
        // 声明并初始化一个 consumer
        // 需要一个 consumer group 名字作为构造方法的参数
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("meiwei-consumer-simple-async");

        // 同样也要设置 NameServer 地址，须要与提供者的地址列表保持一致
        consumer.setNamesrvAddr("127.0.0.1:9876");

        // 设置 consumer 的消费策略
        // CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
        // CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
        // CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // 设置 consumer 所订阅的 Topic 和 Tag，*代表全部的 Tag
        consumer.subscribe(MQ_CONFIG_TOPIC, MQ_CONFIG_TAG_PUSH);

        // 设置一个Listener，主要进行消息的逻辑处理
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println(Thread.currentThread().getName() + " Receive new message: " + list);
                for (MessageExt msg : list) {
                    System.out.printf("Thread: %s, Topic: %s, Tags: %s, MsgId: %s, Message: %s %n",
                            Thread.currentThread().getName(),
                            msg.getTopic(),
                            msg.getTags(),
                            msg.getMsgId(),
                            new String(msg.getBody()));
                }

                // 返回消费状态
                // CONSUME_SUCCESS 消费成功
                // RECONSUME_LATER 消费失败，需要稍后重新消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 调用 start() 方法启动 consumer
        consumer.start();
        System.out.println("Simple Consumer Started.");
    }
}

4.2.2 测试及结果

异步消息生产者（Producer）发送结果：

异步消息消费者（Consumer）消费结果：

4.3. 单向发送

**原理简解：**单向（Oneway）发送特点为发送方只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答。此方式发送消息的过程耗时非常短，一般在微秒级别。

**应用场景：**适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集。

4.3.1 源码与示例

单向消息生产者（Producer）

package com.meiwei.service.mq.tcp.producer;

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

/**
 * 单向（Oneway）发送 - 生产者
 * <p>
 * 原理
 * 单向（Oneway）发送特点为发送方只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答。
 * 此方式发送消息的过程耗时非常短，一般在微秒级别。
 * <p>
 * 应用场景
 * 适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集。
 */
public class SimpleOnewayMqProducer {

    // Topic 为 Message 所属的一级分类，就像学校里面的初中、高中
    // Topic 名称长度不得超过 64 字符长度限制，否则会导致无法发送或者订阅
    private static final String MQ_CONFIG_TOPIC = "TOPIC_MEIWEI_SMS_NOTICE_TEST";

    // Tag 为 Message 所属的二级分类，比如初中可分为初一、初二、初三；高中可分为高一、高二、高三
    private static final String MQ_CONFIG_TAG_ONEWAY = "PID_MEIWEI_SMS_ONEWAY";

    public static void main(String[] args) throws Exception {
        // 声明并实例化一个 producer 生产者来产生消息
        // 需要一个 producer group 名字作为构造方法的参数
        DefaultMQProducer producer = new DefaultMQProducer("meiwei-producer-oneway");

        // 指定 NameServer 地址列表，多个nameServer地址用半角分号隔开。此处应改为实际 NameServer 地址
        // NameServer 的地址必须有，但也可以通过启动参数指定、环境变量指定的方式设置，不一定要写死在代码里
        producer.setNamesrvAddr("127.0.0.1:9876");

        // 在发送MQ消息前，必须调用 start 方法来启动 Producer，只需调用一次即可
        producer.start();

        // 循环发送MQ测试消息
        String content = "";
        for (int i = 0; i < 5; i++) {
            content = "【MQ测试消息】单向消息发送 " + i;

            // Message Body 可以是任何二进制形式的数据，消息队列不做任何干预，需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
            Message message = new Message(MQ_CONFIG_TOPIC, MQ_CONFIG_TAG_ONEWAY, content.getBytes(RemotingHelper.DEFAULT_CHARSET));

            // 单向发送模式没有返回值，就是说只管发不管发送投递是否成功
            producer.sendOneway(message);

            // 消息发送成功
            System.out.printf("Send MQ message success! Topic: %s, Tag: %s, Message: %s %n",
                    message.getTopic(), message.getTags(), new String(message.getBody()));
        }

        // 在发送完消息之后，销毁 Producer 对象。如果不销毁也没有问题
        producer.shutdown();
    }
}

单向消息消费者（Consumer）【Push消费方式】

package com.meiwei.service.mq.tcp.consumer;

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.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 单向（Oneway）发送 - 消费者（Push模式）
 */
public class SimpleMqPushOnewayConsumer {
    // Topic 为 Message 所属的一级分类，就像学校里面的初中、高中
    // Topic 名称长度不得超过 64 字符长度限制，否则会导致无法发送或者订阅
    // Message 所属的 Topic 一级分类，须要与提供者的频道保持一致才能消费到消息内容
    private static final String MQ_CONFIG_TOPIC = "TOPIC_MEIWEI_SMS_NOTICE_TEST";
    private static final String MQ_CONFIG_TAG_ONEWAY = "PID_MEIWEI_SMS_ONEWAY";

    public static void main(String[] args) throws Exception {
        // 声明并初始化一个 consumer
        // 需要一个 consumer group 名字作为构造方法的参数
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("meiwei-consumer-oneway-push");

        // 同样也要设置 NameServer 地址，须要与提供者的地址列表保持一致
        consumer.setNamesrvAddr("127.0.0.1:9876");

        // 设置 consumer 的消费策略
        // CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
        // CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
        // CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // 设置 consumer 所订阅的 Topic 和 Tag，*代表全部的 Tag
        consumer.subscribe(MQ_CONFIG_TOPIC, MQ_CONFIG_TAG_ONEWAY);

        // 设置一个Listener，主要进行消息的逻辑处理
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println(Thread.currentThread().getName() + " Receive new message: " + list);
                for (MessageExt msg : list) {
                    System.out.printf("Thread: %s, Topic: %s, Tags: %s, MsgId: %s, Message: %s %n",
                            Thread.currentThread().getName(),
                            msg.getTopic(),
                            msg.getTags(),
                            msg.getMsgId(),
                            new String(msg.getBody()));
                }

                // 返回消费状态
                // CONSUME_SUCCESS 消费成功
                // RECONSUME_LATER 消费失败，需要稍后重新消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 调用 start() 方法启动 consumer
        consumer.start();
        System.out.println("Simple Consumer Started.");
    }
}

4.3.2 测试及结果

单向消息生产者（Producer）发送结果：

单向消息消费者（Consumer）消费结果：

5. 发送方式对比

如下概括了三种发送方式的特点和主要区别：

发送方式	发送 TPS	发送结果反馈	可靠性
同步发送	快	有	不丢失
异步发送	快	有	不丢失
单向发送	最快	无	可能丢失

参考资料：
RocketMQ 官网：http://rocketmq.apache.org/docs/motivation/
阿里云消息队列 MQ：https://help.aliyun.com/document_detail/29532.html
阿里巴巴中间件团队：http://jm.taobao.org/2016/11/29/apache-rocketmq-incubation/

RocketMQ进击（二）一个默认生产者，两种消费方式，三类普通消息详解分析

1. 在过去的周末

2. 一个默认生产者

2.1. 源码分析

2.2 实现示例

3. 两种消费方式

3.1. 拉取式消费（Pull Consumer）

3.2. 推动式消费（Push Consumer）

4. 三类普通消息

4.1 可靠同步发送

4.1.1 源码与示例

4.1.2 测试及结果

4.1.3 源码分析

4.2. 可靠异步发送

4.2.1 源码与示例

4.2.2 测试及结果

4.3. 单向发送

4.3.1 源码与示例

4.3.2 测试及结果

5. 发送方式对比

相关文章

RocketMQ进击（二）一个默认生产者，两种消费方式，三类普通消息详解分析

1. 在过去的周末

2. 一个默认生产者

2.1. 源码分析

2.2 实现示例

3. 两种消费方式

3.1. 拉取式消费（Pull Consumer）

3.2. 推动式消费（Push Consumer）

4. 三类普通消息

4.1 可靠同步发送

4.1.1 源码与示例

4.1.2 测试及结果

4.1.3 源码分析

4.2. 可靠异步发送

**4.2.1 **源码与示例

4.2.2 测试及结果

4.3. 单向发送

4.3.1 源码与示例

4.3.2 测试及结果

5. 发送方式对比

相关文章

4.2.1 源码与示例