RockerMq之事务消息

分手后的思念是犯贱 2023-07-02 08:29 96阅读 0赞

本篇博客的RocketMq版本为4.3.0，首先附上RocketMq事务消息的流程图，以下的实践也都是基于此流程图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzEzNjM1_size_16_color_FFFFFF_t_70][]

##  原理 ##

1.  投递消息：Producer向Broker投递一个事务消息，并且带有唯一的key作为参数（幂等性)
2.  Broker预提交消息（在Broker本地做了存储，但是该消息的状态对Consumer不可见）
3.  Broker预提交成功后回调Producer的executeLocalTransaction方法
4.  Producer提交业务(比如记录最终成功投递的日志），并根据业务提交的执行情况，向Broker反馈Commit 或者回滚
5.  Broker最终处理

*  Broker监听到Producer发来的Commit反馈时，会最终提交这个消息到本地，此时该事务消息对Consumer可见，事务消息最终投递成功，事务结束
 *  Broker监听到Producer发来的RollBack反馈时，会最终回滚掉本地的预提交的消息，事务消息最终投递失败，事务结束
 *  Broker超时未接受到Producer的反馈，会定时重试调用Producer.checkLocalTransaction，Producer会根据自己的执行情况Ack给Broker

### Ack消息的3种状态 ###

Broker是根据Producer发送过来的状态码，来决定下一步的操作（提交、回滚、重试）

1.  TransactionStatus.CommitTransaction: commit transaction，it means that allow consumers to consume this message.
2.  TransactionStatus.RollbackTransaction: rollback transaction，it means that the message will be deleted and not allowed to consume.
3.  TransactionStatus.Unknown: intermediate state，it means that MQ is needed to check back to determine the status.

### Producer实现2个接口方法： ###

实际上官方的这种模式，重试指的是check的重试而不是execute的重试，因为execute方法只会执行一次，要特别注意。

*  execute：最终执行本地事务,并Ack执行状态给Broker
 *  check：检查Producer是否成功执行了事务,并Ack执行状态给Broker  
    实际上是可以写在一个方法里面的，execute的时候先根据key进行check，已经执行了就Ack OK，没有的话就执行。执行成功Ack Ok，执行失败就Ack RollBack。  
    但是这里官方把这个功能拆分的更细了，降低单一方法的复杂度

### 事务消息的优点 ###

1.  消息的投递失败时(比如MQ宕机或者网络丢失),Producer是可以感知到的，因为最终的业务提交是在回调的execute方法里面执行的
2.  如果消息成功发送到Broker，但是没有Producer最终Commit Ack时（比如Producer宕机了），该事务消息仍然处于预提交的状态，不会被消费者读取到，这保证了消息在 P和C端的状态一致性

## 实践 ##

### 生产端项目 ###

RocketMqConfig配置生产者

package com.yj.producer.config;
    
    import org.apache.rocketmq.client.producer.TransactionMQProducer;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import com.yj.producer.listener.MyTransactionListener;
    
    @Configuration
    public class RocketMqConfig {
    
    	@Value("${rocketmq.producer.groupName}")
    	private String producerGroupName;
    
    	@Value("${rocketmq.namesrv.addr}")
    	private String nameServerAddress;
    
    	@Bean(initMethod = "start", destroyMethod = "shutdown")
    	public TransactionMQProducer producer(MyTransactionListener myTransactionListener) {
    		TransactionMQProducer producer = new TransactionMQProducer(producerGroupName);
    		// VipChannel阿里内部使用版本才用，开源版本没有，默认为true，占用10909端口，此时虚拟机需要开放10909端口，否则会报
    		// ：connect to <：10909> failed异常，可以直接设置为false
    		// producer.setVipChannelEnabled(false);
    		producer.setNamesrvAddr(nameServerAddress);
    		producer.setTransactionListener(myTransactionListener);
    		return producer;
    	}
    }

MyTransactionListener事务监听器，主要有两个方法，一个执行本地事务的方法，一个提供给MqServer进行消息状态回查的方法，对于上面的流程图中的步骤③和⑥

package com.yj.producer.listener;
    
    import java.util.HashMap;
    import java.util.concurrent.ConcurrentHashMap;
    import org.apache.rocketmq.client.producer.LocalTransactionState;
    import org.apache.rocketmq.client.producer.TransactionListener;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    import com.yj.producer.model.MsgStatus;
    import com.yj.producer.service.AccountService;
    
    @Component
    public class MyTransactionListener implements TransactionListener {
    
    	private static final Logger log = LoggerFactory.getLogger(MyTransactionListener.class);
    
    	@Autowired
    	private AccountService accountService;
    
    	private ConcurrentHashMap<String, Integer> transStatusMap = new ConcurrentHashMap<>();
    
    	@Override
    	public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
    		String transactionId = msg.getTransactionId();
    		log.info("准备开始更新账户,当前状态为:"+MsgStatus.UNKNOW);
    		transStatusMap.put(transactionId, MsgStatus.UNKNOW.getStatus());
    
    		HashMap<String, Object> map = (HashMap<String, Object>) arg;
    		long userId = (long) map.get("user_id");
    		long productId = (long) map.get("product_id");
    		String errorFlag = (String) map.get("errorFlag");
    		String timeFlag = (String) map.get("timeFlag");
    		try {
    			accountService.updateAccount(userId, productId, errorFlag,timeFlag);
    			log.info("更新账户成功");
    		} catch (Exception e) {
    			log.info("更新账户发生异常,事务回滚,状态为:"+MsgStatus.ROLLBACK);
    			transStatusMap.put(transactionId, MsgStatus.ROLLBACK.getStatus());
    			return LocalTransactionState.ROLLBACK_MESSAGE;
    		}
    		log.info("更新账户成功,状态为:"+MsgStatus.COMMIT);
    		transStatusMap.put(transactionId, MsgStatus.COMMIT.getStatus());
    		return LocalTransactionState.COMMIT_MESSAGE;
    	}
    
    	@Override
    	public LocalTransactionState checkLocalTransaction(MessageExt msg) {
    		String transactionId = msg.getTransactionId();
    		Integer status = transStatusMap.get(transactionId);
    		if (status == MsgStatus.COMMIT.getStatus()) {
    			log.info("消息回查,状态为:"+MsgStatus.COMMIT);
    			return LocalTransactionState.COMMIT_MESSAGE;
    		} else if (status == MsgStatus.ROLLBACK.getStatus()) {
    			log.info("消息回查,状态为:"+MsgStatus.ROLLBACK);
    			return LocalTransactionState.ROLLBACK_MESSAGE;
    		}
    		log.info("消息回查,状态为:"+MsgStatus.UNKNOW);
    		return LocalTransactionState.UNKNOW;
    	}
    }

自定义MQProducer封装生产者发送事务消息的方法

package com.yj.producer.producer;
    
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.client.producer.TransactionMQProducer;
    import org.apache.rocketmq.common.message.Message;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    import com.yj.producer.model.BusinessException;
    
    @Component
    public class MQProducer {
    
    	@Autowired
    	private TransactionMQProducer producer;
    
    	public SendResult sendTranscationMsg(Message message,Object arg) throws BusinessException {
    		SendResult result = null;
    		try {
    			result = producer.sendMessageInTransaction(message, arg);
    		} catch (Exception e) {
    			e.printStackTrace();
    			throw new BusinessException("发送事务消息发生异常");
    		}
    		return result;
    	}
    }

AccountService

package com.yj.producer.service;
    
    import java.util.HashMap;
    import org.apache.rocketmq.common.message.Message;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Transactional;
    import com.alibaba.fastjson.JSON;
    import com.yj.producer.dao.AccountDao;
    import com.yj.producer.model.Storage;
    import com.yj.producer.producer.MQProducer;
    
    @Service
    public class AccountService {
    
    	private static final Logger log = LoggerFactory.getLogger(AccountService.class);
    
    	@Autowired
    	private AccountDao accountDao;
    	
    	@Autowired
    	private MQProducer producer;
    	
        @Value("${rocketmq.topic}")
        private String topic;
    
    	/**
    	 * 本地事务
    	 */
    	@Transactional
    	public String updateAccount(long userId, long productId, String errorFlag,String timeFlag) {
    		business(userId, productId, errorFlag,timeFlag);
    		return "下单成功";
    	}
    	
    	public String mqTransaction(long userId, long productId, String errorFlag,String timeFlag) {
    		Message message = new Message();
    		message.setTopic(topic);
    		
    		Storage storage=new Storage();
    		storage.setProductId(productId);
    		String storageStr = JSON.toJSONString(storage);
    		message.setBody(storageStr.getBytes());
    		message.setTags(productId + "_tag");
    		
    		HashMap<String,Object> arg=new HashMap<>();
    		arg.put("user_id", userId);
    		arg.put("product_id", productId);
    		arg.put("errorFlag", errorFlag);
    		arg.put("timeFlag", timeFlag);
    		producer.sendTranscationMsg(message,arg);
    		return "下单成功";
    	}
    
    	private void business(long userId, long productId, String errorFlag,String timeFlag) {
    		log.info("==账户服务扣减账户余额开始==");
    		accountDao.updateAccount(userId);
    		if ("1".equals(errorFlag)) {
    			int i = 1 / 0;
    		}
    		if ("1".equals(timeFlag)) {
    			try {
    				Thread.sleep(120000L);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    		log.info("==账户服务扣减账户余额结束==");
    	}
    }

AccountController

package com.yj.producer.controller;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.PostMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.RestController;
    import com.yj.producer.service.AccountService;
    
    @RestController
    public class AccountController {
    
    	@Autowired
    	private AccountService accountService;
    
    	@PostMapping("/mqTransaction")
    	public String mqTransaction(@RequestParam("user_id") long userId, @RequestParam("product_id") long productId,
    			@RequestParam("errorFlag") String errorFlag, @RequestParam("timeFlag") String timeFlag) {
    		return accountService.mqTransaction(userId, productId, errorFlag, timeFlag);
    	}
    }

MsgStatus

package com.yj.producer.model;
    
    public enum MsgStatus {
    	COMMIT(1,"提交"),
    	ROLLBACK(2,"回滚"),
    	UNKNOW(3,"未知");
    	private int status;
    	private String msg;
    	private MsgStatus(int status, String msg) {
    		this.status = status;
    		this.msg = msg;
    	}
    	public int getStatus() {
    		return status;
    	}
    	public String getMsg() {
    		return msg;
    	}
    }

### 消费端项目 ###

消费端的配置没有什么特别的地方，可以参考mq顺序消息的消费者的配置，消费端监听到消息后，执行本地事务

package com.yj.consumer.consumer;
    
    import org.apache.rocketmq.common.message.MessageExt;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Service;
    import com.alibaba.fastjson.JSON;
    import com.yj.consumer.model.Storage;
    import com.yj.consumer.service.StorageService;
    
    @Service
    public class MessageProcessor {
    	
    	@Value("${server.port}")  
        private String port; 
    	
    	@Autowired
    	private StorageService storageService;
    
    	private static final Logger LOGGER = LoggerFactory.getLogger(MessageProcessor.class);
    
    	public Boolean handleMessage(MessageExt messageExt) {
    		try {
    			String body = null;
    			// 消费者消费
    			if (null == messageExt || null == messageExt.getBody()) {
    				LOGGER.info("消息体为空");
    				return false;
    			}
    			body = new String(messageExt.getBody());
    			Storage storage = JSON.parseObject(body, Storage.class);
    			LOGGER.info("port:"+port+",消费Tag:" + messageExt.getTags() + ",storage:" + storage);
    			storageService.updateStorage(storage.getProductId());
    		} catch (Exception e) {
    			e.printStackTrace();
    			LOGGER.error(e.getMessage(), e);
    			return false;
    		}
    		return true;
    	}
    }

StorageService

package com.yj.consumer.service;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Transactional;
    import com.yj.consumer.dao.StorageDao;
    
    @Service
    public class StorageService {
    
    	private static final Logger log = LoggerFactory.getLogger(StorageService.class);
    	 
    	@Autowired
        private StorageDao storageDao;
    	
    	 /**
         * 扣减库存
         */
    	@Transactional
        public void updateStorage(long productId) {
    		log.info("==库存服务扣减库存开始==");
            try {
    			storageDao.updateStorage(productId);
    			//int i=1/0;
    		} catch (Exception e) {
    			log.info("消费方消费时发生异常,此时需人工介入处理");
    			e.printStackTrace();
    		}
            log.info("==库存服务扣库存结束==");
        }
    }

启动MQ环境，开始验证

①生产端，消费端均正常的情况

http://127.0.0.1:8001/mqTransaction?user_id=1&product_id=1&errorFlag=0&timeFlag=0

生产方日志

2020-01-28 16:12:34.859  INFO 12580 --- [nio-8001-exec-1] c.y.p.listener.MyTransactionListener     : 准备开始更新账户,当前状态为:UNKNOW
    2020-01-28 16:12:35.142  INFO 12580 --- [nio-8001-exec-1] com.yj.producer.service.AccountService   : ==账户服务扣减账户余额开始==
    2020-01-28 16:12:35.163 DEBUG 12580 --- [nio-8001-exec-1] c.y.p.dao.AccountDao.updateAccount       : ==>  Preparing: update account set money = money-10 where user_id = ? 
    2020-01-28 16:12:35.266 DEBUG 12580 --- [nio-8001-exec-1] c.y.p.dao.AccountDao.updateAccount       : ==> Parameters: 1(Long)
    2020-01-28 16:12:35.270 DEBUG 12580 --- [nio-8001-exec-1] c.y.p.dao.AccountDao.updateAccount       : <==    Updates: 1
    2020-01-28 16:12:35.270  INFO 12580 --- [nio-8001-exec-1] com.yj.producer.service.AccountService   : ==账户服务扣减账户余额结束==
    2020-01-28 16:12:35.276  INFO 12580 --- [nio-8001-exec-1] c.y.p.listener.MyTransactionListener     : 更新账户成功
    2020-01-28 16:12:35.276  INFO 12580 --- [nio-8001-exec-1] c.y.p.listener.MyTransactionListener     : 更新账户成功,状态为:COMMIT

消费方日志

2020-01-28 16:12:35.560  INFO 15628 --- [MessageThread_1] com.yj.consumer.service.StorageService   : ==库存服务扣减库存开始==
    2020-01-28 16:12:35.575 DEBUG 15628 --- [MessageThread_1] c.y.c.dao.StorageDao.updateStorage       : ==>  Preparing: update storage set num = num-1 where product_id = ? 
    2020-01-28 16:12:35.675 DEBUG 15628 --- [MessageThread_1] c.y.c.dao.StorageDao.updateStorage       : ==> Parameters: 1(Long)
    2020-01-28 16:12:35.682 DEBUG 15628 --- [MessageThread_1] c.y.c.dao.StorageDao.updateStorage       : <==    Updates: 1
    2020-01-28 16:12:35.682  INFO 15628 --- [MessageThread_1] com.yj.consumer.service.StorageService   : ==库存服务扣库存结束==

业务正常执行

②生产端执行异常的情况

http://127.0.0.1:8001/mqTransaction?user_id=1&product_id=1&errorFlag=1&timeFlag=0

生产端日志

2020-01-28 16:14:58.768  INFO 12580 --- [nio-8001-exec-3] c.y.p.listener.MyTransactionListener     : 准备开始更新账户,当前状态为:UNKNOW
    2020-01-28 16:14:58.782  INFO 12580 --- [nio-8001-exec-3] com.yj.producer.service.AccountService   : ==账户服务扣减账户余额开始==
    2020-01-28 16:14:58.782 DEBUG 12580 --- [nio-8001-exec-3] c.y.p.dao.AccountDao.updateAccount       : ==>  Preparing: update account set money = money-10 where user_id = ? 
    2020-01-28 16:14:58.783 DEBUG 12580 --- [nio-8001-exec-3] c.y.p.dao.AccountDao.updateAccount       : ==> Parameters: 1(Long)
    2020-01-28 16:14:58.786 DEBUG 12580 --- [nio-8001-exec-3] c.y.p.dao.AccountDao.updateAccount       : <==    Updates: 1
    2020-01-28 16:14:58.813  INFO 12580 --- [nio-8001-exec-3] c.y.p.listener.MyTransactionListener     : 更新账户发生异常,事务回滚,状态为:ROLLBACK

消费端没有接收到消息，无日志输出

③生产端执行超时的情况

生产端日志，发现进行了消息的回查

2020-01-28 16:18:57.914  INFO 3852 --- [nio-8001-exec-1] c.y.p.listener.MyTransactionListener     : 准备开始更新账户,当前状态为:UNKNOW
    2020-01-28 16:18:58.222  INFO 3852 --- [nio-8001-exec-1] com.yj.producer.service.AccountService   : ==账户服务扣减账户余额开始==
    2020-01-28 16:18:58.238 DEBUG 3852 --- [nio-8001-exec-1] c.y.p.dao.AccountDao.updateAccount       : ==>  Preparing: update account set money = money-10 where user_id = ? 
    2020-01-28 16:18:58.367 DEBUG 3852 --- [nio-8001-exec-1] c.y.p.dao.AccountDao.updateAccount       : ==> Parameters: 1(Long)
    2020-01-28 16:18:58.371 DEBUG 3852 --- [nio-8001-exec-1] c.y.p.dao.AccountDao.updateAccount       : <==    Updates: 1
    2020-01-28 16:20:16.033  INFO 3852 --- [pool-1-thread-1] c.y.p.listener.MyTransactionListener     : 消息回查,状态为:UNKNOW
    2020-01-28 16:20:57.938  INFO 3852 --- [lientSelector_1] RocketmqRemoting                         : closeChannel: close the connection to remote address[192.168.190.131:10909] result: true
    2020-01-28 16:20:58.373  INFO 3852 --- [nio-8001-exec-1] com.yj.producer.service.AccountService   : ==账户服务扣减账户余额结束==
    2020-01-28 16:20:58.376  INFO 3852 --- [nio-8001-exec-1] c.y.p.listener.MyTransactionListener     : 更新账户成功
    2020-01-28 16:20:58.376  INFO 3852 --- [nio-8001-exec-1] c.y.p.listener.MyTransactionListener     : 更新账户成功,状态为:COMMIT

消费端日志，超时结束后，消费端，消费了消息

2020-01-28 16:20:58.382  INFO 15628 --- [MessageThread_2] com.yj.consumer.service.StorageService   : ==库存服务扣减库存开始==
    2020-01-28 16:20:58.383 DEBUG 15628 --- [MessageThread_2] c.y.c.dao.StorageDao.updateStorage       : ==>  Preparing: update storage set num = num-1 where product_id = ? 
    2020-01-28 16:20:58.383 DEBUG 15628 --- [MessageThread_2] c.y.c.dao.StorageDao.updateStorage       : ==> Parameters: 1(Long)
    2020-01-28 16:20:58.388 DEBUG 15628 --- [MessageThread_2] c.y.c.dao.StorageDao.updateStorage       : <==    Updates: 1
    2020-01-28 16:20:58.388  INFO 15628 --- [MessageThread_2] com.yj.consumer.service.StorageService   : ==库存服务扣库存结束==
    2020-01-28 16:21:16.052  INFO 15628 --- [MessageThread_3] c.yj.consumer.consumer.MessageProcessor  : port:8004,消费Tag:1_tag,storage:Storage(id=0, productId=1, num=null)
    2020-01-28 16:21:16.053  INFO 15628 --- [MessageThread_3] com.yj.consumer.service.StorageService   : ==库存服务扣减库存开始==
    2020-01-28 16:21:16.053 DEBUG 15628 --- [MessageThread_3] c.y.c.dao.StorageDao.updateStorage       : ==>  Preparing: update storage set num = num-1 where product_id = ? 
    2020-01-28 16:21:16.053 DEBUG 15628 --- [MessageThread_3] c.y.c.dao.StorageDao.updateStorage       : ==> Parameters: 1(Long)
    2020-01-28 16:21:16.057 DEBUG 15628 --- [MessageThread_3] c.y.c.dao.StorageDao.updateStorage       : <==    Updates: 1
    2020-01-28 16:21:16.057  INFO 15628 --- [MessageThread_3] com.yj.consumer.service.StorageService   : ==库存服务扣库存结束==

④消费端异常的情况

消费端异常的情况下，需人工介入处理。从工程实践角度讲，这种整个流程自动回滚的代价是非常巨大的，不但实现复杂，还会引入新的问题。比如自动回滚失败，又怎么处理？对应这种极低概率的case，采取人工处理，会比实现一个高复杂的自动化回滚系统，更加可靠，也更加简单。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzEzNjM1_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200128163319241.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzEzNjM1,size_16,color_FFFFFF,t_70