北京市门户网站百度高级搜索首页
RabbitMQ实战解决方案
RabbitMQ死信队列
死信队列产生的背景
RabbitMQ死信队列俗称，备胎队列；消息中间件因为某种原因拒收该消息后，可以转移到死信队列中存放，死信队列也可以有交换机和路由key等。
产生死信队列的原因
- 消息投递到MQ中存放 消息已经过期 消费者没有及时的获取到我们消息，消息如果存放到mq服务器中过期之后，会转移到备胎死信队列存放。
- 队列达到最大的长度 （队列容器已经满了）
- 消费者消费多次消息失败，就会转移存放到死信队列中
代码整合 参考 Boyatop-springboot-rabbitmq|#中order-dead-letter-queue项目
死信队列的架构原理
死信队列和普通队列区别不是很大
普通与死信队列都有自己独立的交换机和路由key、队列和消费者。
区别：
1.生产者投递消息先投递到我们普通交换机中，普通交换机在将该消息投到
普通队列中缓存起来，普通队列对应有自己独立普通消费者。
2.如果生产者投递消息到普通队列中，普通队列发现该消息一直没有被消费者消费
的情况下，在这时候会将该消息转移到死信（备胎）交换机中，死信（备胎）交换机
对应有自己独立的 死信（备胎）队列 对应独立死信（备胎）消费者。
死信队列应用场景
1.30分钟订单超时设计
- Redis过期key ：
- 死信延迟队列实现：
采用死信队列，创建一个普通队列没有对应的消费者消费消息，在30分钟过后
就会将该消息转移到死信备胎消费者实现消费。
备胎死信消费者会根据该订单号码查询是否已经支付过，如果没有支付的情况下
则会开始回滚库存操作。
RabbitMQ消息幂等问题
RabbitMQ消息自动重试机制
- 当我们消费者处理执行我们业务代码的时候，如果抛出异常的情况下
在这时候mq会自动触发重试机制，默认的情况下rabbitmq是无限次数的重试。
需要人为指定重试次数限制问题
- 在什么情况下消费者需要实现重试策略？
A.消费者获取消息后，调用第三方接口，但是调用第三方接口失败呢？是否需要重试？
该情况下需要实现重试策略，网络延迟只是暂时调用不通，重试多次有可能会调用通。
B.消费者获取消息后，因为代码问题抛出数据异常，是否需要重试？
该情况下是不需要实现重试策略，就算重试多次，最终还是失败的。
可以将日志存放起来，后期通过定时任务或者人工补偿形式。
如果是重试多次还是失败消息，需要重新发布消费者版本实现消费
可以使用死信队列
Mq在重试的过程中，有可能会引发消费者重复消费的问题。
Mq消费者需要解决 幂等性问题
幂等性 保证数据唯一
方式1：
生产者在投递消息的时候，生成一个全局唯一id，放在我们消息中。
Msg id=123456
Msg id=123456
Msg id=123456
消费者获取到我们该消息，可以根据该全局唯一id实现去重复。
全局唯一id 根据业务来定的 订单号码作为全局的id
实际上还是需要再db层面解决数据防重复。
业务逻辑是在做insert操作 使用唯一主键约束
业务逻辑是在做update操作 使用乐观锁
- 当消费者业务逻辑代码中，抛出异常自动实现重试 （默认是无数次重试）
- 应该对RabbitMQ重试次数实现限制，比如最多重试5次，每次间隔3s；重试多次还是失败的情况下，存放到死信队列或者存放到数据库表中记录后期人工补偿
如何合理选择消息重试
- 消费者获取消息后，调用第三方接口，但是调用第三方接口失败呢？是否需要重试 ？
- 消费者获取消息后，应该代码问题抛出数据异常，是否需要重试？
总结：如果消费者处理消息时，因为代码原因抛出异常是需要从新发布版本才能解决的，那么就不需要重试，重试也解决不了该问题的。存放到死信队列或者是数据库表记录、后期人工实现补偿。
Rabbitmq如何开启重试策略
| spring: |
消费者重试过程中，如何避免幂等性问题
重试的过程中，为了避免业务逻辑重复执行，建议提前全局id提前查询，如果存在
的情况下，就无需再继续做该流程。
重试的次数最好有一定间隔次数，在数据库底层层面保证数据唯一性，比如加上唯一id。
SpringBoot开启消息确认机制
配置文件新增
spring:rabbitmq:####连接地址host: 127.0.0.1####端口号port: 5672####账号username: guest####密码password: guest### 地址virtual-host: /booyaVirtualHostslistener:simple:retry:####开启消费者（程序出现异常的情况下会）进行重试enabled: true####最大重试次数max-attempts: 5####重试间隔次数initial-interval: 3000acknowledge-mode: manualdatasource:url: jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8username: rootpassword: rootdriver-class-name: com.mysql.jdbc.Driver |
消费者ack代码
@Slf4j@Component@RabbitListener(queues = "fanout_order_queue")public class FanoutOrderConsumer {@Autowiredprivate OrderManager orderManager;@Autowiredprivate OrderMapper orderMapper;@RabbitHandlerpublic void process(OrderEntity orderEntity, Message message, Channel channel) throws IOException {// try {log.info(">>orderEntity:{}<<", orderEntity.toString());String orderId = orderEntity.getOrderId();if (StringUtils.isEmpty(orderId)) {log.error(">>orderId is null<<");return;}OrderEntity dbOrderEntity = orderMapper.getOrder(orderId);if (dbOrderEntity != null) {log.info(">>该订单已经被消费过，无需重复消费!<<");// 无需继续重试channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);return;}int result = orderManager.addOrder(orderEntity);log.info(">>插入数据库中数据成功<<");if (result >= 0) {// 开启消息确认机制 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}// int i = 1 / 0;// } catch (Exception e) {// // 将失败的消息记录下来，后期采用人工补偿的形式// }}} |