数据库回滚（消息事务）

1、可靠消息最终一致性

可靠消息最终一致性方案是指当事务发起方执行完成本地事务后并发出一条消息，事务参与方(消息消费者)一定能够接收消息并处理事务成功，此方案强调的是只要消息发给事务参与方最终事务要达到一致。

此方案是利用消息中间件完成，如下图：

事务发起方（消息生产方）将消息发给消息中间件，事务参与方从消息中间件接收消息，事务发起方和消息中间件 之间，事务参与方（消息消费方）和消息中间件之间都是通过网络通信，由于网络通信的不确定性会导致分布式事务问题。因此可靠消息最终一致性方案要解决以下几个问题：

1.1、本地事务与消息发送的原子性问题

本地事务与消息发送的原子性问题即：事务发起方在本地事务执行成功后消息必须发出去，否则就丢弃消息。即实现本地事务和消息发送的原子性，要么都成功，要么都失败。本地事务与消息发送的原子性问题是实现可靠消息最终一致性方案的关键问题。

先来尝试下这种操作，先发送消息，再操作数据库：

begin transaction；

//1.发送MQ

//2.数据库操作

commit transation;

这种情况下无法保证数据库操作与发送消息的一致性，因为可能发送消息成功，数据库操作失败。

第二种方案，先进行数据库操作，再发送消息：

begin transaction；

//1.数据库操作

//2.发送MQ

commit transation;

这种情况下貌似没有问题，如果发送MQ消息失败，就会抛出异常，导致数据库事务回滚。但如果是超时异常，数据库回滚，但MQ其实已经正常发送了，同样会导致不一致。

1.2、事务参与方接收消息的可靠性

事务参与方必须能够从消息队列接收到消息，如果接收消息失败可以重复接收消息。

1.3、消息重复消费的问题

由于网络延时的存在，若某一个消费节点超时但是消费成功，此时消息中间件会重复投递此消息，就导致了消息的重 复消费。要解决消息重复消费的问题就要实现事务参与方的方法幂等性。

1.4、可靠消息最终一致性流程图

2、解决方案，本地事务+本地消息表+RabbitMQ手动ACK机制+定时器

rabbitmq生产者确认机制有很严重的性能问题，如果每秒钟只有几百的消息量，可以使用。所以，我们主要讲了消费者的消息确认机制。

环境：

springboot:2.1.9.RELEASE

springcloud:Greenwich.SR3

mybatisplus:3.3.0

2.1、消息生产者ACK机制：

/** * 定制化amqp模版 * 此处为模版类定义 Jackson消息转换器 * ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调 即消息发送到exchange ack * ReturnCallback接口用于实现消息发送到RabbitMQ 交换器，但无相应队列与交换器绑定时的回调 即消息发送不到任何一个队列中 ack * * @return the amqp template */ @Primary @Bean public AmqpTemplate amqpTemplate() { rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter()); rabbitTemplate.setEncoding("UTF-8"); // 消息发送失败返回到队列中，配置 publisher-returns: true rabbitTemplate.setMandatory(true); //消息发送到队列失败回调 rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> { String messageId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation"); //更新消息状态为失败 updateMessage(Long.valueOf(messageId),CommonConstant.SEND_FAIL); }); // 消息确认，配置 publisher-confirms: true //消息发送到channel回调setConfirmCallback,成功ack为true,失败ack为false rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { if (ack) { if (log.isDebugEnabled()) { log.debug("消息发送到exchange成功,correlationData: {}", correlationData); } if (correlationData != null) { String messageId= correlationData.getId(); //更新消息状态为成功 updateMessage(Long.valueOf(messageId),CommonConstant.SEND_SUCCESS); } } else { log.info("消息发送失败：{} 数据：{}", cause, JSONObject.toJSONString(correlationData)); if (correlationData != null) { String messageId= correlationData.getId(); //更新消息状态为失败 updateMessage(Long.valueOf(messageId),CommonConstant.SEND_FAIL); } } }); return rabbitTemplate; }

定时任务定时发送投递失败的消息

/** *定时器定时轮询发送失败的消息重新发送 */ @Scheduled(cron = "*/6 * * * * ?") public void timer() { QueryWrapper<Message> wrapper = new QueryWrapper<>(); wrapper.eq("message_status", CommonConstant.SEND_FAIL); List<Message> messages = messageDAO.selectList(wrapper); if (!CollectionUtils.isEmpty(messages)) { messages.stream().forEach((item) -> { String messageEntity = item.getMessageEntity(); Integer messageId = item.getId(); JSONObject messageJson = JSONObject.parseObject(messageEntity); sendMessage(messageJson, messageId.toString()); }); } }

交换机，队列

/** * 声明topic交换机 * * @return 交换机 */ @Bean("naocs-order-service.order.syn.stock.exchange") public Exchange createExchange() { DirectExchange directExchange = new DirectExchange(TopicEnum.ORDER_SYN_STOCK.getExchangeName(), true, false); return directExchange; } /** * 声明一个队列 * * @return 持久化队列 */ @Bean("naocs-order-service.order.syn.stock.queue") public Queue createTopicQueue() { // 参数1 name ：队列名 // 参数2 durable ：是否持久化 // 参数3 exclusive ：仅创建者可以使用的私有队列，断开后自动删除 // 参数4 autoDelete : 当所有消费客户端连接断开后，是否自动删除队列 Queue queue = new Queue(TopicEnum.ORDER_syn_STOCK.getQueueName(), true, false, false); return queue; }

消息持久化

/** * 发送消息 * @param messageJson * @param messaegId */ public void sendMessage(JSONObject messageJson, String messaegId) { //如果网络超时发送到channel,或队列网络超时，可能多次投递消息，而网络过会可能已经修复，这样就相当于重复投递 //查询库存消息表是否存在，如果存在，所以消息已经发送出去，无需重复发送 Boolean flage = nacosStockFeignClient.exit(Long.valueOf(messaegId)); if (!flage) { //发送消息 rabbitTemplate.convertAndSend(TopicEnum.ORDER_SYN_STOCK.getExchangeName(), TopicEnum.ORDER_SYN_STOCK.getRoutingKey(), messageJson, message -> { //消息持久化 message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT); return message; }, new CorrelationData(messaegId)); } }

3、事务发起者

OrderService

注意：@Transactional事务注解必须有

事务发起者消息表保证消息发送和本地业务逻辑为同一个事务

@Transactional public void add(String userId, String commodityCode, Integer count) { BigDecimal orderMoney = new BigDecimal(count).multiply(new BigDecimal(5)); Order order = new Order().setUserId(userId).setCommodityCode(commodityCode).setCount(count).setMoney( orderMoney); orderDAO.insert(order); JSONObject messageJson = new JSONObject(); messageJson.put("commodityCode", commodityCode); messageJson.put("count", count); Message messageEntity = new Message(); messageEntity.setMessageStatus("生成订单成功"); messageEntity.setExchange(TopicEnum.ORDER_SYN_STOCK.getExchangeName()); messageEntity.setRoutingKey(TopicEnum.ORDER_SYN_STOCK.getRoutingKey()); messageEntity.setMessageStatus(CommonConstant.PRE_SEND_SUCCESS); messageEntity.setMessageEntity(messageJson.toJSONString()); //插入消息 messageDAO.insert(messageEntity); sendMessage(messageJson, messageEntity.getId().toString()); //服务异常,消息已经发出去了 // int i=1/0; }

4、事务接收者

事务消费者：

手动ACK保证消息被正确消费掉

public static final String queue = "naocs-order-service.order.syn.stock.queue"; @RabbitListener(queues = {queue}) public void alarmHandler(Message message, Channel channel) throws IOException { int count = 0; String orderMessageId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation"); String msgStr = new String(message.getBody(), "UTF-8"); try { log.info("消费者接收,msgStr={} ", msgStr); channel.basicQos(1); //同步查询订单消息表是否存在，如果存在再往下处理,订单消息表和生成订单是同一个本地事务，如果订单生成回滚，本地不会有订单消息数据， // 消费者订单消息查询必须是独占锁，保证订单消息生成或回滚再去查询 Boolean exit = nacosOrderFeignClient.exit(Long.valueOf(orderMessageId)); if (exit) { synStock(msgStr, orderMessageId); } } catch (Exception e) { e.printStackTrace(); ++count; //失败自旋3次，一般最多三次，太多会严重影响型性能 if (count <= 3) { synStock(msgStr, orderMessageId); } //失败插入一条消息 StockMessage stockMessage = stockMessageDAO.synGetById(Long.valueOf(orderMessageId)); if(stockMessage==null){ stockMessage = new StockMessage(); stockMessage.setId(Long.valueOf(orderMessageId)); stockMessage.setMessage(msgStr); stockMessage.setMessageStatus(fail); stockMessageDAO.insert(stockMessage); } } finally { channel.basicAck(message.getMessageProperties().getDeliveryTag(), true); } }

MessageDAOMapper.xml

<?xml version="1.0" encoding="UTF-8"?> <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd"> <mapper namespace="com.inca.repository.MessageDAO"> <select id="sysGetById" parameterType="long" resultType="com.inca.model.Message"> select * from message_tbl where id=#{id} for update; </select> </mapper>

数据同步：

/** * 数据同步要保证幂等性 * @param msgStr * @param orderMessageId */ @Transactional void synStock(String msgStr, String orderMessageId) { //独占锁方式查询库存消息表是否存在 // 保证消费者的幂等性 StockMessage stockMessage = stockMessageDAO.synGetById(Long.valueOf(orderMessageId)); if (stockMessage == null||fail.equals(stockMessage.getMessage())) { JSONObject msgJson = JSONObject.parseObject(msgStr); String commodityCode = msgJson.getString("commodityCode"); Integer count = msgJson.getInteger("count"); stockService.deduct(commodityCode, count); if(stockMessage==null){ stockMessage = new StockMessage(); } stockMessage.setId(Long.valueOf(orderMessageId)); stockMessage.setMessage(msgStr); stockMessage.setMessageStatus(success); if(stockMessage==null){ stockMessageDAO.insert(stockMessage); }else{ stockMessageDAO.updateById(stockMessage); } } }

StockMessageDaoMapper.xml

<?xml version="1.0" encoding="UTF-8"?> <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd"> <mapper namespace="com.inca.repository.StockMessageDAO"> <select id="synGetById" parameterType="long" resultType="com.inca.entity.StockMessage"> select * from stock_message_tbl where id=#{id} for update; </select> </mapper>

定时器

@Scheduled(cron = "*/6 * * * * ?") void timer(){ //查询消费失败的消息，重新同步数据 QueryWrapper<StockMessage> wrapper = new QueryWrapper<>(); wrapper.eq("message_status",fail); List<StockMessage> stockMessages = stockMessageDAO.selectList(wrapper); if(!CollectionUtils.isEmpty(stockMessages)){ stockMessages.stream().forEach((item)->{ synStock(item.getMessage(),item.getMessage()); }); } //查询消费成功的消费，删除订单消息表和库存消息表 wrapper.eq("message_status",success); stockMessages = stockMessageDAO.selectList(wrapper); if(!CollectionUtils.isEmpty(stockMessages)){ stockMessages.stream().forEach((item)->{ //删除订单消息表 nacosOrderFeignClient.deleteMessageById(item.getId()); //删除库存消息表 Boolean exit = nacosOrderFeignClient.exit(item.getId()); if(!exit){ stockMessageDAO.deleteById(item.getId()); } }); } }

5、总结

可靠消息最终一致性相比seate的优势是尽可能地少使用锁机制，从而保证更大的吞吐量，缺点是有一定的消息延时，对于实时性要求不高的可以用可靠性消息，对于实时性要求较高的可以用seata来解决分布式事务。