springboot整合rocketmq实现分布式事务

更新时间：2021年05月30日 09:29:01 作者：你携秋月揽星河丶

大多数情况下很多公司是使用消息队列的方式实现分布式事务。本篇文章重点讲解springboot环境下整合rocketmq实现分布式事务，感兴趣的可以了解一下

1 执行流程

在这里插入图片描述

(1) 发送方向 MQ 服务端发送消息。
(2) MQ Server 将消息持久化成功之后，向发送方 ACK 确认消息已经发送成功，此时消息为半消息。
(3) 发送方开始执行本地事务逻辑。
(4) 发送方根据本地事务执行结果向 MQ Server 提交二次确认（Commit 或是 Rollback），MQ Server 收到Commit 状态则将半消息标记为可投递，订阅方最终将收到该消息；MQ Server 收到 Rollback 状态则删除半消息，订阅方将不会接受该消息。
(5) 在断网或者是应用重启的特殊情况下，上述步骤4提交的二次确认最终未到达 MQ Server，经过固定时间后MQ Server 将对该消息发起消息回查。
(6) 发送方收到消息回查后，需要检查对应消息的本地事务执行的最终结果。
(7) 发送方根据检查得到的本地事务的最终状态再次提交二次确认，MQ Server 仍按照步骤4对半消息进行操作。

2 工程

在这里插入图片描述

2.1 pom

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.0.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.71</version>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-collections4</artifactId>
            <version>4.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-logging</artifactId>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-spring-boot-starter -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.0.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.3.2</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>2.3.0.RELEASE</version>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

2.2 application.yml

rocketmq:
  name-server: 192.168.38.50:9876
  producer:
    group: transcation-group

2.3 TransactionListenerImpl

@RocketMQTransactionListener(txProducerGroup = "transaction-producer-group")
@Slf4j
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    private static Map<String, RocketMQLocalTransactionState> STATE_MAP = new HashMap<>();

    /**
     *  执行业务逻辑
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        String transId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        try {
            System.out.println("用户A账户减500元.");
            System.out.println("用户B账户加500元.");
            STATE_MAP.put(transId, RocketMQLocalTransactionState.COMMIT);
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            e.printStackTrace();
        }

        STATE_MAP.put(transId, RocketMQLocalTransactionState.ROLLBACK);
        return RocketMQLocalTransactionState.UNKNOWN;

    }

    /**
     * 回查
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        String transId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        log.info("回查消息 -> transId ={} , state = {}", transId, STATE_MAP.get(transId));
        return STATE_MAP.get(transId);
    }
}

2.4 SpringTransactionProducer

@Component
@Slf4j
public class SpringTransactionProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送消息
     *
     */
    public void sendMsg(String topic, String msg) {
        Message<String> message = MessageBuilder.withPayload(msg).build();
        this.rocketMQTemplate.sendMessageInTransaction("transaction-producer-group", topic, message, null);
        log.info("发送成功");
    }
}

2.5 SpringTxConsumer

@Component
@RocketMQMessageListener(topic = "pay_topic",
        consumerGroup = "transaction-consumer-group",
        selectorExpression = "*")
@Slf4j
public class SpringTxConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String msg) {
        log.info("接收到消息 -> {}", msg);
    }
}

2.6 ProducerController

@RestController
@RequestMapping("/producer")
public class ProducerController {

    @Autowired
    private SpringTransactionProducer springTransactionProducer;

    @GetMapping("/sendMsg")
    public String sendMsg() {
        springTransactionProducer.sendMsg("pay_topic", "用户A账户减500元,用户B账户加500元。");
        return "发送成功";
    }

}

2.7 RocketApplication

@SpringBootApplication
public class RocketApplication {

    public static void main(String[] args) {
        SpringApplication.run(RocketApplication.class);
    }

}

3 测试

3.1 正常消费测试

描述： 正常启动及可。

在这里插入图片描述

3.2 回查代码测试

描述： 执行本地事务时添加异常，重启测试，发现消费者没有收到消息。

在这里插入图片描述

到此这篇关于springboot整合rocketmq实现分布式事务的文章就介绍到这了,更多相关springboot 分布式事务内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

Java常用类之System类的使用指南
System类代表系统，系统级的很多属性和控制方法都放置在该类的内部。该类位于java.lang包。本文将通过示例为大家详细讲讲System类的使用，需要的可以参考一下
2022-07-07
SpringBoot数据层处理方案精讲
这篇文章主要介绍了SpringBoot数据层技术的解析，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习吧
2022-10-10
swagger的请求参数不显示,@Apimodel的坑点及解决
这篇文章主要介绍了swagger的请求参数不显示,@Apimodel的坑点及解决方案,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2023-11-11
springboot max-http-header-size最大长度的那些事及JVM调优方式
这篇文章主要介绍了springboot max-http-header-size最大长度的那些事及JVM调优方式，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2022-09-09
SpringCloud_Eureka服务注册与发现基础及构建步骤
Eureka服务注册中心，主要用于提供服务注册功能，当微服务启动时，会将自己的服务注册到Eureka Server，这篇文章主要介绍了SpringCloud中Eureka的配置及详细使用,需要的朋友可以参考下
2023-01-01
Java中的迭代器和foreach原理
这篇文章主要介绍了Java中的迭代器和foreach原理，帮助大家更好的理解和学习Java，感兴趣的朋友可以了解下
2020-09-09
Java后台接口开发初步实战教程
下面小编就为大家分享一篇 Java后台接口开发初步实战教程，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2018-01-01
java虚拟机原理:类加载过程详解
这篇文章主要介绍了Java中类加载过程全面解析，具有一定参考价值，需要的朋友可以了解下，希望能够给你带来帮助
2021-09-09
Java8 Stream flatmap中间操作用法解析
这篇文章主要介绍了Java8 Stream flatmap中间操作用法解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2020-07-07
详解SimpleDateFormat的线程安全问题与解决方案
这篇文章主要介绍了SimpleDateFormat的线程安全问题与解决方案,非常不错，具有参考借鉴价值，需要的朋友可以参考下
2017-03-03