rocketMQ最大事务通知原理 代码怎么写

RocketMQ最大事务通知原理是指在RocketMQ的事务消息中，事务状态的提交和回滚由消息生产者来通知，而不是直接由事务消息的消费者来通知。

代码实现如下：

1. 在事务消息生产者中，定义一个TransactionListener接口的实现类，重写其中的checkLocalTransaction方法和executeLocalTransaction方法，分别用于执行本地事务和检查本地事务状态。

public class MyTransactionListener implements TransactionListener {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行本地事务
        return LocalTransactionState.COMMIT_MESSAGE;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 检查本地事务状态
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

2. 在事务消息生产者中，设置上述实现类为TransactionListener，并发送事务消息。

TransactionMQProducer producer = new TransactionMQProducer("myProducerGroup");
producer.setNamesrvAddr("localhost:9876");
producer.setTransactionListener(new MyTransactionListener());
producer.start();

Message msg = new Message("myTopic", "myTag", "myKey", "myValue".getBytes());
// 发送事务消息
SendResult sendResult = producer.sendMessageInTransaction(msg, null);

3. 在消息消费者中，通过设置ConsumeMessageHook接口的实现类，重写beforeConsume方法和afterConsume方法，分别用于接收事务消息之前和之后的处理。

public class MyConsumeMessageHook implements ConsumeMessageHook {
    @Override
    public String hookName() {
        return "myConsumeMessageHook";
    }

    @Override
    public void consumeMessageBefore(ConsumeMessageContext context) {
        // 接收事务消息之前的处理
    }

    @Override
    public void consumeMessageAfter(ConsumeMessageContext context) {
        // 接收事务消息之后的处理
        MessageExt msg = context.getMessageExt();
        // 判断事务消息的状态
        if (msg.getTransactionId() != null) {
            switch (msg.getPreparedTransactionOffset()) {
                case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                    // 事务提交
                    break;
                case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                    // 事务回滚
                    break;
                default:
                    // 事务未完成
                    break;
            }
        }
    }
}

4. 在消息消费者中，设置上述实现类为ConsumeMessageHook，并启动消费者。

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myConsumerGroup");
consumer.setNamesrvAddr("localhost:9876");
consumer.getDefaultMQPushConsumerImpl().registerConsumeMessageHook(new MyConsumeMessageHook());
consumer.subscribe("myTopic", "*");
consumer.start();