您的位置:

深入了解RocketMQ事务消息

一、什么是RocketMQ事务消息

RocketMQ事务消息是指在消息发送方发送消息时,延迟将消息状态提交给broker,由broker进行二次确认,以确保消息不会因发送失败而丢失,同时可以保证消息的一致性。

具体操作流程为:在事务消息发送完毕后,transactionProducer会等待确认回调中返回TransactionStatus的状态,根据回调的状态执行send或者rollback操作。

实现RocketMQ事务消息至少需要三步操作:执行本地事务 - 提交half消息 - 提交事务。

TransactionMQProducer transactionProducer = new TransactionMQProducer("group_name");
transactionProducer.setTransactionListener(new TransactionListenerImpl());
transactionProducer.start();

Message message = new Message("topic_name", "message_body".getBytes(Charset.defaultCharset()));
transactionProducer.sendMessageInTransaction(message, null);

二、如何实现RocketMQ事务消息

1. 实现本地事务

实现RocketMQ事务消息首先需要实现本地事务,即执行SQL或者其他操作,将需要发送的消息与该事务绑定在一起。

实现本地事务需要保证事务的原子性、一致性、隔离性和持久性。

public class TransactionListenerImpl implements TransactionListener {
    private final ConcurrentHashMap localTrans = new ConcurrentHashMap<>();

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        String transactionId = "transaction_id";
        // 执行本地事务操作,返回事务状态
        int status = executeLocalTransaction(transactionId);
        localTrans.put(transactionId, status);
        return LocalTransactionState.UNKNOW;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        String transactionId = "transaction_id";
        Integer status = localTrans.get(transactionId);
        if (status != null) {
            switch (status) {
                case 1:
                    return LocalTransactionState.COMMIT_MESSAGE;
                case 2:
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                default:
                    return LocalTransactionState.UNKNOW;
            }
        }
        return LocalTransactionState.UNKNOW;
    }
}

  

2. 提交Half消息

Half消息是指在交易“半成品”,是指事务消息发送成功之后,但是transactionProducer等待确认回调的过程中,半成品的消息实体。

提交Half消息需要先构建消息体,然后使用transactionProducer的sendMessageInTransaction方法发送半成品消息。

Message message = new Message("topic_name", "message_body".getBytes(Charset.defaultCharset()));
transactionProducer.sendMessageInTransaction(message, null);

3. 提交事务消息

当本地事务提交成功之后,即可调用sendMessage方法将半成品消息实体提交为事务消息。

producer.sendMessage(message);

三、RocketMQ事务消息的优缺点

1. 优点

事务消息保证了消息发送的可靠性,满足了一些数据一致性比较高的业务场景。

对于分布式事务,RocketMQ使用可靠消息最终一致性方案,以避免分布式锁的使用。

使用RocketMQ事务消息可以在业务代码中实现一些基于消息的分布式事务。

2. 缺点

使用事务消息会增加数据的处理延迟,因为必须等待严格的维护。

事务消息的处理还需要保证SQL的执行过程是幂等的,以避免由于transactionProducer重试而引起数据不一致问题。

事务消息在提高数据可靠性的同时,会带来一些额外的系统复杂度,需要谨慎使用。

四、总结

顾名思义,事务消息通过事务的方式保证消息的可靠性,从而解决了消息发送可靠性的问题,尤其是在某些业务场景下数据一致性非常重要的应用场景。

在具体实践中,需要注意消息的幂等性以及其他一些额外复杂度,谨慎使用。