业务场景

1. 场景一：用户注册后，异步发送积分。如何保证用户注册后，积分最终一定能发送成功呢？
   

2. 场景二：支付扣款业务，如何保证当前业务与账务系统的最终一致性?如何保证第三方支付成功与本地业务DB的一致性？

   • 更新DB的还款业务为扣款中
   • 调用第三方支付扣款
   • 扣款成功后，异步通知账务系统入账
   • 更新DB的还款业务为已扣款
     

本地消息表

1. 消息生产方，需要额外建一个消息表，并记录消息发送状态。消息表和业务数据要在一个事务里提交。然后消息异步发送到消息的消费方，消费方返回是否处理成功，如果成功，则更新本地状态。如果消息发送失败或者消费方处理失败，会进行重试（通过Job）,重试达到最大次数，需要人工介入。

2. 消息消费方，需要处理这个消息，并完成自己的业务逻辑，并且支持幂等操作。

3. 如果是RPC方法发送消息，可以让服务端定期回调客户端(比如Dubbo的参数回调)，更新服务端的状态，可以减少交互

4. 架构图
   

5. 本地消息事务表-MQ方案

   • 本地业务与消息表在一个事务中，事务提交后异步发送MQ。RetryJob会定期重试发送没有更新状态的消息，达到最大次数，需要人工介入
   • 消费方订阅消息，写本地业务数据，如果处理成功则异步通知生产方。异步通知失败，生产方会定期重试，消费方做好消费幂等操作即可
     

6. 实现方式

   • 自定义注解+AOP方式+SPEL表达式(注意此AOP与事务AOP的顺序，一定要在事务内)

     @EnableMessage(key = "#order.uniqueId")
     public Order saveOrder(Order order)
     1
     2

   • 闭包方式（不推荐，代码耦合严重，不推荐）

     public Order saveOrder(Order order,TransactionCallBack<T> action)
     1

7. 本地消息事务表缺点：

   • 业务操作需要耦合本地事务表（可以通过代码的封装解决这个问题，比如通过注解+aop的方式或者通过闭包的方式，去解决业务开发需要关注本地事务表操作的问题），每个业务方都必须启动一个定时任务。
   • 各个服务节点之间的数据是最终一致性
   • 消息重试需要保证幂等，达到重试次数，需要人工处理

RocketMQ的事务消息

1. Producer向MQ发送Prepare状态的消息(需要注册业务状态检查处理器)

2. Producer执行本地业务逻辑(执行本地事务)

   • 本地事务执行成功，MQ收到ack信息，此时消费者可以订阅已经commit的消息
   • 本地事务执行成功，Producer发送超时MQ未收到ack信息（MQ通过长连接定时扫描事务状态，回调Producer判断状态）
   • 本地事务执行失败，此时可以发送rollback指令给MQ(也可以不发送，由MQ定时扫描)

3. 架构图
   

4. 原理分析

   • Produder在发送消息时候设置状态回调接口(根据自己的业务逻辑实现，主要内容是检查当前业务执行是否成功)
   • Producer与MQ保持长连接，MQ扫描事务状态消息(消息存储在文件或者数据库中)，并向Producer发起远程RPC调用

5. 适用场景：对于有唯一流水的情况比较合适(流水只插入不更新的情况)，因为当MQ回查Producer事务是否提交的时候，如果流水被更新可能导致MQ误以为事务没有执行成功，导致消息被作废掉

6. 优点： 相对于本地事务表，业务方只需要提供状态回查接口即可，不需要做其他额外操作(不需要独立启动一个定时任务)。

7. 缺点：

   • 目前只有RocketMQ实现了基于事务消息机制的功能，低版本使用文件存储，高版本使用数据库存储(但是没有完全开源)，所以需要修改源代码去实现这部分功能。
   • 一次消息发送需要两次请求；业务处理服务需要实现消息状态回查接口

通用独立MQ事务消息

1. 场景：对于不支持2PC事务机制的MQ。

2. 优点：

   • 不依赖于具体的MQ
   • 业务方不需要建立本地事务表，业务操作不需要耦合本地事务表，业务方不需要本地编写job
   • 业务放只需要提供一个状态回查接口即可
   • MQ可任意扩展

3. 缺点：

   • 一次消息发送需要两次请求；业务处理服务需要实现消息状态回查接口

4. 对于不支持2PC事务机制的MQ，可以开发一套【消息服务中间件】来解决这个问题，这个中间件包含几大模块

   • 2PC的API模块（一个预发送、一个确认发送）
   • Producer Job 和Consumer Job
   • 后台管理（配置回调地址、动态上传插件、人工处理等）

5. 架构图
   

设计方案

1. 服务功能模块

   • 【api服务模块】:提供两个接口
     • 预存储消息等待确认
     • 确认并发送消息
   • 【Producerjob模块】: 扫描未ack的消息，定时回调业务系统提供的状态回查接口，判断消息是否被ack，如果被ack，则更新消息状态为ack状态，并发送到MQ中
   • 【Consumer job模块】:扫描已经处于ack状态的消息，发送到MQ Server(如果发送失败(包括超时)，更新【消息发送次数】，并且更新状态为【发送失败状态】，如果成功更新状态为【发送成功】状态)，到达一定的失败次数，需要人工后台干预

2. 事务消息状态

   • 1 等待确认：发起方预存消息等待确认的状态
   • 2 已经确认：发起方已经处理，确认的状态
   • 3 废弃：主动方业务操作失败，导致无法通过业务id查询到状态，所以此时记录应该被标记为丢弃或者直接删除掉
   • 4 已经发送MQ但未确认：已经发送到MQ，但是Consumer没有消费或者消费失败（调用回查获取状态）
   • 5 已经发送MQ并且已经确认：已经发送到MQ，但是Consumer消费成功（调用回查获取状态）
   • 6 MQ发送达到最大次数：发送MQ失败或者异常已经达到最大次数，不再重试，人工介入
   • 7 producer消息回查最大重试次数：调用Producer端回查接口最大失败次数
   • 8 consumer消息回查最大重试次数：调用Producer端回查接口最大失败次数

3. 事务消息数据库设计

   CREATE TABLE `transaction_message` (
     `id` varchar(50) COLLATE utf8mb4_bin NOT NULL COMMENT '主键',
     `business_producer` varchar(50) COLLATE utf8mb4_bin NOT NULL COMMENT '业务发起方（业务方描述）',
     `business_unique_id` varchar(100) COLLATE utf8mb4_bin NOT NULL COMMENT '业务数据唯一标识',
     `message_id` varchar(100) COLLATE utf8mb4_bin NOT NULL COMMENT '消息id，唯一标识',
     `message_body` mediumtext COLLATE utf8mb4_bin NOT NULL COMMENT '消息内容',
     `message_data_type` tinyint(4) NOT NULL DEFAULT '1' COMMENT '消息类型,1 json',
     `message_send_times` int(11) NOT NULL DEFAULT '0' COMMENT '消息发送次数',
     `producer_callback_times` int(11) NOT NULL DEFAULT '0' COMMENT '回调发起方次数',
     `consumer_callback_times` int(11) NOT NULL DEFAULT '0' COMMENT '回调消费方次数',
     `message_queue_type` varchar(100) COLLATE utf8mb4_bin NOT NULL COMMENT '消费队列或者topic',
     `status` tinyint(4) NOT NULL COMMENT '状态，1等待确认 ,2已经确认 , 3废弃，4已经发送但未确认，5已经发送并且已经确认，6MQ发送达到最大次数，7producer消息回查最大重试次数，8consumer消息回查最大重试次数',
     `mark` varchar(100) COLLATE utf8mb4_bin DEFAULT NULL COMMENT '备注，额外消息',
     `business_request_time` datetime NOT NULL COMMENT '业务请求时的时间',
     `business_receive_time` datetime NOT NULL COMMENT '接收到业务请求时的时间',
     `version` int(11) NOT NULL DEFAULT '1' COMMENT '乐观锁版本号',
     `create_time` datetime NOT NULL COMMENT '创建时间',
     `last_update_time` datetime NOT NULL ON UPDATE CURRENT_TIMESTAMP COMMENT '最后一次修改时间',
     PRIMARY KEY (`id`),
     UNIQUE KEY `unq_business` (`business_unique_id`,`business_producer`) USING BTREE,
     KEY `idx_create_time` (`create_time`) USING BTREE
   ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='事务消息表';
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22

4. 如果后期对接的业务方非常多，数据量非常大，可以考虑

   • 第一阶段（GB/TB级）：分库分表，按照系统分库，业务和时间分表。按时间归档
   • 第二阶段（PB级）：KV数据库（例如ES结合HBase架构）

5. 逻辑视图
   

6. 技术栈
   

直连设计

1. 比如用户注册，发送积分。由于积分服务是采购的，无法整合MQ，此时可以直接异步调用积分服务的接口进行处理！

   • 方式一：单独启动一个consumer消费proxy服务，此消费服务调用积分服务方式二：不发送到MQ，【消息服务中间件】直接调用积分服务
   • 方式二：不发送到MQ，【消息服务中间件】直接调用积分服务

2. 解决以上问题，又会存在如下问题

   • 不同consumer的接口类型不同（比如webservice、http、dubbo、ftp、SDK等），如何做统一整合呢？难道每对接一个consumer都在【消息服务中间件】编写代码，然后发布吗？
   • 不同的producer和consumer回调的接口可能已经存在（比如Dubbo、SDK等），请求和消费方不想再增加适配消息服务中间件的标准HTTP接口。此时如何集成？

3. 直连架构图，消息中间件可以支持多种MQ扩展，也可以支持内存级别的事件发布
   

如何解耦不同业务的接口规则？

1. 业务场景分析

• 在信贷业务中，需要从外部征信数据源（同盾、支付宝、银行征信、爬虫等）获取用户各种数据来判断用户是否是一个信用良好的用户，并且能给用户贷多少钱
• 一般至少对接上百个数据源，甚至上千个。每一种数据源的接口类型、接口入参、出参、接口的调用形式都不一样？一般情况下，内部系统都直接对接数据源网关，网关负责转发和编排多个数据源，最终返回给业务系统指标数据（比如年龄、权重分、话单等等）

2. 常见的方案

   • 第一种：直接通过写if-else的方式
   • 第二种：策略模式
   • 第三种：网关+服务接口的方式，每个数据源调用都部署为一个独立的服务，网关注册并转发（增加和删除新数据源的运维成本高，两次远程调用性能低下）
   • 第四种：基于类加载机制，动态加载Jar模块

3. 要面临的问题

   • 每增加一个数据源都需要开发、测试、重新部署。
   • 不同三方的接口形式不一样，接口入参和出参也不一样
   • 不同三方的SDK可能存在冲突（依赖版本冲突等）,如何隔离？
   • 对于统一的网关，内核变动较少，变动最多的就是对不同的三方接口进行修修补补。对于增加和减少数据源，没有必要去重新部署,对于出现bug，或者数据源暂停的情况下，必须能动态下掉对应的数据源，而不是重启服务或者重新发布

动态加载模块

1. 解决回调和直连调用consumer下，不同consumer接口异构的问题

• 提供标准的API接口
• 通过插件方式实现标准API接口
• 内核动态加载外部实现的API实现Jar包（类似操作系统，内核保持不变，外部提供扩展。类似Dubbo的SPI机制，只不过是动态加载Jar和动态卸载Jar）

2. 实现原理

   • 实现标准的API接口，放在具体的目录
   • 内核扫描具体的目录，加载所有的实现包。当目录的实现包被删除时，动态卸载Jar包。这里涉及类加载和类的卸载

3. 实现方式

   • 方式一：通过URLClassloader加载Jar
   • 方式二：通过GroovyClassLoader加载Groovy脚本，当修改脚本时，无需重启JVM