springboot 分布式事务 springboot整合rocketmq实现分布式事务
你携秋月揽星河丶 人气:0想了解springboot整合rocketmq实现分布式事务的相关内容吗,你携秋月揽星河丶在本文为您仔细讲解springboot 分布式事务的相关知识和一些Code实例,欢迎阅读和指正,我们先划重点:springboot,分布式事务,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 回查代码测试
描述: 执行本地事务时添加异常,重启测试,发现消费者没有收到消息。
加载全部内容