用RocketMQ这么久，才知道消息可以这样写

淩亂°似流年 2024-04-08 10:51 74阅读 0赞

![format_png][]

在上一章节中，我们讲解了RocketMQ的基本介绍，作为MQ最重要的就是消息的使用了，今天我们就来带大家如何玩转MQ的消息。

消息中间件，英文Message Queue，简称MQ。它没有标准定义，一般认为：消息中间件属于分布式系统中一个子系统，关注于数据的发送和接收，利用高效可靠的异步消息传递机制对分布式系统中的其余各个子系统进行集成。

**高效：** 对于消息的处理处理速度快，RocketMQ可以达到单机10万+的并发。

**可靠：** 一般消息中间件都会有消息持久化机制和其他的机制确保消息不丢失。

**异步：** 指发送完一个请求，不需要等待返回，随时可以再发送下一个请求，既不需要等待。

**消息中间件不生产消息，只是消息的搬运工。**

![fd6ec398a656416584808ffb78f80105.png][]

首先Message包含的内容主要有几个方面组成：id(MQ自动生成)、Topic、tag、proerties、内容。

消息的发送分为：

*  普通消息
 *  顺序消息
 *  延时消息
 *  批量消息
 *  分布式消息

### 普通消息 ###

**普通消息的发送方式主要有三种：发送同步消息、发送异步消息、单向发送**

我们可以先使用 `RocketMQ` 提供的原生客户端的API，在 `SpringBoot、SpringCloudStream` 也进行了集成，但本质上这些也是基于原生API的封装，所以我们只需要掌握原生API的时候，其他的也就无师自通了。

想要使用 `RocketMQ`中的API，就需要先导入对应的客户端依赖

<dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>4.9.2</version>
    </dependency>
    复制代码

**消息发送者的步骤分为：**

1.  创建消息生产者 producer，执行生产者组名
2.  指定Nameserver地址
3.  启动producer
4.  创建消息对象，指定Topic、Tag和消息体
5.  发送消息
6.  关闭生产者producer

**消息消费者的步骤分为：**

1.  创建消费者 Consumer，指定消费者组名
2.  指定Nameserver地址
3.  订阅主题Topic和Tag
4.  设置回调函数，处理消息
5.  启动消费者consumer

#### 发送同步消息 ####

发送同步消息是说消息发送方发出数据后，同步等待，一直等收到接收方发回响应之后才发下一个请求。这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。

流程如下所示：

![264aa50dd77d4b0393fca18a3e34661f.png][]

package org.apache.rocketmq.example.quickstart;
    
    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;
    /**
     * 同步发送
     */
    public class SyncProducer {
        public static void main(String[] args) throws Exception{
            // 实例化消息生产者Producer
            DefaultMQProducer producer = new DefaultMQProducer("group_test");
    
            // 设置NameServer的地址
            producer.setNamesrvAddr("127.0.0.1:9876");
            //producer.setSendLatencyFaultEnable(true);
    
            // 启动Producer实例
            producer.start();
    
    
            for (int i = 0; i < 10; i++) {
                // 创建消息，并指定Topic，Tag和消息体
                Message msg = new Message("TopicTest" /* Topic */,
                        "TagA" /* Tag */,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                );
                // 发送消息到一个Broker
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            }
            //如果不再发送消息，关闭Producer实例。
            producer.shutdown();
        }
    }
    
    复制代码

响应结果如下所示：

![format_png 1][]

*  **msgId：** 消息的全局唯一标识（RocketMQ的ID生成是使用机器IP和消息偏移量的组成），由消息队列 MQ 系统自动生成，唯一标识某条消息。
 *  **sendStatus：** 发送的标识：成功，失败等
 *  **queueId：** queueId是Topic的分区；Producer发送具体一条消息的时，对应选择的该Topic下的某一个Queue的标识ID。
 *  **queueOffset：** Message queue是无限长的数组。一条消息进来下标就会涨1,而这个数组的下标就是queueOffset，queueOffset是从0开始递增。

![040f7343b0494764992a97fb2e8fc98e.png][]

在上面代表的是四个queue，而maxOffset代表我们发送消息的数量，之前发送过消息，所以大家现在看到的数量是17、18...这种，当你在运行一次发送消息时，就会看到十条消息会分布在不同机器上

![d22fcfb6d5244f58a718d991ed90448e.png][]

#### 发送异步消息 ####

异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。消息发送方在发送了一条消息后，不等接收方发回响应，接着进行第二条消息发送。发送方通过回调接口的方式接收服务器响应，并对响应结果进行处理。

![4ac69d4092444cf6bf5c8cab7da0d7f6.png][]

流程如下：

package com.muxiaonong.normal;
    
    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{
            // 实例化消息生产者Producer
            DefaultMQProducer producer = new DefaultMQProducer("group_test");
            // 设置NameServer的地址
            producer.setNamesrvAddr("127.0.0.1:9876");
            // 启动Producer实例
            producer.start();
            for (int i = 0; i < 10; i++) {
                final int index = i;
                // 创建消息，并指定Topic，Tag和消息体
                Message msg = new Message("TopicTest", "TagA", "OrderID888",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                // SendCallback接收异步返回结果的回调
                producer.send(msg, new SendCallback() {
                    //发送成功
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.printf("%s%n", sendResult);
                    }
                    //发送异常
                    @Override
                    public void onException(Throwable e) {
                        System.out.printf("%-10d Exception %s %n", index, e);
                        e.printStackTrace();
                    }
                });
            }
            Thread.sleep(10000);
            // 如果不再发送消息，关闭Producer实例。
            producer.shutdown();
        }
    }
    
    复制代码

发送成功报文：

![format_png 2][]

我们在dashbord下看到已经成功拿到消息了

![48a10f2708884bf0803dcc75bc0d4e3a.png][]

#### 单向发送 ####

这种方式不需要我们特别关心发送结果的场景，比如日志发送、单向发送特点是发送方只需要负责发送消息，不需要等待服务器回应且没有回调函数触发，发送请求不需要等待应答，只管发，这种放松方式过程耗时很短，一般在微妙级别。

流程如下：

![dcdcc8f7816c44e8af2004314585c877.png][]

package com.muxiaonong.normal;
    
    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{
            // 实例化消息生产者Producer对象
            DefaultMQProducer producer = new DefaultMQProducer("group_test");
            // 设置NameServer的地址
            producer.setNamesrvAddr("127.0.0.1:9876");
            // 启动Producer实例
            producer.start();
            for (int i = 0; i < 10; i++) {
                // 创建消息，并指定Topic，Tag和消息体
                Message msg = new Message("TopicTest" /* Topic */,
                        "TagA" /* Tag */,
                        ("Hello RocketMQ " + i

[format_png]: https://img-blog.csdnimg.cn/img_convert/35c73beac868181d341115dddb7ab42f.webp?x-oss-process=image/format,png
[fd6ec398a656416584808ffb78f80105.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/cc483ae15e0543c694f629fce6eb5e30.png
[264aa50dd77d4b0393fca18a3e34661f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/892da9a7484449ac9ff2846301a9b503.png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/7b84593bd525806303084063ad9237a8.webp?x-oss-process=image/format,png
[040f7343b0494764992a97fb2e8fc98e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/a3dbadbf02b64856bb8fe79225ca4a8b.png
[d22fcfb6d5244f58a718d991ed90448e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/a069b458848f44afa7e16f43af6bb682.png
[4ac69d4092444cf6bf5c8cab7da0d7f6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/28a4c42009f54e02b62ef2903ec3186d.png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/5ed4100f19df59174ddba140678dd174.webp?x-oss-process=image/format,png
[48a10f2708884bf0803dcc75bc0d4e3a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/8e586622fb4e456a967317518dea09a4.png
[dcdcc8f7816c44e8af2004314585c877.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/db34e906d498432d8d555e4fe56073c1.png