RocketMQ快速入门：消息发送、延迟消息、消费重试

一起学编程，让生活更随和！

如果你觉得是个同道中人，欢迎关注博主gzh：【随和的皮蛋桑】。

专注于Java基础、进阶、面试以及计算机基础知识分享🐳。偶尔认知思考、日常水文🐌。

---

1、RocketMQ消息结构

1.1、消息结构

RocketMQ的消息包括基础属性和扩展属性两部分：

1）基础属性

1. topic ： 主题相当于消息的一级分类，具有相同topic的消息将发送至该topic下的消息队列中，比方说一个电商系统可以分为商品消息、订单消息、物流消息等，就可以在broker中创建商品主题、订单主题等，所有商品的消息发送至该主题下的消息队列中。
2. 消息体：即消息的内容 ，可以的字符串、对象等类型（可系列化）。消息的最大长度 是4M。
3. 消息Flag：消息的一个标记，RocketMQ不处理，留给业务系统使用。

2）扩展属性：

1. tag ：相当于消息的二级分类，用于消费消息时进行过滤，可为空 。
2. keys： Message 索引键，在运维中可以根据这些 key 快速检索到消息， 可为空 。
3. waitStoreMsgOK：消息发送时是否等消息存储完成后再返回 。Message 的基础属性主要包括消息所属主题 topic ， 消息 Flag(RocketMQ 不做处理）、 扩展属性、消息体 。

1.2、三种消息发松方式

RocketMQ 支持 3 种消息发送方式 ：

1）同步消息（sync message ）

producer向 broker 发送消息，执行 API 时同步等待， 直到broker 服务器返回发送结果 。

2）异步消息（async message）

producer向 broker 发送消息时指定消息发送成功及发送异常的回调方法，调用 API 后立即返回，
producer发送消息线程不阻塞 ，消息发送成功或失败的回调任务在一个新的线程中执行 。

3）单向消息（oneway message）

producer向 broker 发送消息，执行 API 时直接返回，不等待broker 服务器的结果 。

2、快速搭建工程

2.1、创建rocketmq-demo父工程

pom.xml引入一下依赖：

		<!-- spring-boot父依赖 --><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.3.12.RELEASE</version><relativePath/></parent><dependencies><!-- starter-test --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency><!-- starter-web --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- rocketmq --><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.1.0</version></dependency></dependencies>

2.2、生产者工程

1）创建rocketmq-producer生产者工程

2）新建rocketmq-producer工程的application.yml文件

# 端口号、上下文路径
server:port: 8181servlet:context-path: /rocketmq-producer# 服务名
spring:application:name: rocketmq-producer# rocketmq
rocketmq:name-server: 106.15.0.30:9876 #命名空间地址和端口号producer:group: demo-producer-group #生产者组

3） 新建启动类

package com.wbs;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;/*** @author yixiujun* @version Id: ProducerApplication.java, v 0.1 Administrator Exp $$* @date 2023-02-15 09:22:20* @desc 生产者启动类*/
@SpringBootApplication
public class ProducerApplication {public static void main(String[] args) {SpringApplication.run(ProducerApplication.class, args);}}

4）创建发送同步消息方法

package com.wbs.test.message;import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component;import javax.annotation.Resource;/*** @author yixiujun* @version Id: ProducerSimple.java, v 0.1 Administrator Exp $$* @date 2023-02-15 09:24:13* @desc rocketmq发送消息类*/
@Component
public class ProducerSimple {@Resourceprivate RocketMQTemplate rocketMQTemplate;/*** 发送同步消息** @param topic 主题* @param msg   消息*/public void sendSyncMsg(String topic, String msg) {rocketMQTemplate.syncSend(topic, msg);}}

5）测试

在test包下创建单元测试ProducerSimpleTest 类：

package com.wbs.test.message;import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;import javax.annotation.Resource;/*** @author yixiujun* @version Id: ProducerSimpleTest.java, v 0.1 Administrator Exp $$* @date 2023-02-19 16:24:20* @desc 生产者发送消息测试*/
@SpringBootTest
@RunWith(SpringRunner.class)
public class ProducerSimpleTest {@Resourceprivate ProducerSimple producerSimple;/*** 测试发送同步消息*/@Testpublic void testSendSyncMsg() {this.producerSimple.sendSyncMsg("my-topic", "第一条同步消息");System.out.println("end...");}}

启动服务器端

• NameServer
• Broker
• console管理端

执行上述单元测试testSendSyncMsg方法，观察控制台和管理端控制台出现end...表示消息发送成功。

进入管理端，查询消息。

2.3、消费者工程

1）创建消息消费者工程rocketmq-consumer

**2）创建application.yml文件 **

# 端口号、上下文路径
server:port: 8182servlet:context-path: /rocketmq-consumer# 服务名
spring:application:name: rocketmq-consumer# rocketmq
rocketmq:name-server: 106.15.0.30:9876producer:group: demo-consumer-group

3）创建启动类

package com.wbs;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;/*** @author yixiujun* @version Id: ConsumerApplication.java, v 0.1 Administrator Exp $$* @date 2023-02-19 16:46:00* @desc 消费者启动类*/
@SpringBootApplication
public class ConsumerApplication {public static void main(String[] args) {SpringApplication.run(ConsumerApplication.class, args);}}

4）消费消息

编写消费消息监听类：

package com.wbs.test.message;import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;/*** @author yixiujun* @version Id: ConsumerSimple.java, v 0.1 Administrator Exp $$* @date 2023-02-19 16:48:16* @desc 消费者监听类*/
@Component
@RocketMQMessageListener(topic = "my-topic", consumerGroup = "demo-consumer-group")
public class ConsumerSimple implements RocketMQListener<String> {/*** 接收到消息后调用此方法* @param s 消息内容*/@Overridepublic void onMessage(String s) {System.out.println(s + "：已被消费!");}}

监听消息队列 需要指定：

• topic：监听的主题
• consumerGroup：消费组，相同消费组的消费者共同消费该主题的消息，它们组成一个集群。

2.4、消息发送过程

启动消费者工程，观察控制台输出“第一条同步消息”消息内容，这说明从消息队列已经读取到消息。

保证消费者工程已启动，再次发送消息，观察控制台是否输出“第一条同步消息”消息内容，输出则说明接收消息成功。

3、消息发送过程

通过快速入门对消息的发送和接收有一个粗略的认识，下边分析具体的消息发送过程，如下图：

消息发送流程如下：

1、Producer从NameServer中获取主题路由信息

Broker将自己的状态上报给NameServer，NameServer中存储了每个Broker及主题、消息队列的信息。

Producer根据 topic从NameServer查询所有消息队列，查询到的结果例如：

[{"brokerName":"Broker‐1","queueId":0},{"brokerName":"Broker‐1","queueId":1},{"brokerName":"Broker‐2","queueId":0},{"brokerName":"Broker‐2","queueId":1}
]

Producer按选择算法从以上队列中选择一个进行消息发送，如果发送消息失败则在下次选择的时候 会规避掉失败的broker。

2、构建消息，发送消息

发送消息前进行校验，比如消息的内容长度不能为0、消息最大长度、消息必要的属性是否具备等（topic、消息体，生产组等）。

如果该topic下还没有队列则自动创建，默认一个topic下自动创建4个写队列，4个读队列 。为什么要多个队列 ？

• 高可用：当某个队列不可用时其它队列顶上。
• 提高并发：发送消息是选择队列进行发送，提高发送消息的并发能力。消息消费时每个消费者可以监听多个队列，提高消费消息的并发能力。

生产组有什么用？

在事务消息中broker需要回查producer，同一个生产组的producer组成一个集群，提高并发能力。

3、监听队列，消费消息

一个消费组可以包括多个消费者，一个消费组可以订阅多个主题。

一个队列同时只允许一个消费者消费，一个消费者可以消费多个队列中的消息。

消费组有两种消费模式：

1）集群模式

一个消费组内的消费者组成一个集群，主题下的一条消息只能被一个消费者消费。

2）广播模式

主题下的一条消息能被消费组下的所有消费者消费。

消费者和broker之间通过推模式和拉模式接收消息，推模式即broker推送给消费者，拉模式是消费者主动从broker查询消息。

4、三种消息发送方式

RocketMQ 支持 3 种消息发送方式 ，即

• 同步消息（sync message ）
• 异步消息（async message）
• 单向消息（oneway message）

4.1、同步消息

参考2快速搭建工程（同步消息的简单示例）。

4.2、异步消息

producer向broker发送消息时指定消息发送成功及发送异常的回调方法，调用 API 后立即返回，producer发送消息线程不阻塞 ，消息发送成功或失败的回调任务在一个新的线程中执行 。

在ProducerSimple中编写发送异步消息的方法：

	/*** 发送异步消息** @param topic 主题* @param msg   消息*/public void sendAsyncMsg(String topic, String msg) {rocketMQTemplate.asyncSend(topic, msg, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {// 回调成功System.out.println(sendResult.getSendStatus());}@Overridepublic void onException(Throwable throwable) {// 回调异常System.out.println(throwable.getMessage());}});}

单元测试：

	/*** 测试发送异步消息** @throws InterruptedException 异常*/@Testpublic void testSendASyncMsg() throws InterruptedException {this.producerSimple.sendAsyncMsg("my-topic", "第一条异步消息");System.out.println("end……");// 异步消息，为跟踪回调线程这里加入延迟Thread.sleep(3000);}

4.3、单向消息

producer向broker发送消息，执行 API 时直接返回，不等待broker服务器的结果 。

在ProducerSimple中编写发送单项消息的方法：

	/*** 发送单向消息** @param topic 主题* @param msg   消息*/public void sendOneWayMsg(String topic, String msg) {this.rocketMQTemplate.sendOneWay(topic, msg);}

测试：

	/*** 测试发送异步消息*/@Testpublic void testSendOneWayMsg() {this.producerSimple.sendOneWayMsg("my-topic", "第一条单项消息");System.out.println("end……");}

5、自定义消息格式

前边我们发送的消息内容格式都是字符串，在生产开发中消息内容格式是相对较复杂的，下面介绍如何对消息格式进行自定义。

JSON是互联网开发中非常常用的数据格式，它具有格式标准，扩展方便的特点，将消息的格式使用JSON进行定义，可以提高消息内容的扩展性，RocketMQ支持传递JSON数据格式。

在生产端和消费端定义模型类：

package com.wbs.test.model;/*** 自定义消息实体类** @author yixiujun* @version Id: OrderExt.java, v 0.1 Administrator Exp $$* @date 2023-02-19 20:55:25*/
@Data
@NoArgsConstructor
@AllArgsConstructor
@ToString
public class OrderExt implements Serializable {private final static Long SERIALIZABLE_UID = -1L;/*** 主键id*/private String id;/*** 创建时间*/private Date createTime;/*** money*/private Long money;/*** 标题*/private String title;}

创建ProducerUserDefineSimple用户发送自定义消息模板类，创建发送消息内容为json格式的方法，

生产端：

package com.wbs.test.message;/*** 用户发送自定义消息模板** @author yixiujun* @version Id: ProducerUserDefineSimple.java, v 0.1 Administrator Exp $$* @date 2023-02-19 21:07:16*/
@Component
public class ProducerUserDefineSimple {@Resourceprivate RocketMQTemplate rocketMQTemplate;/*** 消息内容为json格式** @param topic 主题* @param orderExt   消息实体*/public void sendMsgByJson(String topic, OrderExt orderExt) {// 发送同步消息，消息内容将orderExt转为jsonthis.rocketMQTemplate.convertAndSend(topic, orderExt);System.out.printf("send msg ： %s", orderExt);}}

编写单元测试方法：

	/*** 测试发送JSON格式的内容消息*/@Testpublic void sendMsgByJson() {OrderExt orderExt = new OrderExt();orderExt.setId("001");orderExt.setCreateTime(new Date());orderExt.setMoney(10000L);orderExt.setTitle("这是JSON格式数据");this.producerUserDefineSimple.sendMsgByJson("my-topic", orderExt);System.out.println("end……");}

消费端：

@Component
@RocketMQMessageListener(topic = "my-topic-obj", consumerGroup = "demo-consumer-group-obj")
public class ConsumerUserDefineSimple implements RocketMQListener<String> {/*** 接收到消息后调用此方法** @param s 消息内容*/@Overridepublic void onMessage(String s) {// 如果是json数据，可以将json转为对象OrderExt orderExt = JSON.parseObject(s, OrderExt.class);System.out.println(orderExt);}}

上例实现了RocketMQ传输JSON消息的过程，消费端在接收到JSON手动将JSON转成对象，也可以自动转换成对象，定义新的监听类，RocketMQListener泛型指定要转换的对象类型。

6、延迟消息

6.1、延迟消息介绍

延迟消息也叫作定时消息，比如在电商项目的交易系统中，当用户下单之后超过一段时间之后仍然没有支付，此时就需要将该订单关l闭。

要实现该功能的话，可以在用户创建订单时就发送一条包含订单内容的延迟消息，该消息在一段时间之后投递给消息消费者，当消息消费者接收到该消息后，判断该订单的支付状态，如果处于未支付状态，则将该订单关闭。

RocketMQ的延迟消息实现非常简单，只需要发送消息前设置延迟的时间，延迟时间存在十八个等级
（1s/5s/10s/30s/1m/2m/3m/4m/5m/6m/7m/8m/9m/10m/20m/30m/1h/2h），调用setDelayTimeLevel()设置与时间相对应的延迟级别即可。

6.2、同步消息延迟发送

生产端：

    /*** 发送同步延迟消息（消息内容为json格式）** @param topic     主题* @param orderExt  消息体*/public void sendMsgByJsonDelay(String topic, OrderExt orderExt) {// 发送同步消息，消息内容将orderExt转为jsonMessage<OrderExt> message = MessageBuilder.withPayload(orderExt).build();// 指定发送超时时间（毫秒）和延迟等级this.rocketMQTemplate.syncSend(topic, message, 1000, 3);System.out.printf("send msg ： %s", orderExt);}

消费端监听：

/*** 消费者监听类* * @author yixiujun* @version Id: ConsumerSimple.java, v 0.1 Administrator Exp $$* @date 2023-02-19 16:48:16*/
@Component
@RocketMQMessageListener(topic = "my-topic-obj", consumerGroup = "demo-consumer-group-obj")
public class ConsumerUserDefineSimple implements RocketMQListener<OrderExt> {/*** 根据泛型默认接收将接收到的json数据转化为对应的实体类** @param orderExt json对应实体*/@Overridepublic void onMessage(OrderExt orderExt) {System.out.println(orderExt);}
}

单元测试：

    /*** 测试发送同步延迟消息*/@Testpublic void testSendMsgByJsonDelay() {OrderExt orderExt = new OrderExt();orderExt.setId(UUID.randomUUID().toString());orderExt.setCreateTime(new Date());orderExt.setMoney(20000L);orderExt.setTitle("测试订单");this.producerUserDefineSimple.sendMsgByJsonDelay("my-topic-obj", orderExt);System.out.println("end……");}

6.3、异步消息延迟发送

生产端：

    /*** 发送异步延迟消息（消息内容为json格式）** @param topic     主题* @param orderExt  消息体*/public void sendAsyncMsgByJsonDelay(String topic, OrderExt orderExt) throws JsonProcessingException, RemotingException, InterruptedException, MQClientException {// 消息内容将orderExt转为jsonObjectMapper objectMapper = new ObjectMapper();String jsonData = objectMapper.writeValueAsString(orderExt);org.apache.rocketmq.common.message.Message message = new org.apache.rocketmq.common.message.Message(topic, jsonData.getBytes(StandardCharsets.UTF_8));// 设置延迟等级message.setDelayTimeLevel(3);// 发送异步消息this.rocketMQTemplate.getProducer().send(message, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {try {System.out.println(objectMapper.writeValueAsString(sendResult));} catch (JsonProcessingException e) {e.printStackTrace();}}@Overridepublic void onException(Throwable throwable) {System.out.println(throwable.getMessage());}});System.out.printf("send msg : %s",orderExt);}

消费端监听：

/*** 消费者监听类* * @author yixiujun* @version Id: ConsumerSimple.java, v 0.1 Administrator Exp $$* @date 2023-02-19 16:48:16*/
@Component
@RocketMQMessageListener(topic = "my-topic-obj", consumerGroup = "demo-consumer-group-obj")
public class ConsumerUserDefineSimple implements RocketMQListener<OrderExt> {/*** 根据泛型默认接收将接收到的json数据转化为对应的实体类** @param orderExt json对应实体*/@Overridepublic void onMessage(OrderExt orderExt) {System.out.println(orderExt);}
}

单元测试：

	/*** 测试发送异步延迟消息* * @throws InterruptedException* @throws RemotingException* @throws JsonProcessingException* @throws MQClientException*/@Testpublic void testSendAsyncMsgByJsonDelay() throws InterruptedException, RemotingException, JsonProcessingException, MQClientException {OrderExt orderExt = new OrderExt();orderExt.setId(UUID.randomUUID().toString());orderExt.setCreateTime(new Date());orderExt.setMoney(30000L);orderExt.setTitle("测试订单");this.producerUserDefineSimple.sendAsyncMsgByJsonDelay("my-topic-obj", orderExt);System.out.println("end……");Thread.sleep(20000);}

启动消费者，查看控制台，生产者发送几次，消费者只要监听到就会接收：

7、消费重试

7.1、什么是消费重试

当消息发送到Broker成功，在被消费者消费时如果消费者没有正常消费，此时消息会重试消费,消费重试存在两种场景：

1. 消息没有被消费者接收，比如消费者与broker存在网络异常。此种情况消息会一直被消费重试。
2. 当消息已经被消费者成功接收，但是在进行消息处理时出现异常，消费端无法向Broker返回成功，这种情况下RocketMQ会不断重试。

针对第二种消费重试的场景，borker是怎么知道重试呢？

消费者在消费消息成功会向broker返回成功状态，否则会不断进行消费重试。

7.2、处理策略

当消息在消费时出现异常，此时消息被不断重试消费。RocketMQ会一直重试消费吗？

答案是不会！

消息会按照延迟消息的延迟时间等级（1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h）从第3级开始重试，每试一次如果还不成功则延迟等级加1。

比如：一条消息消费失败，等待10s(第3级)进行重试，如果还没有被成功消费则延迟等级加1，即按第4级别延迟等待，等30s继续进行重试，如此进行下去，直到重试16次。

当重试了16次还未被成功消费将会投递到死信队列，到达死信队列的消息将不再被消费。

实际生产中的处理策略是什么呢？

实际生产中不会让消息重试这么多次，通常在重试一定的次数后将消息写入数据库，由另外单独的程序或人工去处理。

项目使用的Spring整合RocketMQ的方式，消费者实现RocketMQListener的onMessage方法，在此方法中实现处理策略的示例代码如下：

/*** 消费者监听类** @author yixiujun* @version Id: ConsumerSimple.java, v 0.1 Administrator Exp $$* @date 2023-02-19 16:48:16*/
@Component
@RocketMQMessageListener(topic = "my-topic-obj", consumerGroup = "demo-consumer-group-obj")
public class ConsumerUserDefineSimple implements RocketMQListener<MessageExt> {/*** 根据泛型默认接收将接收到的json数据转化为对应的实体类** @param messageExt 消息实体*/@Overridepublic void onMessage(MessageExt messageExt) {// 取出当前重试次数int reconsumeTimes = messageExt.getReconsumeTimes();// 当大于一定的次数后将消息写入数据库，由单独的程序或人工去处理if (reconsumeTimes >= 2) {// 将消息写入数据库，之后正常返回// ……return;}throw new RuntimeException(String.format("第%s次处理失败..",reconsumeTimes));}}

处理策略可根据实际业务场景需要自定义后续操作处理。

妥了，简单入门就先到这！

---

---