RocketMQ笔记（七）SpringBoot整合RocketMQ发送事务消息

- 一、简介
- - 1.1、流程图
  - 1.2、事务消息流程介绍
- 二、Maven依赖
- 三、生产者
- - 3.1、application配置
  - 3.2、员工表
  - 3.3、实体
  - 3.4、持久层
  - 3.5、监听器
- 四、测试
- - 4.1、普通消息
  - 4.2、事务消息
  - - 4.2.1、消费者
    - 4.2.2、正常提交
    - 4.2.3、异常提交
- 五、其他
- - 5.1、接口说明
  - 5.2、checkLocalTransaction不回调

一、简介

在之前的文章中，我讲过了，同步发送单条消息，异步发送单条消息，发送单向消息，发送顺序消息，以及批量发送消息，延迟消息。今天说下发送事务消息。

1.1、流程图

事务消息交互流程如下图所示。
在这里插入图片描述

1.2、事务消息流程介绍

事务消息交互流程如下图所示。

生产者将消息发送至Apache RocketMQ服务端
Apache RocketMQ服务端将消息持久化成功之后，向生产者返回Ack确认消息已经发送成功，此时消息被标记为"暂不能投递"，这种状态下的消息即为半事务消息
生产者开始执行本地事务逻辑
生产者根据本地事务执行结果向服务端提交二次确认结果（Commit或是Rollback），服务端收到确认结果后处理逻辑如下：

二次确认结果为Commit：服务端将半事务消息标记为可投递，并投递给消费者
二次确认结果为Rollback：服务端将回滚事务，不会将半事务消息投递给消费者

在断网或者是生产者应用重启的特殊情况下，若服务端未收到发送者提交的二次确认结果，或服务端收到的二次确认结果为Unknown未知状态，经过固定时间后，服务端将对消息生产者即生产者集群中任一生产者实例发起消息回查。说明服务端回查的间隔时间和最大回查次数
生产者收到消息回查后，需要检查对应消息的本地事务执行的最终结果
生产者根据检查到的本地事务的最终状态再次提交二次确认，服务端仍按照步骤4对半事务消息进行处理

二、Maven依赖

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"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"><parent><artifactId>rocketmq</artifactId><groupId>com.alian</groupId><version>1.0.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>06-send-transactional-message</artifactId><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-jpa</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>druid-spring-boot-starter</artifactId><version>1.2.6</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.26</version><scope>runtime</scope></dependency><dependency><groupId>com.alian</groupId><artifactId>common-rocketmq-dto</artifactId><version>1.0.0-SNAPSHOT</version></dependency></dependencies></project>

父工程已经在我上一篇文章里，通用公共包也在我上一篇文章里有说明，包括消费者。具体参考：RocketMQ笔记（一）SpringBoot整合RocketMQ发送同步消息

三、生产者

3.1、application配置

application.properties

server.port=8006spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.type= com.alibaba.druid.pool.DruidDataSource
spring.datasource.username=test
spring.datasource.password=Alian!@34
spring.datasource.url=jdbc:mysql://192.168.0.139:3306/test?characterEncoding=utf8&useUnicode=true&useSSL=false&zeroDateTimeBehavior=convertToNull&autoReconnect=true&allowMultiQueries=true&failOverReadOnly=false&connectTimeout=6000&maxReconnects=5
spring.datasource.initialSize=5
spring.datasource.minIdle= 5
spring.datasource.maxActive=20# rocketmq地址
rocketmq.name-server=192.168.0.234:9876
# 默认的生产者组
rocketmq.producer.group=transactional_group
# 发送同步消息超时时间
rocketmq.producer.send-message-timeout=3000
# 用于设置在消息发送失败后，生产者是否尝试切换到下一个服务器。设置为 true 表示启用，在发送失败时尝试切换到下一个服务器
rocketmq.producer.retry-next-server=true
# 用于指定消息发送失败时的重试次数
rocketmq.producer.retry-times-when-send-failed=3
# 设置消息压缩的阈值,为0表示禁用消息体的压缩
rocketmq.producer.compress-message-body-threshold=0

在 RocketMQ 中，RocketMQTemplate的syncSend方法，它允许你批量发送同步消息，主要参数：

topic：主题
Message：消息内容
timeout：发送超时时间
delayLevel：延迟级别

3.2、员工表

员工实体

CREATE TABLE `employee` (`id` int(10) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',`code` varchar(8) NOT NULL DEFAULT '' COMMENT '编号',`emp_name` varchar(20) NOT NULL DEFAULT '' COMMENT '姓名',`age` int(2) NOT NULL DEFAULT '0' COMMENT '年龄',`salary` double(8,2) NOT NULL DEFAULT '0.00' COMMENT '工资',`department` varchar(20) NOT NULL DEFAULT '' COMMENT '部门',`hire_date` date NOT NULL DEFAULT '1970-07-01' COMMENT '入职时间',PRIMARY KEY (`id`),UNIQUE KEY `code_UNIQUE` (`code`),KEY `idx_code` (`code`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4

3.3、实体

实体类

@Data
@Entity
public class Employee implements Serializable {private static final long serialVersionUID = 1L;/*** 员工编号*/@Id@Column(name = "id")@GeneratedValue(strategy = GenerationType.IDENTITY)private Integer id;/*** 员工编号*/@Column(name = "code")private String code;/*** 员工姓名*/@Column(name = "emp_name")private String name;/*** 员工年龄*/@Column(name = "age")private int age;/*** 工资*/@Column(name = "salary")private double salary = 0.00;/*** 部门*/@Column(name = "department")private String department;/*** 入职时间*/@Column(name = "hire_date")private LocalDate hireDate;}

3.4、持久层

持久层

public interface EmployeeRepository extends PagingAndSortingRepository<Employee, Integer> {Employee findByCode(String code);
}

3.5、监听器

首先我们需要配置自定义的 RocketMQTemplate，最重要的是设置 TransactionMQProducer 的生产者组名称，这里是custom_transactional_group。

配置扩展的 RocketMQTemplate的类，并指定了 RocketMQ 生产者组的名称为custom_transactional_group。这个扩展的 RocketMQTemplate类可以用于发送事务消息以及其他类型的消息。

package com.alian.transactional.listener;import com.alian.transactional.domain.Employee;
import com.alian.transactional.repository.EmployeeRepository;
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.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;@Slf4j
@RocketMQTransactionListener(rocketMQTemplateBeanName = "rocketMQTemplate")
public class EmployeeTransactionListener implements RocketMQLocalTransactionListener {@Autowiredprivate EmployeeRepository employeeRepository;@Overridepublic RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {try {log.info("事务消息Headers为:{}", message.getHeaders());String payload = new String((byte[]) message.getPayload());log.info("事务消息为:{}", payload);Employee employee = JSON.parseObject(payload, Employee.class);employee.setId(null);Employee save = employeeRepository.save(employee);if (save.getId() != null) {log.info("保存员工成功:{}", save.getId());return RocketMQLocalTransactionState.COMMIT;}log.info("保存员工失败:{}", save.getId());} catch (Exception e) {log.error("发送事务消息异常:{}", e.getMessage());return RocketMQLocalTransactionState.UNKNOWN;}return RocketMQLocalTransactionState.ROLLBACK;}@Overridepublic RocketMQLocalTransactionState checkLocalTransaction(Message message) {String id = message.getHeaders().get("rocketmq_KEYS").toString();log.info("事务消息key为:{}", id);Employee employee = employeeRepository.findByCode(id);if (employee == null) {return RocketMQLocalTransactionState.ROLLBACK;}return RocketMQLocalTransactionState.COMMIT;}}

四、测试

4.1、普通消息

@Slf4j
@SpringBootTest
public class SendTransactionalMessageTest {@Autowiredprivate RocketMQTemplate rocketMQTemplate;@Testpublic void syncSendStringMessage() {String topic = "string_message_topic";String message = "我是一条同步文本消息:syncSendStringMessage";SendResult sendResult = rocketMQTemplate.syncSend(topic, message);log.info("同步发送返回的结果:{}", sendResult);}@AfterEachpublic void waiting() {try {Thread.sleep(10000L);} catch (InterruptedException e) {e.printStackTrace();}}}

运行结果：

同步发送返回的结果:SendResult [sendStatus=SEND_OK, msgId=7F000001109818B4AAC24B519EC40000, offsetMsgId=C0A800EA00002A9F000000000000371F, messageQueue=MessageQueue [topic=string_message_topic, brokerName=rocketmq, queueId=2], queueOffset=1]字符串消费者接收到的消息: 我是一条同步文本消息:syncSendStringMessage

此处我们还是能正常发送普通消息。

4.2、事务消息

4.2.1、消费者

首先我们要在之前的消费者，增加一个消费监听

@Slf4j
@Component
@RocketMQMessageListener(topic = "transaction_message_topic", consumerGroup = "CONCURRENT_GROUP_TRANSACTION")
public class TransactionMessageConsumer implements RocketMQListener<JSONObject> {@Overridepublic void onMessage(JSONObject json) {log.info("接收到事务消息:{}", json);}
}

4.2.2、正常提交

接着，我们先测试正常发送消息（发送半事务消息，本地保存记录，成功提交事务）。

@Slf4j
@SpringBootTest
public class SendTransactionalMessageTest {@Autowiredprivate RocketMQTemplate rocketMQTemplate;@Testpublic void sendMessageInTransaction() {// 计算过期时间戳String uuid = UUID.randomUUID().toString().replace("-", "");String topic = "transaction_message_topic";String code = "BAT10015";JSONObject json = new JSONObject();json.put("code", code);json.put("name", "张若尘");json.put("age", "25");json.put("salary", "15000");json.put("department", "测试部");json.put("hireDate", "2020-05-21");Message<JSONObject> rocketMessage = MessageBuilder.withPayload(json)// 根据需要设置.setHeader(RocketMQHeaders.TRANSACTION_ID, uuid)// 设置一个业务的key,以便事务回查时使用.setHeader(RocketMQHeaders.KEYS, code).build();// 发送事务消息TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(topic, rocketMessage, null);log.info("【发送状态】：{}", transactionSendResult.getLocalTransactionState());}@AfterEachpublic void waiting() {try {// 休眠时间3分钟Thread.sleep(180000L);} catch (InterruptedException e) {e.printStackTrace();}}}

生产者运行结果：

事务消息Headers为:{rocketmq_TOPIC=transaction_message_topic, rocketmq_FLAG=0, __transactionId__=7F000001326018B4AAC24B72107D0000, rocketmq_TRANSACTION_ID=7F000001326018B4AAC24B72107D0000, rocketmq_KEYS=BAT10015, id=c98dca25-50b7-9c9a-fc5f-b7ab7bea18eb, TRANSACTION_ID=0055a562d5af4db0a19f2939124a82f0, contentType=application/json, timestamp=1710488166582}
事务消息为:{"hireDate":"2020-05-21","code":"BAT10015","name":"张若尘","salary":"15000","department":"测试部","age":"25"}
保存员工成功:25
【发送状态】：COMMIT_MESSAGE

消费者运行结果：

接收到事务消息:{"hireDate":"2020-05-21","code":"BAT10015","name":"张若尘","salary":"15000","department":"测试部","age":"25"}

从上面的监听器保存成功记录后，返回了 RocketMQLocalTransactionState.COMMIT，所以我们的事务是正常提交的，半事务消息也被推送到消息者队列了。

4.2.3、异常提交

因为，我们数据库表设计时，code字段的长度是8位，我们就插入一个大于8位的值，然后抛出一个异常，然后rocketmq 会进行回查

@Slf4j
@SpringBootTest
public class SendTransactionalMessageTest {@Autowiredprivate RocketMQTemplate rocketMQTemplate;@Testpublic void sendMessageInTransaction() {// 计算过期时间戳String uuid = UUID.randomUUID().toString().replace("-", "");String topic = "transaction_message_topic";String code = "BAT8888888";JSONObject json = new JSONObject();json.put("code", code);json.put("name", "张若尘");json.put("age", "25");json.put("salary", "15000");json.put("department", "测试部");json.put("hireDate", "2020-05-21");Message<JSONObject> rocketMessage = MessageBuilder.withPayload(json)// 根据需要设置.setHeader(RocketMQHeaders.TRANSACTION_ID, uuid)// 设置一个业务的key,以便事务回查时使用.setHeader(RocketMQHeaders.KEYS, code).build();// 发送事务消息TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(topic, rocketMessage, null);log.info("【发送状态】：{}", transactionSendResult.getLocalTransactionState());}@AfterEachpublic void waiting() {try {// 休眠时间3分钟Thread.sleep(180000L);} catch (InterruptedException e) {e.printStackTrace();}}}

运行结果：

事务消息Headers为:{rocketmq_TOPIC=transaction_message_topic, rocketmq_FLAG=0, __transactionId__=7F00000107B018B4AAC24B6E46F00000, rocketmq_TRANSACTION_ID=7F00000107B018B4AAC24B6E46F00000, rocketmq_KEYS=BAT888888, id=9d148765-9f66-2242-af17-0591c8378c7a, TRANSACTION_ID=26930ad65e7c40738aa24c6cb5c3da61, contentType=application/json, timestamp=1710487918377}
事务消息为:{"hireDate":"2020-05-25","code":"BAT888888","name":"Alian","salary":"35000","department":"研发部","age":"28"}
SQL Error: 1406, SQLState: 22001
Data truncation: Data too long for column 'code' at row 1
发送事务消息异常:could not execute statement; SQL [n/a]; nested exception is org.hibernate.exception.DataException: could not execute statement
【发送状态】：UNKNOW事务消息key为:BAT888888
本地事务失败，删除消息:BAT888888

从上面的监听器保存异常（Data too long for column ‘code’），返回了 RocketMQLocalTransactionState.UNKNOWN（只是模拟返回），就不知道本地事务到底是成功还是失败，所以需要进行事务回查，也就是要调用：checkLocalTransaction放，检查本地是否正常，因为不存在记录，我们返回RocketMQLocalTransactionState.ROLLBACK，半事务消息就被删除了。

五、其他

5.1、接口说明

在springBoot整合中，实现的接口是RocketMQLocalTransactionListener接口，而不是TransactionListener。其中executeLocalTransaction 是半事务消息发送成功后，执行本地事务的方法，具体执行完本地事务后，可以在该方法中返回以下三种状态：

RocketMQLocalTransactionState.COMMIT：提交事务，允许消费者消费该消息
RocketMQLocalTransactionState.ROLLBACK：回滚事务，消息将被丢弃不允许消费。
RocketMQLocalTransactionState.UNKNOWN：暂时无法判断状态，等待固定时间以后Broker端根据回查规则向生产者进行消息回查。

checkLocalTransaction是由于二次确认消息没有收到，Broker端回查事务状态的方法。回查规则：本地事务执行完成后，若Broker端收到的本地事务返回状态为RocketMQLocalTransactionState.UNKNOWN，或生产者应用退出导致本地事务未提交任何状态。则Broker端会向消息生产者发起事务回查，第一次回查后仍未获取到事务状态，则之后每隔一段时间会再次回查。

5.2、checkLocalTransaction不回调

一般来说的原因：

在springboot整合rocketmq，和直接使用rocketmq实现的接口是不一样的，具体看上一小点
编码错误，不会模拟情景，rocketmq未收到半事务提交的结果（Commit或是Rollback），才会进行回查
最大的一个可能是，你使用springboot整合rocketmq时的版本和服务端的版本不一致（我遇到过），比如我这里整合版本是2.2.3，对应的rocketmq的版本是5.0.0，所以我的服务端，肯定也安装一个5.0的版本。如果服务端是4.x，就会出现消息转化报错，从而无法回调，具体可以看broker的日志
参数配置，比如（transactionCheckEnable=true、transactionCheckInterval=15000），是不是配置关闭，或者检查时间过长
测试不用心，比如写了单元测试，测试完，程序就结束了，导致服务端无法检查，所以我的测试程序，都休眠了一会