RabbitMq | 解决分布式事务(最终一致性原理)
·
rabbitMQ解决分布式事务原理: 采用最终一致性原理。
需要保证以下三要素
1、确认生产者一定要将数据投递到MQ服务器中(采用MQ消息确认机制)
2、MQ消费者消息能够正确消费消息,采用手动ACK模式(注意重试幂等性问题)
3、如何保证第一个事务先执行,采用补偿机制,在创建一个补单消费者进行监听,如果订单没有创建成功,进行补单。
外卖点餐项目中,订单服务中产生订单,生成订单号。
此时需要调用派单服务送餐。两个服务之见不能使用http协议传输,我们需要采用mq实现异步派单系统
前提:如果生产者投递消息到MQ服务器成功:
场景一: 如果消费者从消息队列中获取消息时,失败了,生产者是不需要回滚事务,那么如何解决呢?
答:消费者采用手动ACK,进行补偿机制(重试3次),注意幂等性问题(全局ID)
================================================================
场景二: 如何确保生产者投递消息到MQ服务器一定是成功?
答:confirm机制(确认应答机制)
如果失败了,如何解决
答:使用生产者重试机制进行发消息
1)配置文件开启confirm机制(确认应答机制)
生产者:
@Service
public class OrderService extends BaseApiService implements RabbitTemplate.ConfirmCallback {
@Autowired
private OrderMapper orderMapper;
@Autowired
private RabbitTemplate rabbitTemplate;
public ResponseBase addOrderAndDispatch() {
OrderEntity orderEntity = new OrderEntity();
orderEntity.setName("永久会员充值");
orderEntity.setOrderCreatetime(new Date());
// 价格是300元
orderEntity.setOrderMoney(300d);
// 状态为 未支付
orderEntity.setOrderState(0);
Long commodityId = 30l;
// 商品id
orderEntity.setCommodityId(commodityId);
String orderId = UUID.randomUUID().toString();
orderEntity.setOrderId(orderId);
// ##################################################
// 1.先下单,创建订单 (往订单数据库中插入一条数据)
int orderResult = orderMapper.addOrder(orderEntity);
System.out.println("orderResult:" + orderResult);
if (orderResult <= 0) {
return setResultError("下单失败!");
}
// 2.使用消息中间件将参数存在派单队列中
send(orderId);
return setResultSuccess();
}
private void send(String orderId) {
JSONObject jsonObect = new JSONObject();
jsonObect.put("orderId", orderId);
String msg = jsonObect.toJSONString();
System.out.println("msg:" + msg);
// 封装消息
Message message = MessageBuilder.withBody(msg.getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON)
.setContentEncoding("utf-8").setMessageId(orderId).build();
// 构建回调返回的数据
CorrelationData correlationData = new CorrelationData(orderId);
// 发送消息
this.rabbitTemplate.setMandatory(true);
this.rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.convertAndSend("order_exchange_name", "orderRoutingKey", message, correlationData);
}
// 生产消息确认机制
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
String orderId = correlationData.getId();
System.out.println("消息id:" + correlationData.getId());
if (ack) {
System.out.println("消息发送确认成功");
} else {
send(orderId);
System.out.println("消息发送确认失败:" + cause);
}
}
}
3、如何保证第一个事务先执行,生产者投递消息到队列成功,消费者成功从队列中取出消息,但是订单事务出现错误,发生回滚,出现了数据不一致,即派单表里有数据,订单表里因为回滚没有数据。

答:订单服务产生消息时,通过路由键,分别给派单队列和补单队列发送ID,补单队列有ID,查询订单是否创建,如果每创建,就去创建
@Component
public class CreateOrderConsumer {
@Autowired
private OrderMapper orderMapper;
@RabbitListener(queues = "order_create_queue")
public void process(Message message, @Headers Map<String, Object> headers, Channel channel) throws Exception {
String messageId = message.getMessageProperties().getMessageId();
String msg = new String(message.getBody(), "UTF-8");
System.out.println("补单消费者" + msg + ",消息id:" + messageId);
JSONObject jsonObject = JSONObject.parseObject(msg);
String orderId = jsonObject.getString("orderId");
// 判断订单是否存在,如果不存在 实现自动补单机制
OrderEntity orderEntityResult = orderMapper.findOrderId(orderId);
if (orderEntityResult != null) {
System.out.println("订单已经存在 无需补单 orderId:" + orderId);
return;
}
// 订单不存在 ,则需要进行补单
OrderEntity orderEntity = new OrderEntity();
orderEntity.setName("蚂蚁课堂永久会员充值");
orderEntity.setOrderCreatetime(new Date());
// 价格是300元
orderEntity.setOrderMoney(300d);
// 状态为 未支付
orderEntity.setOrderState(0);
Long commodityId = 30l;
// 商品id
orderEntity.setCommodityId(commodityId);
orderEntity.setOrderId(orderId);
// ##################################################
// 1.先下单,创建订单 (往订单数据库中插入一条数据)
try {
int orderResult = orderMapper.addOrder(orderEntity);
System.out.println("orderResult:" + orderResult);
if (orderResult >= 0) {
// 手动签收消息,通知mq服务器端删除该消息
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
} catch (Exception e) {
// 丢弃该消息
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
}
}
}
更多推荐

所有评论(0)