rabbitmq04rabbitmq解决分布式事务

1.RabbitMQ消息重试机制

• 消费者在消费消息的时候&＃xff0c;如果消费者业务逻辑出现程序异常&＃xff0c;这时候应该如何处理&＃xff1f;
  答案&＃xff1a;使用消息重试机制。(springboot默认有消息重试机制)

1.1 如何合适选择重试机制

1. 消费者获取到消息后&＃xff0c;调用第三方接口&＃xff0c;但接口暂时无法访问&＃xff0c;是否需要重试? &＃xff08;需要重试机制&＃xff09;
2. 消费者获取到消息后&＃xff0c;抛出数据转换异常&＃xff0c;是否需要重试?&＃xff08;不需要重试机制&＃xff09;需要发布进行解决。

1.2 如何实现重试机制

对于情况2&＃xff0c;如果消费者代码抛出异常是需要发布新版本才能解决的问题&＃xff0c;那么不需要重试&＃xff0c;重试也无济于事。应该采用日志记录&＃43;定时任务job健康检查&＃43;人工进行补偿

2.解决幂等性

网络延迟传输中&＃xff0c;消费出现异常或者是消费延迟消费&＃xff0c;会造成MQ进行重试补偿&＃xff0c;在重试过程中&＃xff0c;可能会造成重复消费。

2.1 消费者如何保证消息幂等性&＃xff0c;不被重复消费

1. 使用全局MessageID判断消费方使用同一个&＃xff0c;解决幂等性。
2. 或者使用业务逻辑保证唯一&＃xff08;比如订单号码,保证一个订单号码只可能被插入一次数据库&＃xff09;

2.2 代码实现

2.2.1 生产者

• FanoutIdemProducer

2.2.2 消费者

• FanoutIdemEamilConsumer

2.2.3 配置文件

spring:rabbitmq:####连接地址host: 127.0.0.1####端口号port: 5672####账号username: guest####密码password: guest### 地址virtual-host: /springbootfanoutlistener:simple:retry:####开启消费者重试enabled: true####最大重试次数max-attempts: 5####重试间隔次数initial-interval: 3000


3.RabbitMQ签收模式配置

3.1 开启手动应答模式

spring:rabbitmq:listener:simple:####开启手动ackacknowledge-mode: manual


3.2 消费者

/*** 描述:签收模式演示* &＃64;author: myx* &＃64;date: 2019-05-01* Copyright © 2019-grape. All rights reserved.*/
&＃64;Component
public class AckEamilConsumer {/*** 消费者* &＃64;param message*/&＃64;RabbitListener(queues &＃61; FanoutConstrant.QUEUE_EMAIL_NAME)public void process(Message message, &＃64;Headers Map headers, Channel channel) throws IOException {System.out.println(Thread.currentThread().getName()&＃43; ",邮件消费者获取生产者消息msg:"&＃43; new String(message.getBody(), "UTF-8")&＃43; ",messageId:" &＃43; message.getMessageProperties().getMessageId());// 手动ackLong deliveryTag &＃61; (Long) headers.get(AmqpHeaders.DELIVERY_TAG);// 手动签收channel.basicAck(deliveryTag, false);}
}


4.RabbitMQ死信队列

4.1 什么是死信队列

死信队列是当消息在一个队列因为下列原因:

• 消息被拒绝&＃xff08;basic.reject或basic.nack&＃xff09;并且requeue&＃61;false.
• 消息TTL过期
• 队列达到最大长度&＃xff08;队列满了&＃xff0c;无法再添加数据到mq中&＃xff09;
  变成了 “死信” 后被重新投递&＃xff08;publish&＃xff09;到另一个Exchange,该Exchange 就是DLX然后该Exchange 根据绑定规则转发到对应的队列上监听该队列 就可以重新消费.说白了就是没有被消费的消息换个地方重新被消费

生产者 --> 消息 --> 交换机 --> 队列 --> 变成死信 --> DLX交换机 -->队列 --> 消费者


4.2 应用场景分析

在定义业务队列的时候&＃xff0c;可以考虑指定一个死信交换机&＃xff0c;并绑定一个死信队列&＃xff0c;当消息变成死信时&＃xff0c;该消息就会被发送到该死信队列上&＃xff0c;这样就方便我们查看消息失败的原因了

channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false); 丢弃消息

4.3 如何使用死信交换机

定义业务&＃xff08;普通&＃xff09;队列的时候指定参数

• x-dead-letter-exchange: 用来设置死信后发送的交换机
• x-dead-letter-routing-key&＃xff1a;用来设置死信的routingKey

4.4 代码实现

4.4.1 配置

• DeadFanoutConfig

4.4.2 邮件消费者

• DeadEamilConsumer

4.4.3 死信消费者

• DeadConsumer

5.RabbitMQ解决分布式事务

案例&＃xff1a;

经典案例&＃xff0c;以目前流行点外卖的案例&＃xff0c;用户下单后&＃xff0c;调用订单服务&＃xff0c;让后订单服务调用派单系统通知送外卖人员送单&＃xff0c;这时候订单系统与派单系统采用MQ异步通讯。

5.1 RabbitMQ解决分布式事务原理

采用最终一致性原理。
需要保证以下三要素:

1. 确认生产者一定要将数据投递到MQ服务器中&＃xff08;采用MQ消息确认机制&＃xff09;
2. MQ消费者消息能够正确消费消息&＃xff0c;采用手动ACK模式&＃xff08;注意重试幂等性问题&＃xff09;
3. 如何保证第一个事务先执行&＃xff0c;采用补偿机制&＃xff0c;在创建一个补单消费者进行监听&＃xff0c;如果订单没有创建成功&＃xff0c;进行补单。(如果第一个事务中出错&＃xff0c;补单消费者会在重新执行一次第一个事务&＃xff0c;例如第一个事务是添加订单表&＃xff0c;如果失败在补单的时候重新生成订单记录&＃xff0c;由于订单号唯一&＃xff0c;所以不会重复)

5.2 模拟代码实现

5.2.1 生产者

&＃64;Service
public class OrderService extends BaseApiService implements RabbitTemplate.ConfirmCallback {&＃64;Autowiredprivate OrderMapper orderMapper;&＃64;Autowiredprivate RabbitTemplate rabbitTemplate;public ResponseBase addOrderAndDispatch() {OrderEntity orderEntity &＃61; new OrderEntity();orderEntity.setName("订单名称");orderEntity.setOrderCreatetime(new Date());// 价格是300元orderEntity.setOrderMoney(300d);// 状态为 未支付orderEntity.setOrderState(0);Long commodityId &＃61; 30l;// 商品idorderEntity.setCommodityId(commodityId);String orderId &＃61; UUID.randomUUID().toString();orderEntity.setOrderId(orderId);// ##################################################// 1.先下单&＃xff0c;创建订单 (往订单数据库中插入一条数据)int orderResult &＃61; orderMapper.addOrder(orderEntity);System.out.println("orderResult:" &＃43; orderResult);if (orderResult <&＃61; 0) {return setResultError("下单失败!");}// 2.使用消息中间件将参数存在派单队列中send(orderId);return setResultSuccess();}private void send(String orderId) {JSONObject jsonObect &＃61; new JSONObject();jsonObect.put("orderId", orderId);String msg &＃61; jsonObect.toJSONString();System.out.println("msg:" &＃43; msg);// 封装消息Message message &＃61; MessageBuilder.withBody(msg.getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("utf-8").setMessageId(orderId).build();// 构建回调返回的数据CorrelationData correlationData &＃61; new CorrelationData(orderId);// 发送消息this.rabbitTemplate.setMandatory(true);this.rabbitTemplate.setConfirmCallback(this);rabbitTemplate.convertAndSend("order_exchange_name", "orderRoutingKey", message, correlationData);}// 生产消息确认机制&＃64;Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {String orderId &＃61; correlationData.getId();System.out.println("消息id:" &＃43; correlationData.getId());if (ack) {System.out.println("消息发送确认成功");} else {send(orderId);System.out.println("消息发送确认失败:" &＃43; cause);}}}


5.2.2 补单消费者

&＃64;Component
public class CreateOrderConsumer {&＃64;Autowiredprivate OrderMapper orderMapper;&＃64;RabbitListener(queues &＃61; "order_create_queue")public void process(Message message, &＃64;Headers Map headers, Channel channel) throws Exception {String messageId &＃61; message.getMessageProperties().getMessageId();String msg &＃61; new String(message.getBody(), "UTF-8");System.out.println("补单消费者" &＃43; msg &＃43; ",消息id:" &＃43; messageId);JSONObject jsonObject &＃61; JSONObject.parseObject(msg);String orderId &＃61; jsonObject.getString("orderId");// 判断订单是否存在&＃xff0c;如果不存在 实现自动补单机制OrderEntity orderEntityResult &＃61; orderMapper.findOrderId(orderId);if (orderEntityResult !&＃61; null) {System.out.println("订单已经存在 无需补单 orderId:" &＃43; orderId);return;}// 订单不存在 &＃xff0c;则需要进行补单OrderEntity orderEntity &＃61; new OrderEntity();orderEntity.setName("订单名称");orderEntity.setOrderCreatetime(new Date());// 价格是300元orderEntity.setOrderMoney(300d);// 状态为 未支付orderEntity.setOrderState(0);Long commodityId &＃61; 30l;// 商品idorderEntity.setCommodityId(commodityId);orderEntity.setOrderId(orderId);// ##################################################// 1.先下单&＃xff0c;创建订单 (往订单数据库中插入一条数据)try {int orderResult &＃61; orderMapper.addOrder(orderEntity);System.out.println("orderResult:" &＃43; orderResult);if (orderResult >&＃61; 0) {// 手动签收消息,通知mq服务器端删除该消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}} catch (Exception e) {// 丢弃该消息channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);}}
}


5.2.3 RabbitmqConfig

&＃64;Component
public class RabbitmqConfig {// 下单并且派单存队列public static final String ORDER_DIC_QUEUE &＃61; "order_dic_queue";// 补单队列&＃xff0c;判断订单是否已经被创建public static final String ORDER_CREATE_QUEUE &＃61; "order_create_queue";// 下单并且派单交换机private static final String ORDER_EXCHANGE_NAME &＃61; "order_exchange_name";// 1.定义订单队列&＃64;Beanpublic Queue directOrderDicQueue() {return new Queue(ORDER_DIC_QUEUE);}// 2.定义补订单队列&＃64;Beanpublic Queue directCreateOrderQueue() {return new Queue(ORDER_CREATE_QUEUE);}// 2.定义交换机&＃64;BeanDirectExchange directOrderExchange() {return new DirectExchange(ORDER_EXCHANGE_NAME);}// 3.订单队列与交换机绑定&＃64;BeanBinding bindingExchangeOrderDicQueue() {return BindingBuilder.bind(directOrderDicQueue()).to(directOrderExchange()).with("orderRoutingKey");}// 3.补单队列与交换机绑定&＃64;BeanBinding bindingExchangeCreateOrder() {return BindingBuilder.bind(directCreateOrderQueue()).to(directOrderExchange()).with("orderRoutingKey");}}


5.2.4 派单服务-消费者

&＃64;Component
public class DispatchConsumer {&＃64;Autowiredprivate DispatchMapper dispatchMapper;&＃64;RabbitListener(queues &＃61; "order_dic_queue")public void process(Message message, &＃64;Headers Map headers, Channel channel) throws Exception {String messageId &＃61; message.getMessageProperties().getMessageId();String msg &＃61; new String(message.getBody(), "UTF-8");System.out.println("派单服务平台" &＃43; msg &＃43; ",消息id:" &＃43; messageId);JSONObject jsonObject &＃61; JSONObject.parseObject(msg);String orderId &＃61; jsonObject.getString("orderId");if (StringUtils.isEmpty(orderId)) {// 日志记录return;}DispatchEntity dispatchEntity &＃61; new DispatchEntity();// 订单iddispatchEntity.setOrderId(orderId);// 外卖员iddispatchEntity.setTakeoutUserId(12l);// 外卖路线dispatchEntity.setDispatchRoute("40,40");try {int insertDistribute &＃61; dispatchMapper.insertDistribute(dispatchEntity);if (insertDistribute > 0) {// 手动签收消息,通知mq服务器端删除该消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}} catch (Exception e) {e.printStackTrace();// // 丢弃该消息channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);}}}