【架构篇】一个基于 RabbitMQ 的可复用的分布式事务消息架构方案!
来源 | 黄哥
前提
分布式事务是微服务实践中一个比较棘手的问题,在笔者所实施的微服务实践方案中,都采用了折中或者规避强一致性的方案。参考Ebay多年前提出的本地消息表方案,基于RabbitMQ和MySQL(JDBC)做了轻量级的封装,实现了低入侵性的事务消息模块。本文的内容就是详细分析整个方案的设计思路和实施。环境依赖如下:
JDK1.8+
spring-boot-start-web:2.x.x
spring-boot-start-jdbc:2.x.x
spring-boot-start-amqp:2.x.x
HikariCP:3.x.x(spring-boot-start-jdbc自带)
mysql-connector-java:5.1.48
redisson:3.12.1
方案设计思路
事务消息原则上只适合弱一致性(或者说最终一致性)的场景,常见的弱一致性场景如:
用户服务完成了注册动作,向短信服务推送一条营销相关的消息。
信贷体系中,订单服务保存订单完毕,向审批服务推送一条待审批的订单记录信息。
……
强一致性的场景一般不应该选用事务消息。
一般情况下,要求强一致性说明要严格同步,也就是所有操作必须同时成功或者同时失败,这样就会引入同步带来的额外消耗。
如果一个事务消息模块设计合理,补偿、查询、监控等等功能都完毕,由于系统交互是异步的,整体吞吐要比严格同步高。在笔者负责的业务系统中基于事务消息使用还定制了一条基本原则:消息内容正确的前提下,消费方出现异常需要自理。
简单来说就是:上游保证了自身的业务正确性,成功推送了正确的消息到RabbitMQ就认为上游义务已经结束。
为了降低代码的入侵性,事务消息需要借助Spring的编程式事务或者声明式事务。编程式事务一般依赖于TransactionTemplate,而声明式事务依托于AOP模块,依赖于注解@Transactional。
接着需要自定义一个事务消息功能模块,新增一个事务消息记录表(其实就是本地消息表),用于保存每一条需要发送的消息记录。事务消息功能模块的主要功能是:
保存消息记录。
推送消息到RabbitMQ服务端。
消息记录的查询、补偿推送等等。
事务消息的补偿
在事务方法内,保存事务消息的时候,标记消息记录推送状态为处理中。 事务同步器接口TransactionSynchronization的afterCommit()方法的实现中,推送对应的消息到RabbitMQ,然后更变事务消息记录的状态为推送成功。
表设计
CREATE TABLE `t_transactional_message`
(
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
edit_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
creator VARCHAR(20) NOT NULL DEFAULT 'admin',
editor VARCHAR(20) NOT NULL DEFAULT 'admin',
deleted TINYINT NOT NULL DEFAULT 0,
current_retry_times TINYINT NOT NULL DEFAULT 0 COMMENT '当前重试次数',
max_retry_times TINYINT NOT NULL DEFAULT 5 COMMENT '最大重试次数',
queue_name VARCHAR(255) NOT NULL COMMENT '队列名',
exchange_name VARCHAR(255) NOT NULL COMMENT '交换器名',
exchange_type VARCHAR(8) NOT NULL COMMENT '交换类型',
routing_key VARCHAR(255) COMMENT '路由键',
business_module VARCHAR(32) NOT NULL COMMENT '业务模块',
business_key VARCHAR(255) NOT NULL COMMENT '业务键',
next_schedule_time DATETIME NOT NULL COMMENT '下一次调度时间',
message_status TINYINT NOT NULL DEFAULT 0 COMMENT '消息状态',
init_backoff BIGINT UNSIGNED NOT NULL DEFAULT 10 COMMENT '退避初始化值,单位为秒',
backoff_factor TINYINT NOT NULL DEFAULT 2 COMMENT '退避因子(也就是指数)',
INDEX idx_queue_name (queue_name),
INDEX idx_create_time (create_time),
INDEX idx_next_schedule_time (next_schedule_time),
INDEX idx_business_key (business_key)
) COMMENT '事务消息表';
CREATE TABLE `t_transactional_message_content`
(
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
message_id BIGINT UNSIGNED NOT NULL COMMENT '事务消息记录ID',
content TEXT COMMENT '消息内容'
) COMMENT '事务消息内容表';
小结
架构师交流群
「顶级架构师」建立了读者架构师交流群,大家可以添加小编微信进行加群
扫描添加好友邀你进架构师群,加我时注明【姓名+公司+职位】
版权申明:内容来源网络,版权归原作者所有。如有侵权烦请告知,我们会立即删除并表示歉意。谢谢。
猜你还想看
【实战篇】ElasticSearch的基本概念和集群分布式底层实现
【经验篇】Java 虚拟机 9 : Java 类加载机制!
【实战篇】Java 生鲜电商平台 - API 接口设计之 token、timestamp、sign 具体架构与实现
【实战篇】彻底理解cookie、session、token
长按识别图片二维码关注,订阅更多精彩
顶级架构师,企业架构、系统架构、网站架构、大规模分布式架构、高可用架构等架构讨论,以及结合互联网技术的架构调整。欢迎有想法、乐于分享的架构师交流学习