RocketMQ（三）——————javaAPI (7.事务消息)

Half Message：

预处理消息，当broker收到此类消息后，会存储到RMQ_SYS_TRANS_HALF_TOPIC的消息消费队列中

检查事务状态：

Broker会开启一个定时任务，消费RMQ_SYS_TRANS_HALF_TOPIC队列中的消息，

每次执行任务会向消息发送者确认事务执行状态（提交、回滚、未知），如果是未知，等待下一次回调。

超时：

如果超过回查次数，默认回滚消息

TransactionListener的两个方法

executeLocalTransaction

半消息发送成功触发此方法来执行本地事务

checkLocalTransaction

broker将发送检查消息来检查事务状态，并将调用此方法来获取本地事务状态

本地事务执行状态

LocalTransactionState.COMMIT_MESSAGE

执行事务成功，确认提交

LocalTransactionState.ROLLBACK_MESSAGE

回滚消息，broker端会删除半消息

LocalTransactionState.UNKNOW

暂时为未知状态，等待broker回查

1、生产者样例

//1.发送事务消息
    public static void main(String[] args) throws Exception {

        TransactionMQProducer producer = new TransactionMQProducer("TransactionGroup");

        producer.setNamesrvAddr("127.0.0.1:9876");

        producer.setTransactionListener(new TransactionListener() {
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {

                System.out.println("==e xecuteLocalTransaction==");
                System.out.println("message-body : "+message.getBody());
                System.out.println("message-TransactionId : "+message.getTransactionId());

                try {

                    //业务

                }catch (Exception e){

                    //回滚消息，broker端会删除半消息
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                //执行事务成功，确认提交
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {

                System.out.println("==c heckLocalTransaction==");
                System.out.println("message-body : "+new String(messageExt.getBody()));
                System.out.println("message-TransactionId : "+messageExt.getTransactionId());

                //暂时为未知状态，等待broker回查
                //return LocalTransactionState.UNKNOW;
                //回滚消息，broker端会删除半消息
                //return LocalTransactionState.ROLLBACK_MESSAGE;

                //执行事务成功，确认提交
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        producer.start();
        TransactionSendResult sendResult = producer.sendMessageInTransaction(new Message
                ("TransactionTopic", "事务消息！".getBytes()), null);

        System.out.println("sendResult : "+sendResult);
        producer.shutdown();
        System.out.println("生产者下线！");

    }

2、消费者样例

//1.接收事务消息
    public static void main(String[] args) throws Exception {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("TransactionGroup");

        consumer.setNamesrvAddr("127.0.0.1:9876");

        consumer.subscribe("TransactionTopic","*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                for (MessageExt mes: list) {

                    System.out.println("mes : "+new String(mes.getBody()));
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.println("Consumer  start...");
    }

事务消息使用上的限制：

1. 事务消息不支持延时消息和批量消息。

2. 为了避免单个消息被检查太多次而导致半队列消息累积，我们默认将单个消息的检查次数限制为 15 次，

但是用户可以通过 Broker 配置文件的 transactionCheckMax参数来修改此限制。

如果已经检查某条消息超过 N 次的话（ N = transactionCheckMax ）则 Broker 将丢弃此消息，

并在默认情况下同时打印错误日志。用户可以通过重写 AbstractTransactionalMessageCheckListener 类来修改这个行为。

3. 事务消息将在 Broker 配置文件中的参数 transactionTimeout 这样的特定时间长度之后被检查。

当发送事务消息时，用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制，

该参数优先于 transactionTimeout 参数。

4. 事务性消息可能不止一次被检查或消费。

5. 提交给用户的目标主题消息可能会失败，目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证，

如果希望确保事务消息不丢失、并且事务完整性得到保证，建议使用同步的双重写入机制。

6. 事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，

事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。

RocketMQ（三）——————javaAPI (7.事务消息)

相关推荐