微服务架构设计RocketMQ进阶事务消息原理详解

目录

• 前言
• 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

@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
		);
	}
}

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);
	}
}

@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()
	);
}

@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");
    }
}