rocketmq实现事务消息方案(代码片段)

public interface RocketMQLocalTransactionListener 
/**
- 发送prepare消息成功此方法被回调，该方法用于执行本地事务
- @param msg 回传的消息，利用transactionId即可获取到该消息的唯一Id
- @param arg 调用send方法时传递的参数，当send时候若有额外的参数可以传递到send方法中，这里能获取到
- @return 返回事务状态，COMMIT：提交 ROLLBACK：回滚 UNKNOW：回调
*/
RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg);
/**
- @param msg 通过获取transactionId来判断这条消息的本地事务执行状态
- @return 返回事务状态，COMMIT：提交 ROLLBACK：回滚 UNKNOW：回调
*/
RocketMQLocalTransactionState checkLocalTransaction(Message msg);

TransactionMQProducer producer = new TransactionMQProducer("ProducerGroup");
    producer.setNamesrvAddr("127.0.0.1:9876");
    producer.start();
    //设置TransactionListener实现
    producer.setTransactionListener(transactionListener）；
    //发送事务消息
    SendResult sendResult = producer.sendMessageInTransaction(msg, null);

package com.example.rocketmq.common;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Slf4j
@Service
public class SyncProducer 
 
    @Resource
    private RocketMQTemplate rocketMQTemplate;
 
    public TransactionSendResult sendSyncMessage(String msg, String topic, String tag)

        log.info("【发送消息】：", msg);

        //构建消息体
        JSONObject jsonObject = new JSONObject();

        jsonObject.put("message",msg);

        Message<String> message = MessageBuilder.withPayload(jsonObject.toJSONString()).build();

        TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction("tx_group", topic, message, tag);

        log.info("【发送状态】：", result.getLocalTransactionState());

        return result;

    
 

package com.example.rocketmq.common;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
@RocketMQTransactionListener(txProducerGroup = "tx_group")
public class SyncProducerListener implements RocketMQLocalTransactionListener 

    private ConcurrentHashMap<String, Object> localTrans = new ConcurrentHashMap<>();

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) 

        try 

            log.info("【本地业务执行完毕】 msg:, Object:", message, o);

            localTrans.put(message.getHeaders().getId()+"", message.getPayload());

            return RocketMQLocalTransactionState.COMMIT;

         catch (Exception e) 

            e.printStackTrace();

            log.error("【执行本地业务异常】 exception message:", e.getMessage());

            return RocketMQLocalTransactionState.ROLLBACK;

        

    
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) 

        log.info("【执行检查任务】");

        return RocketMQLocalTransactionState.UNKNOWN;

    

package com.example.rocketmq.common;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

@Component
@RocketMQMessageListener(topic = "TransactionTopic",consumerGroup = "spring_boot_consumer_group")
@Slf4j
public class TxmsgConsumer implements RocketMQListener<String> 

    @Override
    public void onMessage(String s) 

        log.info("开始消费消息:",s);

    

rocketmq.name-server=localhost:9876
rocketmq.producer.group = spring_boot_producer_group

       <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.0.3</version>
        </dependency>

package com.example.rocketmq.controller;


import com.example.rocketmq.common.SyncProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class TestController 


    @Autowired
    private SyncProducer syncProducer;

    @GetMapping("test")
    String test() 

        syncProducer.sendSyncMessage("我随便传的一条测试消息，内容保密","TransactionTopic","TransactionTAG");

        return "ok";