package com.lynch.simple.demo;import com.alibaba.rocketmq.client.producer.LocalTransactionExecuter;
import com.alibaba.rocketmq.client.producer.LocalTransactionState;
import com.alibaba.rocketmq.common.message.Message;/*** 执行业务逻辑的部分* * @author jqlin**/
public class TransactionExecuterImpl implements LocalTransactionExecuter {@Overridepublic LocalTransactionState executeLocalTransactionBranch(Message msg, Object arg) {System.out.println("执行本地事务msg = " + new String(msg.getBody()));  System.out.println("执行本地事务arg = " + arg);  //DB操作 应该带上事务 service -> dao//如果数据操作失败  需要回滚    同时返回RocketMQ一条失败消息  意味着消费者无法消费到这条失败的消息//如果成功 就要返回一条rocketMQ成功的消息，意味着消费者将读取到这条消息//o就是attachmentString tags = msg.getTags();  if (tags.equals("transaction2")) {  System.out.println("===> 本地事务执行失败，进行MQ ROLLBACK");return LocalTransactionState.ROLLBACK_MESSAGE;  }  System.out.println("===> 本地事务执行成功，发送确认消息");// return LocalTransactionState.UNKNOW;  return LocalTransactionState.COMMIT_MESSAGE;  }}

package com.lynch.simple.demo;import com.alibaba.rocketmq.client.producer.LocalTransactionState;
import com.alibaba.rocketmq.client.producer.TransactionCheckListener;
import com.alibaba.rocketmq.common.message.MessageExt;/*** 处理事务回查的代码部分* * @author jqlin**/
public class TransactionCheckListenerImpl implements TransactionCheckListener {@Overridepublic LocalTransactionState checkLocalTransactionState(MessageExt msg) {System.out.println("未决事务，服务器回查客户端msg =" + new String(msg.getBody().toString()));//由于RocketMQ迟迟没有收到消息的确认消息，因此主动询问这条prepare消息，是否正常？//可以查询数据库看这条数据是否已经处理return LocalTransactionState.COMMIT_MESSAGE;}}

package com.lynch.simple.demo;import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.TransactionCheckListener;
import com.alibaba.rocketmq.client.producer.TransactionMQProducer;
import com.alibaba.rocketmq.common.message.Message;public class TransactionProducer {public static void main(String[] args) throws MQClientException, InterruptedException {// 也就是上文所说的，当RocketMQ发现`Prepared消息`时，会根据这个Listener实现的策略来决断事务TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl();// 构造事务消息的生产者TransactionMQProducer producer = new TransactionMQProducer("transactionProducer");producer.setNamesrvAddr("127.0.0.1:9876");// 事务回查最小并发数producer.setCheckThreadPoolMinSize(2);// 事务回查最大并发数producer.setCheckThreadPoolMaxSize(2);// 队列数producer.setCheckRequestHoldMax(2000);// 设置事务决断处理类
        producer.setTransactionCheckListener(transactionCheckListener);producer.start();// 本地事务的处理逻辑，相当于示例中检查Bob账户并扣钱的逻辑TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl();for (int i = 1; i <= 2; i++) {try {String tags = "transaction" + i;String keys = "KEY" + i;byte[] body = ("Hello RocketMQ " + i).getBytes();Message msg = new Message("topicTransaction", tags, keys, body);SendResult sendResult = producer.sendMessageInTransaction(msg, tranExecuter, null);System.out.println(sendResult);} catch (MQClientException e) {e.printStackTrace();}}producer.shutdown();}
}

package com.lynch.simple.demo;import java.util.List;import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;public class TransactionConsumer {public static void main(String[] args) throws InterruptedException, MQClientException {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("transactionConsumer");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.setConsumeMessageBatchMaxSize(10);/*** 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>* 如果非第一次启动，那么按照上次消费的位置继续消费*/consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);consumer.subscribe("topicTransaction", "*");consumer.registerMessageListener(new MessageListenerConcurrently() {public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {for (MessageExt msg : msgs) {try {System.out.println(msg + ",内容：" + new String(msg.getBody()));} catch (Exception e) {e.printStackTrace();return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
                    }}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
            }});consumer.start();System.out.println("transaction_Consumer Started.");}
}