RocketMQ快速入门

RocketMQ提供了发送多种发送消息的模式，例如同步消息，异步消息，顺序消息，延迟消息，事务消息等，我们一一学习

1.1 消息发送和监听的流程
我们先搞清楚消息发送和监听的流程，然后我们在开始敲代码

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

1.2 消息消费者

1.创建消费者consumer，制定消费者组名
2.指定Nameserver地址
3.创建监听订阅主题Topic和Tag等
4.处理消息
5.启动消费者consumer

下面是搭建的一个基础使用案例

创建一个sp工程，导入rocketmq依赖

<!-- 原生的api--><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.9.2</version>
<!--            docker的用下面这个版本<version>4.4.0</version>&ndash;&gt;--></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version></dependency>

编写生产者

/*** 发消息*/@Testvoid contextLoads() throws MQClientException, MQBrokerException, RemotingException, InterruptedException {// 创建一个生产者(定制一个生产者组名)DefaultMQProducer producer = new DefaultMQProducer("test-producer-test");// 连接nameServerproducer.setNamesrvAddr("127.0.0.1:9876");// 启动producer.start();// 创建一个消息Message message = new Message("testTopic", "我是一个简单的消息".getBytes());// 发送消息SendResult send = producer.send(message);System.out.println(send.getSendStatus());// 关闭生产者producer.shutdown();}

编写消费者：

@Testvoid simpleConsumer() throws MQClientException, IOException {// 创建一个消费者DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group");// 连接namesrvconsumer.setNamesrvAddr("127.0.0.1:9876");// 订阅一个主题 * 表示订阅这个主题中的所有消息，后期会有消息过滤consumer.subscribe("testTopic","*");// 设置一个监听器consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {// 这个就是消费的方法（业务处理）System.out.println("我是消费者");System.out.println("消息的内容是："+new String(msgs.get(0).getBody()));System.out.println("消费上下文:"+consumeConcurrentlyContext);// 返回值 CONSUME_SUCCESS成功，消息会从mq出队// RECONSUME_LATER (报错/null) 代表失败，消息会重新回到队列，过一会重新投递出来，给当前消费者或者其他消费者消费的return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 启动consumer.start();// 挂起当前jvm,就是让其不要停止，因为消费逻辑是异步执行的，这里是为了防止主线程执行完而异步线程还没执行完System.in.read();}

执行效果：

在这里插入图片描述
2. 消费模式

MQ的消费模式可以大致分为两种，一种是推Push，一种是拉Pull。
Push是服务端【MQ】主动推送消息给客户端，优点是及时性较好，但如果客户端没有做好流控，一旦服务端推送大量消息到客户端时，就会导致客户端消息堆积甚至崩溃。
Pull是客户端需要主动到服务端取数据，优点是客户端可以依据自己的消费能力进行消费，但拉取的频率也需要用户自己控制，拉取频繁容易造成服务端和客户端的压力，拉取间隔长又容易造成消费不及时。
Push模式也是基于pull模式的，只能客户端内部封装了api，一般场景下，上游消息生产量小或者均速的时候，选择push模式。在特殊场景下，例如电商大促，抢优惠券等场景可以选择pull模式，rocketmq拉取消息的规则大致如下所示：

在这里插入图片描述

在这里插入图片描述

3. RocketMQ发送同步消息
上面的快速入门就是发送同步消息，发送过后会有一个返回值，也就是mq服务器接收到消息后返回的一个确认，这种方式非常安全，但是性能上并没有这么高，而且在mq集群中，也是要等到所有的从机都复制了消息以后才会返回，所以针对重要的消息可以选择这种方式

在这里插入图片描述
4. RocketMQ发送异步消息
异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。发送完以后会有一个异步消息通知

代码如下：

    @Testpublic void asyncProducer() throws MQClientException, RemotingException, InterruptedException, IOException {DefaultMQProducer producer = new DefaultMQProducer();producer.setNamesrvAddr("127.0.0.1:9876");producer.start();Message message = new Message("asyncTopic", "我是一个异步消费者".getBytes());producer.send(message, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.println("发送成功");}@Overridepublic void onException(Throwable throwable) {System.out.println("发送失败:"+throwable.getMessage().toString());}});System.out.println("主线程先执行");System.in.read();}

消费者端代码不变：

 @Testvoid simpleConsumer() throws MQClientException, IOException {// 创建一个消费者DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group");// 连接namesrvconsumer.setNamesrvAddr("127.0.0.1:9876");// 订阅一个主题 * 表示订阅这个主题中的所有消息，后期会有消息过滤consumer.subscribe("testTopic","*");// 设置一个监听器consumer.registerMessageListener(new MessageListenerConcurrently() {// // 注册一个消费监听 MessageListenerConcurrently是并发消费    _//_// 默认是20个线程一起消费，可以参看 consumer.setConsumeThreadMax()@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {// 这个就是消费的方法（业务处理） 这里执行消费的代码，默认是多线程执行System.out.println("我是消费者");System.out.println("消息的内容是："+new String(msgs.get(0).getBody()));System.out.println("消费上下文:"+consumeConcurrentlyContext);// 返回值 CONSUME_SUCCESS成功，消息会从mq出队// RECONSUME_LATER (报错/null) 代表失败，消息会重新回到队列，过一会重新投递出来，给当前消费者或者其他消费者消费的return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 启动consumer.start();// 挂起当前jvm,就是让其不要停止，因为消费逻辑是异步执行的，这里是为了防止主线程执行完而异步线程还没执行完System.in.read();}

5.RocketMQ发送单向消息
这种方式主要用在不关心发送结果的场景，这种方式吞吐量很大，但是存在消息丢失的风险，例如日志信息的发送

单向生产者代码：

/*** 单向消息发送，不需要在乎消息是否发送成功，*  这种模式下生产者只需要负责发送数据就行，不需要关注*  消息有没有发送成功，例如：日志处理* @throws Exception*/@Testpublic void  OneWayProducer() throws Exception{DefaultMQProducer producer = new DefaultMQProducer("oneway-producer-group");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();Message message = new Message("oneWayTopic", "日志xxx".getBytes());producer.sendOneway(message);System.out.println("成功");producer.shutdown();}

消费者代码和上面一样

6. RocketMQ发送延迟消息
消息放入mq后，过一段时间，才会被监听到，然后消费
比如下订单业务，提交了一个订单就可以发送一个延时消息，30min后去检查这个订单的状态，如果还是未付款就取消订单释放库存。

生产者代码：

/*** 延迟消息* @throws Exception*/@Testpublic void msproducer() throws Exception{DefaultMQProducer producer = new DefaultMQProducer("ms-producer-group");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();Message message = new Message("orderTopicTest", "订单号，座位号".getBytes());// 为消息设置过期时间  1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h// 消息的不同level对应不同的时间，因此这里的3对应着10smessage.setDelayTimeLevel(3);producer.send(message);System.out.println("发送时间:"+new Date());producer.shutdown();}

消费者端代码：

    @Testpublic void  msConsumer() throws Exception{DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ms-consumer-group");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.subscribe("orderTopicTest","*");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {System.out.println("收到了消息"+new Date());System.out.println(new String(list.get(0).getBody()));return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();}

7.RocketMQ发送批量消息

Rocketmq可以一次性发送一组消息，那么这一组消息会被当做一个消息消费

批量消息生产者：

/*** 批量发送消息*/@Testpublic void testBatchProducer() throws Exception{DefaultMQProducer producer = new DefaultMQProducer("batch-producer-group");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();List<Message> msgs = Arrays.asList(new Message("batchTopic","我是一组消息的A消息".getBytes()),new Message("batchTopic","我是一组消息的B消息".getBytes()),new Message("batchTopic","我是一组消息的C消息".getBytes()));SendResult send = producer.send(msgs);System.out.println(send.getSendStatus().toString());// 关闭实例producer.shutdown();}

消费者端代码：

@Testpublic void  msConsumer() throws Exception{DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("batch-consumer-group");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.subscribe("batchTopic","*");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {System.out.println("收到了消息"+new Date());System.out.println(new String(list.get(0).getBody()));return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();}

8.RocketMQ发送顺序消息

消息有序指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ可以严格的保证消息有序，可以分为：分区有序或者全局有序。
rocketMq的broker的机制，导致了rocketMq会有这个问题
因为一个broker中对应了四个queue

在这里插入图片描述

顺序消费的原理解析，在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列)；而消费消息的时候从多个queue上拉取消息，这种情况发送和消费是不能保证顺序。但是如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的。

下面用订单进行分区有序的示例。一个订单的顺序流程是：下订单、发短信通知、物流、签收。订单顺序号相同的消息会被先后发送到同一个队列中，消费时，同一个顺序获取到的肯定是同一个队列。

8.1 创建一个消息模型

@Data
@AllArgsConstructor
@NoArgsConstructor
public class MsgModel {private String orderSn;private Integer userId;private String desc;
}

8.2 顺序消息生产者

private List<MsgModel> msgModels = Arrays.asList(new MsgModel("qwer",1,"下单"),new MsgModel("qwer",1,"短信"),new MsgModel("qwer",1,"物流"),new MsgModel("zxcv",1,"下单"),new MsgModel("zxcv",1,"短信"),new MsgModel("zxcv",1,"物流"));/*** 顺序发送，这里是让相同订单号的消息进入到同一个队列* @throws Exception*/@Testpublic void orderlyProducer() throws Exception{DefaultMQProducer producer = new DefaultMQProducer("orderly-producer-group");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();// 发送顺序消息，发送时要确保有序，并且要发送到同一个队列里去msgModels.forEach(msgModel -> {Message message = new Message("orderlyTopic", msgModel.toString().getBytes());try {producer.send(message, new MessageQueueSelector() {@Overridepublic MessageQueue select(List<MessageQueue> list, Message message, Object o) {// 在这里选择队列，让相同订单号进入到同一个队列int hashCode = o.toString().hashCode();// 2 % 4 = 2// 3 % 4 = 3   周期性函数，这里得到的结果永远比模数小，保证了不会被越界int i = hashCode % list.size();return list.get(i);}},msgModel.getOrderSn());} catch (Exception e) {throw new RuntimeException(e);}});producer.shutdown();System.out.println("发送完成");}

8.3 顺序消费消息

/*** 顺序消费* @throws Exception*/@Testpublic void  orderlyConsumer() throws Exception{DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderly-producer-group");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.subscribe("orderlyTopic","*");// MessageListenerConcurrently 代表并发模式,多线程,默认20个线程  // 重试16次//consumer.setConsumeThreadMax();// 这种模式下可以设置最大线程数// MessageListenerOrderly 顺序模式，单线程 无线重试Integer.Max_Valueconsumer.registerMessageListener(new MessageListenerOrderly() {@Overridepublic ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {System.out.println("线程id:"+Thread.currentThread().getId());System.out.println(new String(list.get(0).getBody()));return ConsumeOrderlyStatus.SUCCESS;}});consumer.start();System.in.read();}

9.RocketMQ发送带标签的消息，消息过滤
Rocketmq提供消息过滤功能，通过tag或者key进行区分
我们往一个主题里面发送消息的时候，根据业务逻辑，可能需要区分，比如带有tagA标签的被A消费，带有tagB标签的被B消费，还有在事务监听的类里面，只要是事务消息都要走同一个监听，我们也需要通过过滤才区别对待：

在这里插入图片描述

在这里插入图片描述
生产者代码：

@Testpublic void tagProducer() throws Exception{DefaultMQProducer producer = new DefaultMQProducer("tag-producer-group");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();Message message = new Message("tagTopic", "vip1", "我是vip1的文章".getBytes());Message message2 = new Message("tagTopic", "vip2", "我是vip2的文章".getBytes());producer.send(message);producer.send(message2);System.out.println("发送成功");producer.shutdown();}

消费者代码：

/*** 订阅关系一致性* @throws Exception*/@Testpublic void tagConsumer1() throws Exception{DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-group-a");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.subscribe("tagTopic","vip1");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {System.out.println("我是vip1的消费者，我正在消费消息"+new String(list.get(0).getBody().toString()));return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();}@Testpublic void tagConsumer2() throws Exception{DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-group-b");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.subscribe("tagTopic","vip1 || vip2");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {System.out.println("我是vip2的消费者，我正在消费消息"+new String(list.get(0).getBody().toString()));return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();}

9.1什么时候该用 Topic，什么时候该用 Tag？

总结：不同的业务应该使用不同的Topic如果是相同的业务里面有不同表的表现形式，那么我们要使用tag进行区分
可以从以下几个方面进行判断：
1.消息类型是否一致：如普通消息、事务消息、定时（延时）消息、顺序消息，不同的消息类型使用不同的 Topic，无法通过 Tag 进行区分。
2.业务是否相关联：没有直接关联的消息，如淘宝交易消息，京东物流消息使用不同的 Topic 进行区分；而同样是天猫交易消息，电器类订单、女装类订单、化妆品类订单的消息可以用 Tag 进行区分。
3.消息优先级是否一致：如同样是物流消息，盒马必须小时内送达，天猫超市 24 小时内送达，淘宝物流则相对会慢一些，不同优先级的消息用不同的 Topic 进行区分。
4.消息量级是否相当：有些业务消息虽然量小但是实时性要求高，如果跟某些万亿量级的消息使用同一个 Topic，则有可能会因为过长的等待时间而“饿死”，此时需要将不同量级的消息进行拆分，使用不同的 Topic。
总的来说，针对消息分类，您可以选择创建多个 Topic，或者在同一个 Topic 下创建多个 Tag。但通常情况下，不同的 Topic 之间的消息没有必然的联系，而 Tag 则用来区分同一个 Topic 下相互关联的消息，例如全集和子集的关系、流程先后的关系。

10.RocketMQ中消息的Key

在rocketmq中的消息，默认会有一个messageId当做消息的唯一标识，我们也可以给消息携带一个key，用作唯一标识或者业务标识，包括在控制面板查询的时候也可以使用messageId或者key来进行查询

在这里插入图片描述

带key的消息生产者代码：

/*** 业务参数，我们自身要确保唯一性* 为了查阅和去重* @throws Exception*/@Testpublic void keyProducer() throws Exception{DefaultMQProducer producer = new DefaultMQProducer("key-producer-group");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();String key = UUID.randomUUID().toString();System.out.println(key);Message message = new Message("keyTopic","vip1",key,"我是vip1的文章".getBytes());producer.send(message);producer.shutdown();}

在这里插入图片描述

消费者端代码：

@Testpublic void keyConsumer() throws Exception{DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("key-consumer-group");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.subscribe("keyTopic","*");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {MessageExt messageExt = list.get(0);System.out.println("我是vip1,我正在消费消息"+list.get(0).getBody());System.out.println("我们的业务唯一标识是:"+messageExt.getKeys());return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});}

11.Rocketmq重复消费问题

BROADCASTING(广播)模式下，所有注册的消费者都会消费，而这些消费者通常是集群部署的一个个微服务，这样就会多台机器重复消费，当然这个是根据需要来选择。
CLUSTERING（负载均衡）模式下，如果一个topic被多个consumerGroup消费，也会重复消费。即使是在CLUSTERING模式``下，同一个consumerGroup下，一个队列只会分配给一个消费者，看起来好像是不会重复消费。但是，有个特殊情况：一个消费者新上线后，同组的所有消费者要重新负载均衡（反之一个消费者掉线后，也一样）。一个队列所对应的新的消费者要获取之前消费的offset（偏移量，也就是消息消费的点位），此时之前的消费者可能已经消费了一条消息，但是并没有把offset提交给broker，那么新的消费者可能会重新消费一次。虽然orderly模式是前一个消费者先解锁，后一个消费者加锁再消费的模式，比起concurrently要严格了，但是加锁的线程和提交offset的线程不是同一个，所以还是会出现极端情况下的重复消费。
还有在发送批量消息的时候，会被当做一条消息进行处理，那么如果批量消息中有一条业务处理成功，其他失败了，还是会被重新消费一次。
那么如果在CLUSTERING（负载均衡）模式下，并且在同一个消费者组中，不希望一条消息被重复消费，改怎么办呢？我们可以想到去重操作，找到消息唯一的标识，可以是msgId也可以是你自定义的唯一的key，这样就可以去重了

在这里插入图片描述

生产者代码：伪造重复消息：

@Testpublic void rePeatProducer() throws Exception{DefaultMQProducer producer = new DefaultMQProducer("repeat-producer-group");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();java.lang.String key = UUID.randomUUID().toString();Message message = new Message("tagTopic", null, key,"扣减库存-1".getBytes());Message message2 = new Message("tagTopic",null, key,"扣减库存-1".getBytes());// 测试发送两个一样的key的消息producer.send(message);producer.send(message2);System.out.println("发送成功");producer.shutdown();}

消费者端消费幂等性处理：

/*** 我们设计一个去重表，对消息的唯一key添加唯一索引* 每次消费消息的时候，先插入数据库，如果成功则执行业务逻辑* 如果插入失败，则说明消息来过了，直接签收了* @throws Exception*/@Testpublic void rePeatConsumer() throws Exception{DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("repeat-consumer-group");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.subscribe("keyTopic","*");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {MessageExt messageExt = list.get(0);java.lang.String keys = messageExt.getKeys();// 插入数据库，因为我们key做了唯一索引
//                int i = jdbcTemplate.update("insert into order_per_log(`type`,`order_dn`,`user`) values(1,keys,'123')");// 新增，要么成功，要么报错，修改要么成功，要么返回0，要么报错// 如果成功，就处理业务逻辑,如果报错，就不执行System.out.println(new String(messageExt.getBody()));return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();}

12.RocketMQ重试机制

一般我们的消息在投递过程中出现问题时，比如我们投递到队列失败了，这时候我们可以引入一个重试机制来进行重新投递

/*** 生产者发送消息失败进行重试* @throws Exception*/@Testpublic void retryProducer() throws Exception{DefaultMQProducer producer = new DefaultMQProducer("retry-producer-group");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();// 生产者发送消息 重试次数producer.setRetryTimesWhenSendFailed(2);
//            producer.setRetryTimesWhenSendAsyncFailed(2);String key = UUID.randomUUID().toString();System.out.println(key);Message message = new Message("retryTopic","vip1",key,"我是vip1的文章".getBytes());producer.send(message);producer.shutdown();}

在消费者端进行消费时，也可以进行重试：

/*** 重试的时间间隔* 10s 30s 1m,2m,3m,4m,5m,6m,7m 8m 9m 10m 20m 30m 1h 2h* 默认重试16次** 1. 能否自定义重试次数* 2. 如果重试16次都是失败的?*     (并发模式下可以重试16次，顺序模式性下可以重试int的最大值次)，如果都失败了，就认为是一个死信消息*      则会放置在一个死信主题中，主题的名称：%DLQ%retry-consumer-group，然后在死信队列中进行处理，找人工，电话啥的* 3. 当消息失败时该如何处理** 重试次数一般五次* @throws Exception*/@Testpublic void retryConsumer() throws Exception{DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-consumer-group");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.subscribe("retryTopic","*");// 设置重试次数consumer.setMaxReconsumeTimes(2);consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {MessageExt messageExt = list.get(0);System.out.println(new Date());System.out.println(messageExt.getReconsumeTimes());// 业务报错，返回 返回RECONSUME_LATER都会重试return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});}/*** 监听死信队列* @throws Exception*/@Testpublic void retryDeadConsumer() throws Exception{DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-dead-consumer-group");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.subscribe("%DLQ%retry-consumer-group","*");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {MessageExt messageExt = list.get(0);System.out.println(new Date());System.out.println("记录到特别的位置，文件 mysql 通知人工处理");// 业务报错，返回 返回RECONSUME_LATER都会重试return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});}/*** 第二种方案 用法比较多* @throws Exception*/@Testpublic void retryConsumer2() throws Exception{DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-consumer-group");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.subscribe("retryTopic","*");// 设置重试次数consumer.setMaxReconsumeTimes(2);consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {MessageExt messageExt = list.get(0);System.out.println(new Date());// 业务处理try {handDB();}catch (Exception e){// 重试int reconsumeTimes = messageExt.getReconsumeTimes();if (reconsumeTimes >= 3){// 不要重试了System.out.println("记录到特别的位置，文件，mysql,通知人工处理");return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}return ConsumeConcurrentlyStatus.RECONSUME_LATER;}System.out.println(messageExt.getReconsumeTimes());// 业务报错，返回 返回RECONSUME_LATER都会重试return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});}// 加上事务控制业务，报错回滚private void handDB() {int i = 10 / 0;}