本地任务消息组件：让数据库事务与外部消息推送（HTTP/RabbitMQ）达成最终一致性的通用组件方案

文档教程后端【免费下载链接】CodeGuide:books: 本代码库是作者小傅哥多年从事一线互联网 Java 开发的学习历程技术汇总旨在为大家提供一个清晰详细的学习教程侧重点更倾向编写Java核心内容。如果本仓库能为您提供帮助请给予支持(关注、点赞、分享)项目地址https://gitcode.com/gh_mirrors/code/CodeGuide点击查看免费下载本文基于 CodeGuide 仓库中的《本地任务消息组件》文档体系主文档 及同目录下的 6 节课程文档系统讲解该组件如何解决本地数据库事务 外部 MQ/HTTP 调用的最终一致性难题通过本地消息表在同一事务内落库、Spring Event 事件驱动异步通知、策略模式分发 HTTP/RabbitMQ 通道、门牌号分片的定时扫描补偿以及自定义注解 AOP 的轻量化接入方式。读完后你将掌握一套可直接借鉴到业务系统中的最终一致性组件设计范式与完整功能链路。一、问题背景为什么需要本地任务消息组件在业务功能开发中存在一个非常普遍的场景在完成一次数据库写事务的同时还需要向外部系统发送一条 MQ 消息或发起一次 HTTP 远程调用。例如订单落库后要通知结算系统、库存扣减后要同步给搜索服务。问题在于MQ 消息发送和 HTTP 调用无法与数据库写操作纳入同一个事务如果先写库、再发消息消息发送失败网络超时、服务宕机、流量洪峰、线程阻塞等业务数据已经落库但外部系统收不到通知数据不一致如果先发消息、再写库写库回滚后消息已经发出下游会处理一条幽灵消息分布式事务XA/TCC/Saga引入成本高、性能开销大多数业务场景并不需要这么重。在没有统一组件支持之前各业务系统只能各自实现自己建一张本地消息表、自己维护消息表的写入、自己写定时任务扫描补偿、自己处理 HTTP/MQ 的发送与重试——重复开发且难以维护正如 主文档 中所述为了完成业务流程的同时在发送一个 MQ 消息或者远程调用 HTTP 操作都需要自己写一个本地消息表之后还要维护消息表的扫描补偿。本地任务消息组件Local Task Message正是为了把这套通用能力从各业务系统中抽取出来凝练成一个可被上游系统以 jar 包方式引入的领域服务内核统一解决这一共性难题。二、产品概述组件解决什么问题、如何工作按照 主文档 的产品方案描述本地任务消息组件基于 Spring 框架能力实现通用功能内核便于集成到各类业务系统中核心工作链路如下事务内写库在业务系统的事务内完成业务数据写库的同时写入一条本地消息记录上游系统需在自身数据库中创建符合组件规范的本地消息表且与业务数据同库保证同一个事务事件驱动异步通知写入完成后组件同步推送 Spring 事件ApplicationEvent触发事务外的异步处理EventListenerAsync执行 MQ 消息发送或 HTTP 回调分片扫描补偿即使异步处理失败组件内置的本地消息表定时任务会持续检测并重试通知且支持自定义配置门牌号houseNumber多任务并行扫描提升扫描吞吐量确保消息最终一致性和业务流程的可靠执行。使用方式上组件提供两种接入途径用户可以选择注解方式对目标方法配置自定义注解LocalTaskMessage切面会自动获取入参入参需要为TaskMessageEntityCommand对象也可以是某个入参对象中携带该对象之后通过req.command配置也可以获取编程方式直接调用组件内核服务ILocalTaskMessageHandleService的acceptTaskMessage(taskMessageEntityCommand)方法受理任务消息。两种方式的好处是业务项目工程不再需要自己维护本地消息表的写入以及 MQ/HTTP 的处理和补偿逻辑全部由组件内核完成。三、技术架构DDD 分层与端口-适配器模式从 主文档 的技术架构章节可以明确组件的架构定位Local Task Message 组件是为解决本地数据库事务与外部 MQ/HTTP 调用一致性问题而设计的领域服务内核让上游业务系统不必在每个流程中做大量重复编码通过注解或直接调用组件内核服务即可完成消息通知操作它不是一个单纯的工具性功能而是一个具备完整领域功能的内核具备操作数据库表的能力接收 Spring Event 事件对接 MQ、HTTP 完成与外部的交互处理上游系统使用时只需要配置好对应的本地消息表一个事务下连的同一个库、引入组件并完成 yml 配置即可直接使用。在工程结构上组件采用 DDD 分层与端口-适配器模式清晰划分domain / infrastructure / trigger / config模块对应仓库 第2节课程文档 中的说明内核有点类似把一个业务项目中从上到下一整套流程被单独提取出来做成一个独立的小项目之后引入到上游使用方的系统就可以运行。各层职责在后续 6 节课程文档中均有体现模块职责课程文档佐证domain领域层任务消息领域服务按notifyType分发不同的通知操作http、mq第4节infrastructure基础设施层数据库访问原生 JDBC、HTTP 调用Retrofit2、MQ 推送RabbitTemplate第3节、第4节trigger触发层Spring Event 事件监听接收事件后调用领域层的通知服务方法第2节、第4节config配置层自定义注解的切面逻辑、ConfigurationProperties动态调度配置第6节本仓库CodeGuide为文档仓库组件的工程代码以课程配套项目形式逐步拉分支开发本节及以下各节的实现描述均以仓库中对应课程文档为准可通过 docs/md/project/local-task-message/ 目录下的各节文档继续深入。四、本地消息任务表设计与数据写入这是组件一致性的基石对应 第3节课程文档。1. 设计原则任务表由上游系统自行在数据库中配置组件不绑定任何特定库表命名引入组件的上游系统在自己数据库中建表之后调用组件时以同一个数据源DataSource对库表进行操作从而保证业务数据 任务消息记录在同一个本地事务中提交直接使用原生 JDBC 操作数据库不引入 MyBatis 框架。这样做的目的是避免上游系统使用组件时的版本兼容问题——最原始的 JDBC 方式兼容性最好数据访问以 DAO 封装落地完成插入、状态更新、分片条件查询、最小游标查询等操作。2. 关键 DAO 能力结合 主文档 的学习要点与 第5节课程文档 的补偿流程任务表的数据操作可以归纳为四类插入受理任务消息时与业务数据同事务写入一条待通知的任务记录状态更新HTTP/MQ 通知执行完成后无论成功还是失败都要回写任务记录状态分片条件查询按门牌号houseNumber分片条件拉取待处理任务列表最小游标查询根据条件获取满足条件的最小 id再以id 最小id limit x的方式批量拉取数据保证扫描推进有序、可续扫。五、Spring Event 事件消息解耦事务内写入与事务外通知对应 第2节课程文档本节完成了组件的事件驱动骨架构建本地任务消息组件的工程框架与对应的测试工程服务使用 Spring Event 事件消息完成行为触达的通知和监听用于后续处理外部 HTTP、MQ 的调用操作把本地消息组件构建成一个 jar 包让测试工程通过 pom 依赖方式引入使用。从事件链路上看组件使用ApplicationEventEventListenerAsync的组合实现解耦通知链路业务事务提交前任务消息已随业务数据落库事件被发布后监听器在事务之外异步执行外部通知。这一设计的关键收益是——即使通知动作失败也不会影响业务事务本身反之事务未提交的数据外部也不会提前感知。六、通知策略处理HTTP 与 RabbitMQ 双通道对应 第4节课程文档本节为组件增加了 HTTP 与 MQRabbitMQ两种通知通道采用策略模式完成可插拔的通知能力trigger 层监听事件消息的监听放在 trigger 层监听后调用领域层的通知服务方法领域层新增一个按通知类型分发的领域方法领域服务按notifyType分发如http、mq将来想扩展其他通知通道如 Kafka、gRPC只需在此处添加新的策略实现基础设施层完成具体调用HTTP使用Retrofit2底层OkHttp3框架统一封装 HTTP 网关掌握动态 URL、Header、Body 的组合与异常处理MQRabbitMQ 直接使用RabbitTemplate模板 push 消息。此外还有一个重要的工程细节RabbitMQ 事件发布采用可选依赖注入方式避免上游系统未配置 MQ 时因强依赖导致应用启动失败。也就是说组件支持只用 HTTP 不用 MQ或只用 MQ 不用 HTTP的部署形态。七、动态任务补偿处理门牌号分片扫描保证最终一致对应 第5节课程文档这是组件最终一致承诺的兜底机制先更新状态再定时补偿无论 MQ 发送还是 HTTP 调用都有可能失败网络超时、服务宕机、线程阻塞、流量洪峰等因此在完成 MQ/HTTP 处理后首先更新数据库任务表状态成功或失败之后再由定时扫描任务做补偿处理门牌号houseNumber分片扫描为了提高整体扫描效率设计了门牌号机制——可以配置多个定时任务每个任务只扫描自己门牌号范围内的记录多个任务并行扫描从而提升扫描吞吐量游标推进拉取扫描库表时先根据条件获取一个符合条件的最小 id再以id 最小id limit x获取数据列表实现高效的拉取与顺序处理补偿重复与幂等由于从 Spring Event 接收消息 → 执行通知http/mq→ 更新数据库这些步骤都不在同一个事务中即从事务——业务数据 任务表数据写入——往后都是有可能失败的补偿就可能出现重复比如 HTTP 被重复调用一次、MQ 被重复发送一次。因此课程文档明确强调业务方对接这些通知时一定要做幂等操作比如以 OrderId 做唯一索引处理。八、切面拦截任务操作LocalTaskMessage 注解的轻量化接入对应 第6节课程文档。由于这是一个通用组件项目使用方式必须足够轻量因此对编程式调用handleService.acceptTaskMessage(taskMessageEntityCommand)的编码方式提供了更优雅的注解方案添加自定义注解LocalTaskMessage并在 config 配置层编写切面逻辑核心是获取配置了该注解的方法入参从中拿到TaskMessageEntityCommand任务消息对象事务边界处理切面会判断当前是否已有事务操作——如果没有则开启一个新事务如果有则使用同一个事务完成数据库表数据的插入随后推送 Spring Event 事件消息后续流程与编程式调用完全一致。切面方式的优点在于更优雅简洁用户不需要自己维护调用关系同时结合事务边界的统一处理保证了业务数据 消息记录的原子写入。九、配置驱动多任务组动态调度与线程池化管理结合 主文档 的学习要点组件的调度与配置体系具备以下能力ConfigurationProperties驱动的多任务组动态调度配置通过 yml 配置即可定义多组扫描任务每组绑定自己的门牌号houseNumber分片范围两种触发方式支持cron固定时间表达式与fixedDelay固定延迟两种触发方式批次大小 limit 可配每轮扫描拉取的任务数量通过 limit 参数控制配合最小 id 游标 limit策略避免单次拉取过多造成压力ThreadPoolTaskScheduler线程池化调度管理合理设置线程名与池大小提升任务调度的可观测性与稳定性线程名便于日志定位池大小决定并发调度能力。此外组件还通过示例命令对象TaskMessageEntityCommand的构建与调用展示了入参约定、枚举策略通知类型枚举与配置对象之间的协作方式整体上也要求使用者具备异常、日志与枚举的综合使用能力建立稳定的错误处理机制。十、能力清单小结以下是从 主文档 完整继承的能学到啥清单也是本组件覆盖的技术面全景【架构】掌握 DDD 分层与端口-适配器模式清晰划分 domain/infrastructure/trigger/config 模块提升可维护性与扩展性【后端】学习注解 AOP 方式受理任务消息结合事务边界进行统一处理理解LocalTaskMessage与切面配合的落地实践【后端】掌握本地消息表设计与分片扫描策略按门牌号 houseNumber 分片实现高效拉取与顺序处理提升系统可靠性【后端】熟悉 Spring Event 事件驱动与异步消费使用ApplicationEventEventListenerAsync实现解耦通知链路【后端】实践策略模式实现可插拔通知能力支持 HTTP 与 RabbitMQ 两种通知通道并在成功/失败时更新任务状态【后端】熟练使用 OkHttp3 与 Retrofit2 统一封装 HTTP 网关掌握动态 URL、Header、Body 的组合与异常处理【后端】了解 RabbitMQ 事件发布的可选依赖注入方式避免未配置 MQ 时的强依赖导致应用启动失败【配置】掌握ConfigurationProperties驱动的多任务组动态调度配置支持 cron 与 fixedDelay 两种触发方式并可配置批次大小 limit【运维】学习ThreadPoolTaskScheduler的线程池化调度管理合理设置线程名与池大小提升任务调度的可观测性与稳定性【数据】掌握原生 JDBC 访问与 DAO 封装完成插入、状态更新、分片条件查询、最小游标查询等落地实现【测试】通过示例命令对象TaskMessageEntityCommand的构建与调用理解入参约定、枚举策略与配置对象的协作【实践】提升异常、日志与枚举的综合使用能力建立稳定的错误处理。十一、延伸阅读仓库中的配套课程文档本组件的完整开发过程在 CodeGuide 仓库中以 6 节课程文档逐步展开建议按顺序阅读以还原从需求分析到注解拦截的完整实现链路第1节组件需求分析——聚焦数据库事务 对外发送消息MQ/发起 HTTP 调用的最终一致性问题提炼通用技术解决方案第2节SpringEvent事件消息——搭建组件工程框架用 Spring Event 完成事件发布与监听把组件构建成 jar 供测试工程 pom 引入第3节任务表设计和数据写入——设计通用本地消息任务表引入 DataSource 数据源以原生 JDBC 完成插入处理规避 MyBatis 版本兼容问题第4节通知策略处理(HTTPMQ)——接收 Spring Event 监听后以通知行为策略完成 HTTP 远程调用Retrofit2和 MQ 消息推送RabbitTemplate第5节动态任务补偿处理——通知完成后更新任务表状态定时扫描任务按门牌号分片补偿并强调下游幂等设计第6节切面拦截任务操作——增加自定义注解config 层编写切面获取入参中的任务消息对象结合现有事务或新事务完成数据插入并推送 Spring Event。十二、总结本地任务消息组件给出的是一套本地消息表经典模式的组件化落地一致性边界清晰——业务数据与消息记录同库同事务写入事务提交是消息一定会被投递的唯一起点通知链路解耦——Spring Event 事件驱动 异步消费通知动作的失败不影响业务主流程通道可插拔——策略模式分发 HTTPRetrofit2/OkHttp3与 RabbitMQRabbitTemplate两种通道扩展新通道只需增加策略实现RabbitMQ 可选注入避免强依赖补偿可水平扩展——门牌号分片 多任务组并行扫描 最小 id 游标 limit 批次控制保证扫描吞吐量与顺序性接入足够轻量——LocalTaskMessage注解切面自动解析入参、自动识别事务边界编程式调用作为兜底方案工程兼容性优先——原生 JDBC 而非 ORM 框架、jar 包内核引入、yml 配置驱动让上游系统建表 引包 配置三步即可使用。这套设计不依赖任何重量级中间件仅用 Spring 事件、JDBC、Retrofit2 与 RabbitTemplate 即可完成最终一致性的完整闭环非常适合作为业务团队自研业务型通用组件的参考范式。赞分享文档教程后端【免费下载链接】CodeGuide:books: 本代码库是作者小傅哥多年从事一线互联网 Java 开发的学习历程技术汇总旨在为大家提供一个清晰详细的学习教程侧重点更倾向编写Java核心内容。如果本仓库能为您提供帮助请给予支持(关注、点赞、分享)项目地址https://gitcode.com/gh_mirrors/code/CodeGuide点击查看免费下载相关推荐my-tv 我的电视装完就能看的电视直播软件4 个遥控器按键搞定换台my tv 我的电视装完就能看的电视直播软件4 个遥控器按键搞定换台 晚上回家想把老电视当直播入口却发现市面上的电视直播软件要么绑会员、要么广告铺满界面音视频直播分布式事务完全指南XA、TCC、Saga、本地消息表与可靠消息最终一致性方案对比doocs/advanced-java分布式事务完全指南XA、TCC、Saga、本地消息表与可靠消息最终一致性方案对比doocs/advanced java 分布式事务是微服务与分布式系统面试文档教程后端gh_mirrors/ps/psr7与Doctrine集成数据库事务中的HTTP消息gh_mirrors/ps/psr7与Doctrine集成数据库事务中的HTTP消息 在Web应用开发中处理HTTP请求Request和响应Respo后端上一篇RuoYi-Cloud微服务权限系统从零到一的完整企业级开发指南 下一篇bup社区活动日历参与线上meetup与黑客马拉松创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考