/**
	 * 初始化生产者
	 * 
	 * @param rockterMQConfig
	 * @return
	 * @throws MQClientException
	 */
	@Bean
	public DefaultMQProducer defaultMQProducer(RockterMQConfig rockterMQConfig) throws MQClientException {
		TransactionMQProducer producer = new TransactionMQProducer(rockterMQConfig.getProducerGroup());
		producer.setNamesrvAddr(rockterMQConfig.getNamesrvAddr());
		producer.setInstanceName(rockterMQConfig.getInstanceName());
		producer.setTransactionListener( new TransactionListenerImpl() );
		producer.start();
		return producer;
	}

package com.xyebank.hzx.message.listener;

import java.util.concurrent.ConcurrentHashMap;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;


public class TransactionListenerImpl implements TransactionListener  {

	// 存储事务状态信息 key:事务id value：当前事务执行的状态
	private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

	// 执行本地事务
	@Override
	public LocalTransactionState executeLocalTransaction(Message message, Object o) {
		// 事务id
		String transactionId = message.getTransactionId();
		// 0:执行中，状态未知 1：执行成功 2：执行失败
		localTrans.put(transactionId, 0);
		// 业务执行，本地事务，service
		System.out.println("hello-demo-transaction");
		try {
			System.out.println("正在执行本地事务---");
			Thread.sleep( 60*1000 );
			System.out.println("本地事务执行成功---");
			localTrans.put(transactionId, 1);
		} catch (InterruptedException e) {
			e.printStackTrace();
			localTrans.put(transactionId, 2);
			return LocalTransactionState.ROLLBACK_MESSAGE;
		}
		return LocalTransactionState.COMMIT_MESSAGE;
	}

	// 消息回查
	@Override
	public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
		// 获取对应事务的状态信息
		String transactionId = messageExt.getTransactionId();
		// 获取对应事务id执行状态
		Integer status = localTrans.get(transactionId);
		// 消息回查
		System.out.println("消息回查---transactionId:" + transactionId + "状态:" + status);
		switch (status) {
		case 0:
			return LocalTransactionState.UNKNOW;
		case 1:
			return LocalTransactionState.COMMIT_MESSAGE;
		case 2:
			return LocalTransactionState.ROLLBACK_MESSAGE;
		}
		return LocalTransactionState.UNKNOW;
	}

}

	@Override
	@Transactional
	public void send(MessageInfo mqMessage) throws Exception  {
		
		if (mqMessage == null) {
			log.debug("无效的消息");
			return;
		}

		String body = com.alibaba.fastjson.JSONObject.toJSONString(mqMessage);
		Message message = new Message("log-topic", "user-tag", body.getBytes());
		producer.sendMessageInTransaction(message, "args2");
		log.debug("发送了消息:{}", body);
		
		
	}