RocketMQ进阶-事务消息

时间 2020-05-28 标签 rocketmq 进阶事务消息

前言

分布式消息选型的时候是否支持事务消息是一个很重要的考量点，而目前只有RocketMQ对事务消息支持的最好。今天咱们来唠唠如何实现RocketMQ的事务消息！spring

Apache RocketMQ在4.3.0版中已经支持分布式事务消息，这里RocketMQ采用了2PC的思想来实现了提交事务消息，同时增长一个补偿逻辑来处理二阶段超时或者失败的消息，以下图所示。数据库

RocketMQ事务流程概要

RocketMQ实现事务消息主要分为两个阶段：正常事务的发送及提交、事务信息的补偿流程总体流程为：apache

正常事务发送与提交阶段
一、生产者发送一个半消息给MQServer（半消息是指消费者暂时不能消费的消息）
二、服务端响应消息写入结果，半消息发送成功
三、开始执行本地事务
四、根据本地事务的执行状态执行Commit或者Rollback操做app
事务信息的补偿流程
一、若是MQServer长时间没收到本地事务的执行状态会向生产者发起一个确认回查的操做请求
二、生产者收到确认回查请求后，检查本地事务的执行状态
三、根据检查后的结果执行Commit或者Rollback操做
补偿阶段主要是用于解决生产者在发送Commit或者Rollback操做时发生超时或失败的状况。dom

RocketMQ事务流程关键

一、事务消息在一阶段对用户不可见
事务消息相对普通消息最大的特色就是一阶段发送的消息对用户是不可见的，也就是说消费者不能直接消费。这里RocketMQ的实现方法是原消息的主题与消息消费队列，而后把主题改为RMQ_SYS_TRANS_HALF_TOPIC ，这样因为消费者没有订阅这个主题，因此不会被消费。分布式

二、如何处理第二阶段的失败消息？
在本地事务执行完成后会向MQServer发送Commit或Rollback操做，此时若是在发送消息的时候生产者出故障了，那么要保证这条消息最终被消费，MQServer会像服务端发送回查请求，确认本地事务的执行状态。
固然了rocketmq并不会无休止的的信息事务状态回查，默认回查15次，若是15次回查仍是没法得知事务状态，RocketMQ默认回滚该消息。ide

三、消息状态事务消息有三种状态：spring-boot

TransactionStatus.CommitTransaction：提交事务消息，消费者能够消费此消息微服务
TransactionStatus.RollbackTransaction：回滚事务，它表明该消息将被删除，不容许被消费。测试
TransactionStatus.Unknown ：中间状态，它表明须要检查消息队列来肯定状态。

实现

咱们构建这样一个需求：用户请求订单微服务 order-service 接口删除订单（退货），删除订单后须要发送消息给用户服务account-service，用户微服务收到消息后会给用户帐户增长余额。这个需求跟钱相关，确定要保证消息的事务性，接下来咱们根据上面的原理实现整个流程。

基础配置

生产者order-servcie和account-service都要引入RocketMQ相关依赖，增长RocketMQ的相关配置

引入组件

<dependency>
	<groupId>org.apache.rocketmq</groupId>
	<artifactId>rocketmq-spring-boot-starter</artifactId>
</dependency>

添加配置

# within rocketmq
rocketmq:
  name-server: xxx.xx.x.xx:9876; xxx.xx.x.xx:9876
  producer:
    group: cloud-group

发送半消息

order-service在执行删除订单操做时发送一条半消息给MQServer，发送半消息主要是使用rocketMQTemplate.sendMessageInTransaction() 方法，发送事务消息。

@Override
public void delete(String orderNo) {
	Order order = orderMapper.selectByNo(orderNo);
	//若是订单存在且状态为有效，进行业务处理
	if (order != null && CloudConstant.VALID_STATUS.equals(order.getStatus())) {
		String transactionId = UUID.randomUUID().toString();
		//若是能够删除订单则发送消息给rocketmq，让用户中心消费消息
		rocketMQTemplate.sendMessageInTransaction("add-amount",
				MessageBuilder.withPayload(
						UserAddMoneyDTO.builder()
								.userCode(order.getAccountCode())
								.amount(order.getAmount())
								.build()
				)
				.setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId)
				.setHeader("order_id",order.getId())
				.build()
				,order
		);
	}
}

首先先校验一下订单状态，而后发送消息给MQServer，这个逻辑你们都看得懂，主要是关注sendMessageInTransaction() 方法，源码以下：

public TransactionSendResult sendMessageInTransaction(String destination, Message<?> message, Object arg) throws MessagingException {
	try {
		if (((TransactionMQProducer)this.producer).getTransactionListener() == null) {
			throw new IllegalStateException("The rocketMQTemplate does not exist TransactionListener");
		} else {
			org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
			return this.producer.sendMessageInTransaction(rocketMsg, arg);
		}
	} catch (MQClientException var5) {
		throw RocketMQUtil.convert(var5);
	}
}

该方法有三个参数：

destination：目的地(主题)，这里发送给add-amount 这个主题
message：发送给消费者的消息体，须要使用 MessageBuilder.withPayload() 来构建消息
arg：参数

注意，这里咱们生成了一个transactionId，并放在header中跟消息一块儿发送（这里实际也能够构形成一个对象，放在arg里进行发送），做用后面再讲！

执行本地事务与回查

MQServer收到半消息后会告诉生产者order-service确认收到半消息，这时候order-service须要执行本地事务，执行完本地事务后再告诉MQServer本地事务的执行状态，确认消息到底是Commit仍是Rollback。若是在告诉MQServer本地执行状态的时候出异常了还须要让MQServer可以回查到，怎么实现这一些列操做呢？

RocketMQ提供了 RocketMQLocalTransactionListener 接口，本地事务监听器，这个接口类的实现以下：

第一个方法executeLocalTransaction 为执行本地事务；
第二个方法checkLocalTransaction 为检查本地事务的执行状态，也就是回查动做。
有了这个接口类咱们的执行逻辑清楚了，可是还有个问题：本地事务已经执行完成了，怎么去回查本地事务的执行结果呢？

咱们能够在执行本地事务的时候同时生成一个事务日志，让本地事务与日志事务在同一个方法中，同时添加@Transactional 注解，保证两个操做事务是一个原子操做。这样若是事务日志表中有这个本地事务的信息，那就表明本地事务执行成功，须要Commit，相反若是没有对应的事务日志，则表示没执行成功，须要Rollback

思路既然理顺了，我们就开撸。

首先建立一个日志表

很简单的三个字段，主要是这个事务id，须要根据这个事务id回查事务，还记得咱们在发送半消息时生成的事务id吗，就是干这个用的！
在生产者编写方法实现RocketMQLocalTransactionListener

@Slf4j
@RocketMQTransactionListener
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class AddUserAmountListener implements RocketMQLocalTransactionListener {
    private final OrderService orderService;
    private final RocketMqTransactionLogMapper rocketMqTransactionLogMapper;
    /**
     * 执行本地事务
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        log.info("执行本地事务");
        MessageHeaders headers = message.getHeaders();
        //获取事务ID
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        Integer orderId = Integer.valueOf((String)headers.get("order_id"));
        log.info("transactionId is {}, orderId is {}",transactionId,orderId);

        try{
            //执行本地事务，并记录日志
            orderService.changeStatuswithRocketMqLog(orderId, CloudConstant.INVALID_STATUS,transactionId);
            //执行成功，能够提交事务
            return RocketMQLocalTransactionState.COMMIT;
        }catch (Exception e){
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * 本地事务的检查，检查本地事务是否成功
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {

        MessageHeaders headers = message.getHeaders();
        //获取事务ID
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        log.info("检查本地事务,事务ID:{}",transactionId);
        //根据事务id从日志表检索
        QueryWrapper<RocketmqTransactionLog> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("transaction_id",transactionId);
        RocketmqTransactionLog rocketmqTransactionLog = rocketMqTransactionLogMapper.selectOne(queryWrapper);
        if(null != rocketmqTransactionLog){
            return RocketMQLocalTransactionState.COMMIT;
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }
}

执行本地事务的方法

@Transactional(rollbackFor = RuntimeException.class)
@Override
public void changeStatuswithRocketMqLog(Integer id,String status,String transactionId){
    //将订单状态置位无效
	orderMapper.changeStatus(id,status);
    //插入事务表
	rocketMqTransactionLogMapper.insert(
			RocketmqTransactionLog.builder()
					.transactionId(transactionId)
					.log("执行删除订单操做")
			.build()
	);
}

这一块的代码逻辑都是在生产端，即Order-Server，你们不要搞错了

消费消息

Rollback的消息MQServer会给咱们处理，咱们只要关注Commit状态时消费端能够正常消费便可。在account-service监听消息，若是收到消息则给用户帐户增长余额。

@Slf4j
@Service
@RocketMQMessageListener(topic = "add-amount",consumerGroup = "cloud-group")
@RequiredArgsConstructor(onConstructor = @__(@Autowired) )
public class AddUserAmountListener implements RocketMQListener<UserAddMoneyDTO> {
    private final AccountMapper accountMapper;
    /**
     * 收到消息的业务逻辑
     */
    @Override
    public void onMessage(UserAddMoneyDTO userAddMoneyDTO) {
        log.info("received message: {}",userAddMoneyDTO);
        accountMapper.increaseAmount(userAddMoneyDTO.getUserCode(),userAddMoneyDTO.getAmount());
        log.info("add money success");
    }
}

测试

订单表有这样一条记录，用户为jianzh5，amount为200

用户表的记录，执行完成后jianzh5的帐户应该变成250

调用删除订单接口，删除订单
发送半消息
执行本地事务，并生成事务日志
模拟异常状况在发送Commit消息的时候咱们用命令杀掉进程taskkill /pid 19748 -t -f，模拟异常！
从新启动order-service，查看是否会执行回查动做MQServer进行回查，检查事务日志，判断是否能够提交事务
消费者消费事务消息，保证事务的一致性

总结

使用RocketMQ实现事务消息的过程仍是很复杂的，须要好好理解开头的那张图，只有理解了事务消息的交互过程才能编写相应的代码！

好了，各位朋友们，本期的内容到此就所有结束啦，能看到这里的同窗都是优秀的同窗，下一个升职加薪的就是你了！
若是以为这篇文章对你有所帮助的话请扫描下面二维码加个关注。"转发" 加 "在看"，养成好习惯！我们下期再见！