rocketmq实战之发送事务消息

一时失言乱红尘 2023-05-31 03:25 2阅读 0赞

rocketmq支持发送事务消息，即发送消息的事务性，这里便来看一下如何实现？

> 实战方面均以`RocketMQTemplate`形式展现，集成方案详见[《springboot中rocketmq的集成与使用》][springboot_rocketmq]

首先使用`@RocketMQTransactionListener`定义一个监听器来模拟执行本地事务和事务会查：

@Slf4j
    @RocketMQTransactionListener(txProducerGroup = "tx-group")
    public class TransactionListenerImpl implements RocketMQLocalTransactionListener { 
    
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) { 
    
            // 模拟本地事务不通过
            log.info("============== executeLocalTransaction");
    
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) { 
    
            // 模拟回查本地事务
            log.info("============== checkLocalTransaction");
    
    
            return RocketMQLocalTransactionState.COMMIT;
        }
    }

然后再定义一个消费者，设定`topic`为“topic-tx”：

@RocketMQMessageListener(topic = "topic-tx", consumerGroup = "tx-consumer-group")
    public class TransactionConsumer implements RocketMQListener<Message> { 
    
        @Override
        public void onMessage(Message message) { 
            log.info("topic-tx received message: {}", message);
        }
    }

最后定义一个消息生产者，向“topic-tx”主题发送消息，并且需要指定`gorup`为“tx-group”（同事务监听器所设置的`txProducerGroup`一致）：

@Slf4j
    public class TransactionProducer { 
    
        @Resource
        private RocketMQTemplate rocketMQTemplate;
    
        public void produce() { 
            Message<String> message = new Message<>();
            message.setId(UUID.randomUUID().toString());
            message.setContent("transaction message");
            log.info("========sending message=========");
            rocketMQTemplate.sendMessageInTransaction("tx-group", "topic-tx", MessageBuilder.withPayload(message).build(), null);
            log.info("========finish send =========");
        }
    }

跑起项目后，在控制台可看到如下输出：

========sending message=========
     ============== executeLocalTransaction
     ========finish send =========
     ============== checkLocalTransaction
     topic-tx received message: Message(id=168486dd-0814-4060-ace1-5a55449b3f72, content=transaction message)

[springboot_rocketmq]: https://lazycece.github.io/2019/06/16/springboot%E4%B8%ADrocketmq%E7%9A%84%E9%9B%86%E6%88%90%E4%B8%8E%E4%BD%BF%E7%94%A8/