天津网站推广方法,上海建设工程咨询有限公司招聘,在国外做电商网站,服务器平台有哪些#x1f3af; 导读#xff1a;本文档详细介绍了RocketMQ中的重试机制与死信消息处理方法。对于生产者而言#xff0c;文档提供了如何配置重试次数的具体示例#xff1b;而对于消费者#xff0c;它解释了默认情况下消息消费失败后的重试策略#xff0c;并展示了如何通过代… 导读本文档详细介绍了RocketMQ中的重试机制与死信消息处理方法。对于生产者而言文档提供了如何配置重试次数的具体示例而对于消费者它解释了默认情况下消息消费失败后的重试策略并展示了如何通过代码自定义重试次数。当消息经过多次重试仍无法成功消费时RocketMQ会将其标记为死信消息并存入特定的死信队列中。文档还提供了处理死信队列的两种策略一种是编写专门的消费者来处理这些消息另一种是在达到一定重试次数后签收消息并通知人工干预。此外还包括了关于死信消息生产和消费的基本示例代码。 文章目录 RocketMQ 重试机制生产者重试消费者重试 RocketMQ 死信消息消息生产者消息消费者死信消费者控制台显示 RocketMQ 重试机制
生产者重试
// 失败的情况重发3次同步
producer.setRetryTimesWhenSendFailed(3);
// 失败的情况重发3次异步
producer.setRetryTimesWhenSendAsyncFailed(3);
// 消息在1S内没有发送成功就会重试
producer.send(msg, 1000);【示例代码】
Test
public void retryProducer() throws Exception {DefaultMQProducer producer new DefaultMQProducer(retry-producer-group);producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);producer.start();// 如果发送失败要重试几次同步不设置默认值是2producer.setRetryTimesWhenSendFailed(3);// 如果发送失败要重试几次异步
// producer.setRetryTimesWhenSendAsyncFailed(3);String key UUID.randomUUID().toString();System.out.println(key);Message message new Message(retryTopic, vip1, key, 我是vip666的文章.getBytes());producer.send(message);System.out.println(发送成功);producer.shutdown();
}消费者重试
如果消息消费失败默认会重试16次重试的时间间隔10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 能否自定义重试次数 可以重试的次数一般设置为5
// 消费失败重试几次
consumer.setMaxReconsumeTimes(5);如果重试了16次(并发模式是16次顺序模式下重试次数是 int 类型最大值) 都是失败的怎么处理 认为该消息是死信消息将消息放在一个死信主题中去名称:%DLQ%消费者组名最后再实现一个消费者去消费死信消息一般是发邮件发短信通知人工处理、做一些记录 死信队列只有一个队列 当消息处理失败的时候 该如何正确的处理? 方案一处理死信队列如果每个死信队列都写一个消费者很麻烦
/*** 方案一* 死信队列消费者* throws Exception*/
Test
public void retryDeadConsumer() throws Exception {DefaultMQPushConsumer consumer new DefaultMQPushConsumer(retry-dead-consumer-group);consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);consumer.subscribe(%DLQ%retry-consumer-group, *);consumer.registerMessageListener(new MessageListenerConcurrently() {Overridepublic ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs, ConsumeConcurrentlyContext context) {MessageExt messageExt msgs.get(0);System.out.println(new Date());System.out.println(new String(messageExt.getBody()));System.out.println(记录到特别的位置 文件 mysql 通知人工处理);// 业务报错了 返回null 返回 RECONSUME_LATER 都会重试return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();
}方案二在实际生产过程中一般重试3-5次如果还没有消费成功则可以把消息签收了通知人工等处理
/*** 方案二* 重试次数较多直接做日志记录、通知人工处理* throws Exception*/
Test
public void retryConsumer2() throws Exception {DefaultMQPushConsumer consumer new DefaultMQPushConsumer(retry-consumer-group);consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);consumer.subscribe(retryTopic, *);consumer.registerMessageListener(new MessageListenerConcurrently() {Overridepublic ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs, ConsumeConcurrentlyContext context) {MessageExt messageExt msgs.get(0);System.out.println(new Date());try {// 业务处理模拟报错handleDb();} catch (Exception e) {// 重试int reconsumeTimes messageExt.getReconsumeTimes();if (reconsumeTimes 3) {// 重试次数太大不要重试了System.out.println(记录到特别的位置 文件 mysql 通知人工处理);return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}return ConsumeConcurrentlyStatus.RECONSUME_LATER;}// 业务报错了 返回null 返回 RECONSUME_LATER 都会重试return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();
}private void handleDb() {int i 10 / 0;
}RocketMQ 死信消息
当消费重试到达阈值以后消息不会被投递给消费者了而是进入了死信队列
当一条消息初次消费失败 RocketMQ 会自动进行消息重试达到最大重试次数后若消费依然失败则表明消费者在正常情况下无法正确地消费该消息。此时该消息不会立刻被丢弃而是将其发送到该消费者对应的特殊队列中这类消息称为死信消息Dead-Letter Message存储死信消息的特殊队列称为死信队列Dead-Letter Queue死信队列是死信Topic下分区数唯一的单独队列。如果产生了死信消息对应的ConsumerGroup的死信Topic名称为%DLQ%ConsumerGroupName死信队列的消息将不会再被消费。 可以利用 RocketMQ Admin 工具或者 RocketMQ Dashboard 上查询到对应死信消息的信息。也可以监听死信队列进行自己的业务上的逻辑写日志、通知人工处理
消息生产者
Test
public void testDeadMsgProducer() throws Exception {DefaultMQProducer producer new DefaultMQProducer(dead-group);producer.setNamesrvAddr(localhost:9876);producer.start();Message message new Message(dead-topic, 我是一个死信消息.getBytes());producer.send(message);producer.shutdown();
}消息消费者
Test
public void testDeadMsgConsumer() throws Exception {DefaultMQPushConsumer consumer new DefaultMQPushConsumer(dead-group);consumer.setNamesrvAddr(localhost:9876);consumer.subscribe(dead-topic, *);// 设置最大消费重试次数 2 次consumer.setMaxReconsumeTimes(2);consumer.registerMessageListener(new MessageListenerConcurrently() {Overridepublic ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs, ConsumeConcurrentlyContext context) {System.out.println(msgs);// 测试消费失败return ConsumeConcurrentlyStatus.RECONSUME_LATER;}});consumer.start();System.in.read();
}死信消费者
注意权限问题 Test
public void testDeadMq() throws Exception{DefaultMQPushConsumer consumer new DefaultMQPushConsumer(dead-group);consumer.setNamesrvAddr(localhost:9876);// 消费重试到达阈值以后消息不会被投递给消费者了而是进入了死信队列// 队列名称 默认是 %DLQ% 消费者组名consumer.subscribe(%DLQ%dead-group, *);consumer.registerMessageListener(new MessageListenerConcurrently() {Overridepublic ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs, ConsumeConcurrentlyContext context) {System.out.println(msgs);// 处理消息 签收了return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();
}控制台显示
