优秀html5网站,WordPress 公海池,免费网站建设软件大全,腾讯云阻止网站访问文章目录1. 什么是幂等性#xff1f;1.1 消息队列的幂等性1.2 模拟重试机制1.2.1 生产者代码1.2.2 消费者代码1.2.3 消费者 application.yml 配置2. 如何保证消息幂等性#xff0c;不被重复消费#xff1f;解决方法1. 什么是幂等性#xff1f; 在编程中一个幂等操作的特点是…
文章目录1. 什么是幂等性1.1 消息队列的幂等性1.2 模拟重试机制1.2.1 生产者代码1.2.2 消费者代码1.2.3 消费者 application.yml 配置2. 如何保证消息幂等性不被重复消费解决方法1. 什么是幂等性 在编程中一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同。 HTTP方法的幂等性是指一次和多次请求某一个资源应该具有同样的副作用。幂等性属于语义范畴正如编译器只能帮助检查语法错误一样HTTP规范也没有办法通过消息格式等语法手段来定义它。 简之一个请求不管重复来多少次结果是不会改变的。
1.1 消息队列的幂等性
如同HTTP方法的幂等性消息队列同样会出现幂等性问题。
消费者在消费 MQ 中的消息时MQ 已把消息发送给消费者消费者在给 MQ 返回 ack 时网络中断故 MQ 未收到确认信息该条消息会重新发给其他的消费者或者在网络重连后再次发送给该消费者但实际上该消费者已成功消费了该条消息造成消费者消费了重复的消息注意RabbitMQ 这种消息重试(补偿)机制是默认的。
所以MQ 消费者的幂等性问题主要在于 MQ 的重试机制因为网络原因或客户端延迟消费导致重复消费。
那么如何合适选择重试机制我们来看两种情况。
情况1 消费者获取到消息后调用第三方接口但接口暂时无法访问是否需要重试?
需要重试
情况2 消费者获取到消息后抛出数据转换异常是否需要重试?
不需要重试 总结对于情况2如果消费者代码抛出异常是需要发布新版本才能解决的问题那么不需要重试重试也无济于事。应该采用日志记录定时任务 job 健康检查人工进行补偿 1.2 模拟重试机制
我们采用一种短信消费者客户端异常的情况来模拟 RabbitMQ 的重试机制。
RabbitListener(queues fanout_sms_queue)
public void process(String msg) {System.out.println(短信消费者获取生产者消息msg: msg);int i 1/0;
}如上代码很显然会报错一担报错生产者的消息时不会被消费的 RabbitListener 底层使用 AOP 进行异常通知拦截如果程序没有抛出异常信息那么就会自动提交事务如果 AOP 异常通知拦截有捕获到异常信息的话就会自动实现重试(补偿)机制同时这个补偿机制的消息会缓存到 RabbitMQ 服务器端进行存放一直重试到不抛出异常为止。 1.2.1 生产者代码
Component
public class FanoutProducer {Autowiredprivate AmqpTemplate amqpTemplate;/*** 发送消息** param queueName 队列名称*/public void send(String queueName) {String msg my_fanout_msg: System.currentTimeMillis();Message message MessageBuilder.withBody(msg.getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding(utf-8).setMessageId(UUID.randomUUID() ).build();System.out.println(msg : msg);amqpTemplate.convertAndSend(queueName, message);}
}1.2.2 消费者代码
Component
public class FanoutEamilConsumer {RabbitListener(queues fanout_eamil_queue)public void process(Message message) throws Exception {String revMessage Thread.currentThread().getName() ,邮件消费者获取生产者消息msg: new String(message.getBody(), UTF-8) ,messageId: message.getMessageProperties().getMessageId();System.out.println(revMessage);}
}1.2.3 消费者 application.yml 配置
spring:rabbitmq:####连接地址host: 127.0.0.1####端口号 port: 5672####账号 username: guest####密码 password: guest### 地址virtual-host: /admin_hostlistener:simple:retry:####开启消费者重试enabled: true####最大重试次数max-attempts: 5####重试间隔次数initial-interval: 3000server:port: 8081我们通过 RabbitMQ 配置增加了 RabbitMQ 重试时间以及重试次数限制在一定程度上解决了重复消费的问题接下来看一道常问的面试题。 2. 如何保证消息幂等性不被重复消费
其实这个问题也算是 MQ 面试当中经常考察的一点因为无论是什么 MQ 都会有这个问题。
首先通过上边我们了解了什么是“幂等性”以及 MQ 幂等性问题的产生所以我们要清楚为什么会出现重复性消费在什么场景会出现重复消费
解决方法 使用全局 MessageID 判断消费方使用同一个解决幂等性问题。 或者使用业务逻辑保证唯一比如订单号码 生产者关键代码
Autowired
private AmqpTemplate amqpTemplate;/*** 发送消息** param queueName 队列名称*/
public void send(String queueName) {String msg my_fanout_msg: System.currentTimeMillis();Message message MessageBuilder.withBody(msg.getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding(utf-8).setMessageId(UUID.randomUUID() ).build();System.out.println(msg : msg);amqpTemplate.convertAndSend(queueName, message);
}如上生产者在发送消息时convertAndSend给消息对象设置了唯一的 MessageID只有该 MessageID 没有被消费者标记方能在重试机制中再次被消费。 消费者关键代码
RabbitListener(queues fanout_eamil_queue)
public void process(Message message) throws Exception {String revMessage Thread.currentThread().getName() ,邮件消费者获取生产者消息msg: new String(message.getBody(), UTF-8) ,messageId: message.getMessageProperties().getMessageId();System.out.println(revMessage);发送邮件的逻辑XXX
}如上通过 message.getMessageProperties().getMessageId() 获取 MessageID获取的 MessageID 可以用来判断是否已经被消费者消费过了如果已经消费则取消再次消费。 通常怎么判断呢
比如上方是一个邮件发送的消费者在做补偿时假如上一步邮件发送成功了我们会把该 ID 存至 redis中下次再调用时先去 redis 判断是否存在该 ID 了如果存在表明已经消费过了则直接返回不再消费否则消费然后将记录存至 redis。 我创建了一个java相关的公众号用来记录自己的学习之路感兴趣的小伙伴可以关注一下微信公众号哈niceyoo
