Spring boot 2.x 集成Rocketmq实现事物消息

         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    4.0.0com.xxx
    rocket
    1.0-SNAPSHOTorg.springframework.boot
        spring-boot-starter-parent
        2.0.5.RELEASEUTF-8UTF-81.8org.springframework.boot
            spring-boot-starter-web
        org.springframework.boot
            spring-boot-starter-test
            testorg.springframework.boot
            spring-boot-starter
        org.springframework.boot
            spring-boot-actuator-autoconfigure
        
            org.apache.rocketmq
            rocketmq-spring-boot-starter
              2.0.1org.projectlombok
            lombok
            1.16.14org.springframework.boot
                spring-boot-maven-plugin
            

package com.xxx.listener;

import com.alibaba.fastjson.JSON;
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;

@Slf4j
@RocketMQTransactionListener(txProducerGroup = "rocket")
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        System.out.println("本地事务和消息发送:" + JSON.toJSONString(message));

        return RocketMQLocalTransactionState.UNKNOWN;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        System.out.println("回查信息:" + JSON.toJSONString(message));

        return RocketMQLocalTransactionState.COMMIT;
    }
}

package com.xxx;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

@Slf4j
@SpringBootApplication
public class SpringBootRocketMqApplication {
    public static void main(String[] args) throws InterruptedException {
        ConfigurableApplicationContext context = SpringApplication.run(SpringBootRocketMqApplication.class, args);

        RocketMQTemplate template = context.getBean(RocketMQTemplate.class);
        while (true) {
            String msg = "demo msg test";
            log.info("开始发送消息："+msg);

            Message message = MessageBuilder.withPayload(msg).build();
            TransactionSendResult result = template.sendMessageInTransaction("rocket", "ts", message, null);
            log.info("消息发送响应信息："+result.toString());

            Thread.sleep(10);
        }
    }
}

package com.xxx.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@Component
@RocketMQMessageListener(topic = "ts", consumerGroup = "my-consumer-group")
public class ConsumerLifecycleListener implements RocketMQListener, RocketMQPushConsumerLifecycleListener {
    @Override
    public void onMessage(String s) {
        // 实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用
        log.info("实现RocketMQPushConsumerLifecycleListener监听器之后，此方法不调用");
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt messageExt : msgs) {
                    System.out.println("重试次数:" + messageExt.getReconsumeTimes());

                    System.out.println("接受到的消息:" + new String(messageExt.getBody()));
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}