欢迎来到尧图网

客户服务 关于我们

您的位置:首页 > 教育 > 锐评 > RabbitMQ深度探索:消息幂等性问题

RabbitMQ深度探索:消息幂等性问题

2025/5/4 3:12:03 来源:https://blog.csdn.net/SOS_suyan/article/details/145439008  浏览:    关键词:RabbitMQ深度探索:消息幂等性问题
  1. RabbitMQ 消息自动重试机制:
    1. 让我们消费者处理我们业务代码的时候,如果抛出异常的情况下,在这时候 MQ 会自动触发重试机制,默认的情况下 RabbitMQ 时无限次数的重试
    2. 需要认为指定重试次数限制问题
  2. 在什么情况下消费者实现重试策略:
    1. 消费者调用第三方接口,但是调用第三方接口失败后,需要实现重试策略,网络延迟只是暂时调不通,重试多次有可能会调通
    2. 消费者获取代码后,因为代码问题抛出数据异常,此时不需要实现重试策略
      1. 我们需要将日志存放起来,后期通过定时任务或者人工补偿形式
      2. 如果是重试多次还是失败消息,需要重新发布消费者版本实现消费
      3. 可以使用死信队列
    3. MQ 在重试的过程中,可能会引发消费者重复消费的问题
    4. MQ 消费者需要解决幂等性问题
      1. 幂等性:保证数据唯一
  3. 解决幂等性问题:
    1. 生产者在投递消息的时候,生成一个唯一 id 放在我们消息中
    2. 消费者获取到该消息,可以根据全局唯一 id 实现去重
    3. 全局唯一 id 根据业务来定的,订单号码作为全局的 id 
    4. 实际上还是需要在 DB 层面解决数据防重复
    5. 业务逻辑是在做 insert 操作使用唯一主键约束
    6. 业务逻辑是在做 update 操作,使用乐观锁
      1. 当消费者业务逻辑代码中抛出异常自动实现重试(默认是无数次重试)
      2. 应该对 RabbitMQ 重试次数实现限制,比如最多重试 5 次,每次间隔 30 秒
      3. 重试多次还是失败的情况下,存放到死信队列或者存放到数据库表中记录后期人工补偿
  4. 如何选择消息重试:
    1. 消费者获取消息后,调用第三方接口,但是调用第三方接口失败后是否要重试?
    2. 消费者获取消息后,如果代码问题抛出数据异常,是否需要重试?
    3. 总结:
      1. 如果消费者处理消息时,因为代码原因抛出异常是需要重新发布版本才能解决,就不要重试
      2. 存放到死信队列或者是数据库记录、后期人工实现补偿
  5. 实现:
    1. yml 文件:
      spring:rabbitmq:####连接地址host: 127.0.0.1####端口号port: 5672####账号username: guest####密码password: guest### 地址virtual-host: boyatopVirtualHostlistener:simple:retry:#开启消费者进行重试(程序异常的情况)enabled: true#最大重试次数max-attempts: 5#重试间隔时间initial-interval: 3000#手动确认机制acknowledge-mode: manualdatasource:url: jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8username: rootpassword: rootdriver-class-name: com.mysql.jdbc.Driverboyatop:#备胎交换机dlx:exchange: boyatop_dlx_exchangequeue: boyatop_dlx_queueroutingKey: dlx#普通交换机order:exchange: boyatop_order_exchangequeue: boyatop_order_queueroutingKey: order
    2. 配置类:
      @Component
      public class IdempotentExchangeConfig {//交换机@Value("${boyatop.order.exchange}")private  String order_exchange;//普通队列@Value("${boyatop.order.queue}")private String order_queue;//普通队列的 key@Value("${boyatop.order.routingKey}")private String order_rotingKey;//死信交换机@Value("${boyatop.dlx.exchange}")private String dlx_exchange;//死信队列@Value("${boyatop.dlx.queue}")private String dlx_queue;//死信队列的 key@Value("${boyatop.dlx.routingKey}")private String dlx_routingKey;//定义死信交换机@Beanpublic DirectExchange dlxExchange(){return new DirectExchange(dlx_exchange);}//定义死信队列@Beanpublic Queue dlxQueue(){return new Queue(dlx_queue);}//定义普通交换机@Beanpublic DirectExchange orderExchange(){return new DirectExchange(order_exchange);}//定义普通队列@Beanpublic Queue orderQueue(){//订单队列绑定死信交换机Map<String,Object> arguments = new HashMap<>(2);arguments.put("x-dead-letter-exchange",dlx_exchange);arguments.put("x-dead-letter-routing-key",dlx_routingKey);return new Queue(order_queue,true,false,false,arguments);
      //        return QueueBuilder.durable(order_queue).withArguments(arguments).build();}//订单队列绑定交换机@Beanpublic Binding bindingOrderExchange(DirectExchange orderExchange, Queue orderQueue){return BindingBuilder.bind(orderQueue).to(orderExchange).with(order_rotingKey);}//死信队列绑定交换机@Beanpublic Binding bindingDlxExchange(DirectExchange dlxExchange, Queue dlxQueue){return BindingBuilder.bind(dlxQueue).to(dlxExchange).with(dlx_routingKey);}}
    3. 实体类:
      @Data
      @NoArgsConstructor
      public class OrderEntity implements Serializable {private Integer id;private String orderName;private String orderId;public OrderEntity(String orderName, String orderId) {this.orderName = orderName;this.orderId = orderId;}
      }
    4. Mapper:
      public interface OrderMapper {@Insert("INSERT into order_entity value (null,#{orderName},#{orderId})")int addOrder(OrderEntity orderEntity);@Select("select * from order_entity where order_id = #{orderId} ")OrderEntity getOrder(String orderId);
      }
    5. 生产者:
      @Component
      @Slf4j
      public class OrderProducer {@Autowiredprivate RabbitTemplate rabbitTemplate;@Value("${boyatop.order.exchange}")private  String order_exchange;//普通队列的 key@Value("${boyatop.order.routingKey}")private String order_rotingKey;public void sendMsg(String orderName,String orderId){OrderEntity orderEntity = new OrderEntity(orderName,orderId);rabbitTemplate.convertAndSend(order_exchange,order_rotingKey,orderEntity,message -> {message.getMessageProperties().setExpiration("5000");return message;});}
      }
    6. 消费者:
      @Component
      @Slf4j
      @RabbitListener(queues = "boyatop_order_queue")
      public class OrderConsumer {@Autowiredprivate OrderMapper orderMapper;@RabbitHandlerpublic void process(OrderEntity orderEntity, Message message, Channel channel){try{String orderId = orderEntity.getOrderId();if(StringUtils.isEmpty(orderId)){return;}OrderEntity dbOrderEntity = orderMapper.getOrder(orderId);if(dbOrderEntity != null){//出现异常,消息拒收,进入死信队列人为处理channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);}int result = orderMapper.addOrder(orderEntity);//出现异常int i = 1 / 0;channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);System.out.println("监听内容:" + orderEntity);}catch (Exception e){// 记录该消息日志形式  存放数据库db中、后期通过定时任务实现消息补偿、人工实现补偿//将该消息存放到死信队列中,单独写一个死信消费者实现消费。}}
      }

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com

热搜词