全国服务热线：18271592020
资讯

关于长亭空间

超值服务提供卓越产品

   
  
    关于我们
  
新闻公告 News
   
    新闻公告
一文彻底搞透分布式一致性
来源: | 作者:chang | 发布时间: 2023-11-06 | 101 | 分享到:

分布式系统下的数据一致性可以分为两大类：

  1. 事务一致性：当多个节点进行操作时，所有节点终达成的状态都是一致的。这需要通过协调来保证操作的正确性，避免出现数据不一致的情况；
  2. 副本一致性：数据的多个副本之间保持一致性，这需要保证在对数据进行修改时，所有副本都能够及时更新，避免数据出现不同步的情况；

定义都比较抽象，举个例子感受一下：

  1. 事务一致性：电商平台使用优惠券下单场景：

图片图片

  1. 下单成功，优惠券必须处于“已锁定”状态；
  2. 支付成功，优惠券必须处于“已使用”状态；
  3. 订单取消，优惠券需要恢复为“待使用”状态；
  4. 优惠券和订单间就属于“事务一致”，两者间存在强关联关系。
  1. 副本一致性：
  • MySQL 主从复制：是指在主数据库上进行数据操作后，将这些操作同步到一个或多个从数据库上。从库必须与主库保持同步，以便从库中的数据和主库中的数据保持一致；
  • Redis 与 MySQL 一致性：在将 Redis 作为存储使用时，可以将 MySQL 看做主节点，Redis 看做从节点，当 MySQL 数据发生变更时，自动同步到 Redis 中，并保持数据的一致性；
  1. 图片image

【注】本文着重介绍 “事务一致性”，多副本一致性，详见 缓存 或 ES 篇。

1. 脱离数据库事务的怀抱

在关系型数据库中，事务（Transaction）是指一组数据库操作，这些操作要么全部成功要么全部失败。事务可以保证某些数据操作的一致性，当某一条操作失败时，会进行回滚，即撤销已执行的操作，使数据恢复到操作前的状态。

提到事务一致性，不得不说数据库事务 ACID：ACID是指数据库事务的四个关键特性，分别为原子性（Atomicity）、一致性（Consistency）、隔离性（Isolation）和持久性（Durability）：

  1. 原子性（Atomicity）：事务应该被视为一个原子操作，即事务中的所有操作要么全部执行成功，要么全部失败回滚。如果事务执行过程中出现错误，所有修改操作将被回滚撤销，不会对数据造成损坏；
  2. 一致性（Consistency）：事务执行前后，数据应该保持一致状态。所有数据修改操作都必须确保数据库的约束条件、触发器等规则不会被破坏，保持数据完整性；
  3. 隔离性（Isolation）：多个事务同时对同一数据进行操作时，事务之间应该相互隔离，互不干扰。数据库系统应该确保在并发情况下，事务的执行结果和串行执行的结果一致；
  4. 持久性（Durability）：事务完成后，其对数据库所作的所有修改都应该被保存，即使系统崩溃或重启后，修改的数据也应该是可用的；

银行转账应用程序就是典型的 ACID 模型的应用场景。假设用户A要向用户B转账1000元，转账过程就是一个事务，具有原子性、一致性、隔离性和持久性四大特性：

  1. 原子性：转账过程总共涉及两个操作：从A账户中减去1000元，向B账户中加上1000元。如果这两个操作中的任何一个失败，整个事务都将失败回滚；
  2. 一致性：转账前后所有账户的余额总和应该是不变的，不会出现余额不足或超额的情况；
  3. 隔离性：如果同时发起两个转账事务，应该确保每个事务只访问自己的数据，不会互相干扰；
  4. 持久性：一旦转账完成，更改数据的事务就必须写入磁盘，保证即使系统崩溃或重启后，这些数据仍然是可用的；

数据库事务是程序员的一大利器，但由于各种原因，这把利器离我们越来越远：

  1. 负载的挑战：随着业务的快速增长，数据库中的数据量或负载也会达到单一实例的上线，此时，我们：

垂直拆分：将不同的表放到不同的数据库实例，比如拆分出 User 实例，Order 实例；

图片图片

水平拆分：数据量超过单表容量时，将数据分拆到不同的数据库，比如 Order-1 实例、Order-2 实例；

图片图片

垂直+水平拆分：先进行垂直拆分，在进行水平拆分；

图片图片

  1. 微服务的挑战：微服务已经成为系统的事实架构，特别是 Spring Boot 和 Spring Cloud 的流行：

微服务的“自治”要求每个微服务都应该有自己的独立数据存储，避免与其他服务共享数据存储，从而降低服务之间的耦合性；

微服务间通过服务发现、负载均衡等方式，将服务之间的关系解耦，从而使得每个服务都具备独立的自治性；

图片图片

不管触发哪一种条件，都会产生跨数据库事务，从而增加系统设计的难度。

2. 常见一致性保障机制

针对该问题前人已经提出来多种应对方案，特别是关系型数据库。

2.1. MySQL事务一致性

熟悉 MySQL 实现的伙伴知道，MySQL 是通过 Redo log 和 Undo log 来实现事务一致性的：

  1. Redo Log：Redo Log 记录了事务对数据库所作的修改，包括插入、更新、删除等操作，它在事务提交前就被写入磁盘。如果出现故障导致系统崩溃，MySQL 会从 Redo Log 中恢复数据；
  2. Undo Log：Undo Log 记录了事务对数据库所作的修改的「前置操作」，并且在事务回滚时用来撤销事务所做的修改。当事务执行更新时，MySQL 会先将修改前的数据存储到 Undo Log 中，当事务需要回滚时，MySQL 会根据 Undo Log 中的记录将数据还原为修改前的状态。

具体的如下图所示：

图片图片

从图中可知：

  1. 每一个 DML 语句都会为其生成对应的 Redo log 和 Undo log。

Redo log 记录正向修改；

Undo log 记录逆向恢复；

  1. 事务提交应用全部 Redolog 以持久化正向修改；
  2. 事务回滚应用全部 Undolog 以逆向恢复；

其中，可以看出存在两个核心流程：

  1. 向前补偿：redo log 记录了事务执行的过程，以及事务提交前的数据修改，可以通过重做日志来恢复数据，实现向前补偿；
  2. 向后补偿：undo log 记录了事务执行过程中对数据的修改，可以用于回滚事务，实现向后补偿；

除了两种补偿机制外，还涉及一个重要的组件“补偿管理器”，用于对补偿机制进行统一协调。

2.2. 2PC 和 XA

2PC（Two-Phase Commit）和XA是分布式事务中常用的协议和接口：

  • 2PC是分布式事务协议，用于在分布式系统中协调多个参与者的事务提交或回滚。它包括两个阶段：准备阶段和提交阶段，参与者在准备阶段告知协调者它们是否可以正常提交，如果都能正常提交，则在提交阶段所有参与者都提交事务。如果有一个参与者无法正常提交，则所有参与者都需要回滚；
  • XA是一组应用程序接口（API），它使应用程序能够参与分布式事务，并与事务管理器协同工作，以保证事务的一致性。XA接口包括三个接口：XA Transactions、XA Resource、XA Resource Manager，用于实现分布式事务的协调和管理。

MySQL 采用了两阶段提交（Two-Phase Commit，简称 2PC）协议，保证 Redolog 和 Binlog 间的数据一致性，确保事务在所有相关节点（包括 Redolog 和 Binlog）执行的情况下，要么全部提交成功，要么全部回滚失败。

2PC只能应用于两个事务参与者的场景，而XA可以应用于多个事务参与者的场景，具体如图所示：

图片图片

XA 定义了一组接口：

  • XA资源管理器（XA Resource Manager，RM）：用于管理分布式事务的资源，如数据库、消息队列等；
  • XA事务管理器（XA Transaction Manager，TM）：用于协调各个资源管理器的事务处理；
  • XA接口：XA接口允许应用程序参与到分布式事务的协调中，包括开始、提交或回滚事务等操作；

对应的事务提交和回滚流程如下：

  • 应用程序通过XA接口开始一个分布式事务，XA事务管理器为该事务分配一个的全局事务ID；
  • 应用程序使用XA接口将某些操作注册为分布式事务的一部分，这些操作可以涉及多个XA资源管理器；
  • 当应用程序执行到提交事务的代码时，XA事务管理器先协调各个XA资源管理器，检查这些资源管理器是否都能够提交事务；
  • 如果所有的资源管理器都能够提交事务，则XA事务管理器向各个资源管理器发送提交事务的请求，并等待它们的响应；
  • 如果其中有任何一个资源管理器不能提交事务，则XA事务管理器向各个资源管理器发送回滚事务的请求，并等待它们的响应；
  • 当所有的资源管理器都响应提交或回滚事务的请求后，XA事务管理器将事务的状态（提交或回滚）通知给应用程序，并释放资源。

2PC （包括升级后的 3PC），在事务执行的整个流程中都需要对资源进行锁定，在分布式环境下将大幅增加系统响应时间，降低整个系统的吞吐，在实际工作中使用的非常少。

2.3. TCC

TCC 是实现分布式事务解决方案的一种有效方法，更是真正应用于实际工作的一大解决方案。

图片图片

TCC (try-confirm-cancel) 是一种分布式事务解决方案，它将一个分布式事务拆分成三个过程：

  • Try 操作：尝试执行分布式事务中的操作，检查所有参与方是否准备好执行事务。如果准备好，则锁定资源，等待确认或取消操作；
  • Confirm 操作：确认执行分布式事务中的操作，提交所有参与方的操作。如果有任何错误，则回滚所有操作并释放锁定的资源；
  • Cancel 操作：取消执行分布式事务中的操作，回滚所有参与方的操作并释放锁定的资源；

TCC 的操作流程如下：

  • 应用程序向协调者请求分布式事务，并传输所有需要执行的操作;
  • 协调者根据 TCC 的分布式事务处理策略创建一个的分布式事务 ID，并将它分配给每个参与方;
  • 各参与方执行 Try 操作，并锁定需要访问的资源;
  • 协调者检查所有参与方是否准备好执行操作，如果所有参与方都准备好，则进入 Confirm 阶段;
  • Confirm 阶段中，各参与方确认执行操作，并将结果提交给协调者;
  • 如果有任何错误，协调者将回滚所有操作并释放锁定的资源。否则，所有参与方之间的事务将得到确认执行，释放资源并关闭事务;
  • 如果任何参与方在 Try 阶段失败，则进入 Cancel 阶段;
  • Cancel 阶段中，各参与方撤销所有操作并释放锁定的资源;
  • 协调者记录每个阶段的操作，以便处理异常情况;

TCC 是一种补偿型事务机制，通过人工干预来处理异常，本身具备的灵活性，适用于各种不同类型的应用场景。

2.4. 事务一致性本质

看了不少一致性解决方案，不知道有没有发现一些规律？

核心组件基本一致：

  • 应用程序：简单理解为开发的应用系统，借助事务管理器和资源管理的的能力，完成事务一致性保障；
  • 事务管理器：事务的协调者，接收应用程序的请求，对多个资源管理器进行协调，共同完成正向补偿和逆向补偿；
  • 资源管理器：单一资源管理者，对外提供正向补偿接口和逆向补充接口，供应用程序和事务管理器使用；

核心流程基本一致：

  • 正向补偿：应用流程向前推进，终从一个状态变化为另一个状态；
  • 逆向补偿：应用流程向后推进，将所有操作进行回滚，使其恢复到前一状态；

简单来说：事务一致性就是通过协调各个参与节点来实现分布式事务的提交或回滚，确保所有涉及到的操作，要么全部执行成功，要么全部不执行。不同的实现方式只是不同的工具，其实现思路基本一致。

3. 业务一致性保障机制

前人已经为我们提供足够多的工具，如何更好的使用这些工具，就需要对业务场景进行深入分析。

业务系统一致性是指在多个系统或不同的环境中，不同用户或系统操作所产生的数据在逻辑上是相同的。它的本质是确保在任何情况下，不同系统或用户产生的数据都是一致的，并且在系统中的所有操作都是以预期方式进行的。业务系统一致性是确保数据的准确性和可靠性的关键因素，可以有效地避免数据错误和丢失，提高业务系统的可用性和可靠性，保障企业的持续发展。\如下图所示：

图片图片

如果可重试性事务间不存在依赖关系，可以并行执行，具体如下：

图片图片

在一个复杂的业务流程中，可以将事务分为三类：

  • 关键性事务：指的是系统中为关键的一步操作，如果事务提交失败，则进行回滚操作；如果事务提交成功，则成为事实，无法回滚；
  • 可补偿性事务：指的是在关键性事务之前的事务操作，通常提供正向和逆向两组操作，正向操作失败或关键事务失败，在会逆序调用逆向接口，以对操作进行回滚；
  • 可重试性事务：指的是关键事务之后的事务操作，关键事务提交成功，则事实已定，下游通过重试的方式完成事务；

我们以分布式系统中的下单流程为例：

图片图片

  • 关键性事务：就是下单操作，将用户的信息保存到数据库。保存失败，对已经操作的库存和优惠券进行逆向恢复；保存成功，通过重试保障下游事务的一致性；
  • 可补偿性事务：指的是优惠券和库存服务提供的正向和逆向操作，正向操作可以通过逆向操作进行恢复；
  • 可重试性事务：指的是添加自动取消任务、保存操作日志、发送 MQ，当订单数据保存成功后，这三者通过不断重试保障终都会执行；

3.1. 关键性事务

关键性事务：指在分布式系统中，只有当某个事务被成功提交后，整个系统才能认为这个事务是成功的。如果这个事务失败了，那么整个系统就会回滚到之前的状态。例如支付、订单提交等。

从关键性事务的使用场景出发，适合的工具便是关系数据库的事务保障。

图片图片

  • 事务提交成功：整个流程向前补偿，推动可重试性事务通过不断重试终完成业务逻辑；
  • 事务提交失败：触发整个流程回滚，逆序调用可补偿事务的回滚接口恢复状态；

3.2. 可补偿事务

可补偿事务指在某些业务操作中，如果其中一些子操作执行失败，可以由后续补偿操作进行补救，达到一定的业务目的，例如在资金交易中，如果账户余额不足而支付子操作失败，可以通过撤销订单等补偿操作来保障交易的正确性。

对于可补偿事务，需要提供两组操作：

  1. 正向：标准的业务操作，比如库存锁定
  2. 逆向：针对正向操作的恢复操作，比如释放锁定库存
3.2.1. Seata

Seata 是一个开源的分布式事务解决方案，旨在解决分布式系统中的事务一致性问题。在传统的分布式系统中，由于各个服务之间的数据交互和操作都是独立进行的，因此很容易出现数据不一致的情况。这会导致系统出现各种异常情况，如数据丢失、重复提交等，从而影响系统的稳定性和可靠性。

Seata 提供了多种解决方案来解决分布式事务一致性问题。其中包括 XA 模式、TCC 模式和 SAGA 模式等。

  • XA 模式是一种基于数据库的事务管理模式，Seata 通过与数据库进行交互来实现分布式事务的一致性。该模式适用于对数据一致性要求比较高的业务场景，如金融、电商等。但是，由于需要与数据库进行交互，因此该模式的性能相对较低;
  • AT模式（基于应用层的两阶段提交方式）：AT模式实现在应用程序中嵌入事务语义，通过协调维护必要的锁，实现多个业务节点之间跨多个数据库表的事务。适用于关系型数据库的应用场景，如电商下单等。
  • TCC 模式是一种基于补偿的事务管理模式，Seata 通过预留资源、尝试执行、确认执行和回滚执行四个阶段来实现分布式事务的一致性。该模式适用于对性能要求比较高的业务场景，如游戏、社交等。但是，由于需要进行多次异步通信，因此该模式的复杂度较高;
  • SAGA 模式是一种基于事件驱动的事务管理模式，Seata 通过将一个大的分布式事务拆分成多个小的本地事务，并通过异步消息传递来实现分布式事务的一致性。该模式适用于对性能和可用性要求比较高的业务场景，如微服务架构下的系统。但是，由于需要进行多次异步通信和状态管理，因此该模式的复杂度也较高。

Seata 还提供了一些重要的功能，如事务日志记录、故障恢复、动态扩展等，使得用户可以更加方便地使用该框架来解决分布式事务一致性问题。同时，Seata 还具有高性能、高可用性和易用性等特点，可以满足各种不同场景下的需求。

【注】感兴趣的话，可以找下 seata 的官方文档。

3.2.2. Context + Rollback

Seata 虽好，但中间件的引入将大幅提升系统的复杂性，对于一些不太严谨的场景或者一些运维能力不足的小团队可以自己实现回滚方案。

整体方案如下：

图片图片

  • 创建一个 Context 对象，用于保存整个流程的上下文数据。其中存在一个 List 属性，维护待回滚任务列表；
  • 每操作完一个正向流程，向 Context 中注册一个逆向回调，及 Rollback 任务；
  • 如果

关键事务提交成功，Context 注册的 RollbackEntry 便失去意义；

关键事务提交失败，调用 Context 的 fireFallback 方法进行逆向补偿，fireFallback 方法逆向调用注册的回滚方法，从而恢复业务状态

该方案基于内存实现，存在失灵的情况，不建议使用在严谨的场景。

3.3. 可重试性事务

可重试型事务指在业务操作中，如果某些操作由于网络波动等原因导致失败，可以通过重新执行这些操作来达到其预期的结果，例如在发送短信验证码时，由于网络状况不佳而发送失败，可以重新尝试发送，直到发送成功为止。

可重试性事务没有失败，只有成功，哪怕是短暂的失败也会通过不限的重试使其终达到成功状态。

3.3.1. @Retry

@Retry 是 Spring 框架提供的一个注解，用于在方法调用失败时自动进行重试。

通过 @Retry  注解，我们可以定义重试的次数、间隔时间和异常类型等信息，从而实现更可靠的方法调用。

具体来说，@Retry 注解可以通过以下属性来配置：

  • maxAttempts:重试次数；
  • value:重试间隔时间的数值表示；
  • fixedDelay:是否固定等待重试间隔时间后再进行下一次尝试；
  • backoffPolicy:重试间隔时间的退避策略；
  • allowCoreThreadTimeOut:是否允许在核心线程上进行超时等待；
  • excludeExceptions:需要排除的异常类型；
  • excludeClassNames:需要排除的类名列表；
  • loggerMessage:日志输出格式；
  • fallbackMethodName:当所有重试都失败后，执行的方法名称；

我们看下具体的使用：

  • 基于计数器的重试实现
复制 
@Retryable(value = {Exception.class}, maxAttempts = 3, backoff = @Backoff(delay = 1000)) public void doSomething() throws Exception { // 业务逻辑代码 }
				
    
					
					
  • 
						1.
					
    • 

					
  • 
						2.
					
    • 

					
  • 
						3.
					
    • 

					
  • 
						4.
					
    •

该实现会在方法调用失败时进行多3次的重试，每次重试之间会等待1秒的时间。如果超过3次重试仍然失败，则抛出异常。

  • 基于自定义异常处理的重试实现
复制 
@Retryable(value = {Exception.class}, maxAttempts = 3, backoff = @Backoff(delay = 1000), fallback = @Fallback(fallbackMethod = "doDefault")) public void doSomething() throws Exception { // 业务逻辑代码 }

private String doDefault(Exception e) { // 当出现指定异常时，执行该方法进行重试处理 }
				
    
					
					
  • 
						1.
					
    • 

					
  • 
						2.
					
    • 

					
  • 
						3.
					
    • 

					
  • 
						4.
					
    • 

					
  • 
						5.
					
    • 

					
  • 
						6.
					
    • 

					
  • 
						7.
					
    • 

					
  • 
						8.
					
    •

该实现会在方法调用失败时进行多3次的重试，每次重试之间会等待1秒的时间。如果超过3次重试仍然失败，则会执行 doDefault 方法来进行重试处理。在该方法中，我们可以自定义处理方式来处理异常情况。

@Retry 仍旧是一个内存解决方案，在场景下可能出现任务丢失的情况。因此在实际工作中，很少用于可重试性事务这种场景。

3.3.2. MQ

MQ（消息队列）消费者重试机制是指在消费消息时，如果消费者无法成功消费消息（比如网络异常、服务器故障等原因），会自动重试一定次数或间隔一定时间后再次尝试消费消息，以保证消息的可靠性和可用性。

如下图所示：

图片im

具有MQ的可重试性事务，需要以下保障：

  • 保障业务操作与消费发送之间的一致性：业务操作成功，消息必须发送成功；业务操作失败，消息不能发送；
  • 保障消息投递和消费消费之间的一致性：对于消费失败的消息，MQ 会自动进行重试，直至消费成功；

一般情况下会采用多次投递的方式来实现消息投递和消息消费之间的一致性，所以消息消费者需要保障幂等性，避免多次投递造成的业务问题。

3.3.2.1. 半消息

RocketMQ事务消息是一种支持分布式事务的消息模型，将消息生产和消费与业务逻辑绑定在一起，确保消息发送和事务执行的原子性，保证消息的可靠性。

事务消息分为两个阶段：发送消息和确认消息，确认消息分为提交和回滚两个操作。在提交操作执行完毕后，消息才会被消费端消费，而在回滚操作执行完毕后，消息会被删除，从而达到了事务的一致性和可靠性。

事务消息的发生流程如下：

图片图片

  • 生产者发送prepare消息到RocketMQ服务端，RocketMQ将消息存储到本地并返回结果；
  • 生产者开始执行本地事务，并根据本地事务的结果将状态信息提交给RocketMQ服务端；
  • 如果本地事务执行成功，生产者向RocketMQ服务端发送commit消息；
  • 如果本地事务执行失败，生产者向RocketMQ服务端发送rollback消息；
  • RocketMQ接收到commit或rollback消息后，对消息进行投放或删除；

如果生成者发送 prepare 消息后，未在规定时间内发送 commit 或 rollback 消息，RocketMQ 将进入恢复流程，具体如下：

图片图片

  • 如果在回查的时间之前没有收到相应的 commit 或 rollback 消息，则 RocketMQ 会将对该 prepare 消息进行回查；
  • 应用程序接收到回查指令，从业务库中获取数据，并根据业务逻辑进行判断，终是 commit 还是 rollback；
  • RocketMQ 接收到 commit 或 rollback 回复后，进行相应动作，从而实现业务操作和消息发送的一致性；

使用 RocketMQ 的事务消息代码示例如下：

复制 
// 编写事务监听器类 public class TransactionListenerImpl implements TransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0); // 执行本地事务 public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { int value = transactionIndex.getAndIncrement(); System.out.println("executeLocalTransaction " + value); // TODO 执行本地事务，并返回事务状态 // 本例假定 index 为偶数的消息执行成功，奇数的消息执行失败 if (value % 2 == 0) { return LocalTransactionState.COMMIT_MESSAGE; } return LocalTransactionState.ROLLBACK_MESSAGE; } // 检查本地事务状态 public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        System.out.println("checkLocalTransaction " + msg.getTransactionId()); // 模拟检查本地事务状态，返回事务状态 boolean committed = prepare(true); if (committed) { return LocalTransactionState.COMMIT_MESSAGE; } return LocalTransactionState.UNKNOW; } // 模拟操作预处理逻辑 private boolean prepare(boolean commit) {
        System.out.println("prepare " + (commit ? "commit" : "rollback")); return commit; }

} // 编写发送消息的代码 public class Producer {
    private static final String NAME_SERVER_ADDR = "localhost:9876"; public static void main(String[] args) throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer("MyGroup"); producer.setNamesrvAddr(NAME_SERVER_ADDR); // 注册事务监听器 producer.setTransactionListener(new TransactionListenerImpl()); producer.start(); // 发送事务消息 String[] tags = {"TagA", "TagB", "TagC"}; for (int i = 0; i < 3; i++) {
            Message msg = new Message("TopicTest", tags[i], ("Hello RocketMQ " + i).getBytes(StandardCharsets.UTF_8)); // 在消息发送时传递给事务监听器的参数 SendResult sendResult = producer.sendMessageInTransaction(msg, null); System.out.printf("%s%n", sendResult); } // 关闭生产者 producer.shutdown(); }
}
				
    
					
					
  • 
						1.
					
    • 

					
  • 
						2.
					
    • 

					
  • 
						3.
					
    • 

					
  • 
						4.
					
    • 

					
  • 
						5.
					
    • 

					
  • 
						6.
					
    • 

					
  • 
						7.
					
    • 

					
  • 
						8.
					
    • 

					
  • 
						9.
					
    • 

					
  • 
						10.
					
    • 

					
  • 
						11.
					
    • 

					
  • 
						12.
					
    • 

					
  • 
						13.
					
    • 

					
  • 
						14.
					
    • 

					
  • 
						15.
					
    • 

					
  • 
						16.
					
    • 

					
  • 
						17.
					
    • 

					
  • 
						18.
					
    • 

					
  • 
						19.
					
    • 

					
  • 
						20.
					
    • 

					
  • 
						21.
					
    • 

					
  • 
						22.
					
    • 

					
  • 
						23.
					
    • 

					
  • 
						24.
					
    • 

					
  • 
						25.
					
    • 

					
  • 
						26.
					
    • 

					
  • 
						27.
					
    • 

					
  • 
						28.
					
    • 

					
  • 
						29.
					
    • 

					
  • 
						30.
					
    • 

					
  • 
						31.
					
    • 

					
  • 
						32.
					
    • 

					
  • 
						33.
					
    • 

					
  • 
						34.
					
    • 

					
  • 
						35.
					
    • 

					
  • 
						36.
					
    • 

					
  • 
						37.
					
    • 

					
  • 
						38.
					
    • 

					
  • 
						39.
					
    • 

					
  • 
						40.
					
    • 

					
  • 
						41.
					
    • 

					
  • 
						42.
					
    • 

					
  • 
						43.
					
    • 

					
  • 
						44.
					
    • 

					
  • 
						45.
					
    • 

					
  • 
						46.
					
    • 

					
  • 
						47.
					
    • 

					
  • 
						48.
					
    • 

					
  • 
						49.
					
    • 

					
  • 
						50.
					
    • 

					
  • 
						51.
					
    • 

					
  • 
						52.
					
    • 

					
  • 
						53.
					
    • 

					
  • 
						54.
					
    • 

					
  • 
						55.
					
    • 

					
  • 
						56.
					
    • 

					
  • 
						57.
					
    • 

					
  • 
						58.
					
    • 

					
  • 
						59.
					
    •

单看代码很难理解，简单画了张图，具体如下：

图片图片

其核心部分就是 TransactionListener 实现，其他部分与正常的消息发送基本一致，TransactionListener 主要完成：

  • 执行本地事务，也就是业务操作；
  • 执行结果检测，通过反查业务数据，决定消息的后续处理策略；

为了使用事务消息，我们不得不在TransactionListener中编写进行大量的适配逻辑，增加研发成本，同时由于逻辑被拆分到多处，也增加了代码的理解成本。

事务消息存在一定的问题：

  • 与 MQ 实现强相关，并不是每个 MQ 实现都对事务消息提供支持；
  • API 比较晦涩，存在一定的学习成本，同时需要对业务逻辑拆分到 Listener 中，增加理解成本；

有没有实用性强、使用简单的方案，那可以使用 事务消息表 方案。

3.3.2.2. 事务消息表

事务消息表方案是一种常用的保证消息发送与业务操作一致性的方法。该方案基于数据库事务和消息队列，将消息发送和业务操作放入同一个事务中，并将业务操作和消息发送的状态记录在数据库的消息表中，以实现消息的可靠性和幂等性。

如下图所示：

图片图片

核心流程如下：

  • 应用程序开启一个数据库事务，并在事务中执行业务操作和消息发送；
  • 在事务中，将业务操作和消息发送的状态记录到消息表中；
  • 如果业务操作执行成功，并且消息发送成功，提交事务，否则回滚事务；
  • 定时扫描消息表，并根据消息状态重新发送未被确认的消息。如果消息发送成功，更新消息状态；否则根据重试次数更新消息状态或者丢弃消息；

通过事务消息表方案，可以保证消息的可靠性和幂等性。即使在消息发送失败或应用程序崩溃的情况下，也可以通过重新发送消息将业务操作和消息发送的状态同步。同时，该方案可以避免消息重复发送和漏发的情况。

作为一种通用解决方案，lego 对其进行支持，可参考 reliable-message 模块。

4. 业务补偿

不管在设计时使用哪种方案，都是在尽力降低不一致出现的概率，但可怕的是不一致问题终究会发生。

是不是有些奇怪，做了这么多还是无法从根源上彻底解决一致性问题，在实际工作中就是这样：

  • 并不是所有的可补偿事务都能回滚成功：在正向流程中我们都会对资源进行锁定，如果其他操作破坏了锁定资源或者破坏了准入条件，程序将无法正常回滚，必须人工介入进行解决。比如，生单时成功锁定优惠券，但超管发现优惠券发放错误对其进行回收，在进行优惠券回滚时，由于优惠券处于不可用状态，导致无法正常回滚；
  • 并不是所有的可重试事务都能重试成功：业务执行到可重试事务，只能证明其满足关键事务之前的条件，并不一定满足下游可重试事务的条件。比如，支付成功后需要给用户发送微信消息，但用户授权信息已经过期导致消息无法发送；
  • 业务迭代引入 bug 会破坏事务机制：这个就更常见了，由于bug导致流程错误，不得不修复问题和数据

除了主动降低不一致性概率，还需要添加一些被动保护机制，也就是常说的业务补偿。

4.1. 查询模式

查询模型是常用的一种方式，主要用于应对网络传输中的第三态问题。

第三态指的是在分布式系统中，在进行跨网络调用时，调用方无法确定被调用方的状态是否改变了，因为这两者之间存在一段未知而不可控的网络延迟时间，导致调用方无法立即得到被调用方的结果。这种情况下，第三态可以看做是一个未知的状态，需要通过一些机制来解决这个问题。

图片图片

当网络调用出现第三态时，简单的方式便是对不确定的状态进行查询，如上图所示：

  • 调用方调用服务完成业务操作，如果成功拿到执行结果，则直接进行后续流程；
  • 如果发生网络超时，将通过状态查询接口来检查之前的操作是否完成，如果：

已完成，则继续执行后续流程；

未完成，在重新发起业务调用；

RocketMQ 的事务消息便是基于该机制进行实现。

4.2. 任务检测模式

当一个业务操作完成后，需要处理多个后续任务，为了保障所有任务都会被执行，可以使用该模式。

如下图所示：

图片图片

image

  • 业务操作后，将业务变更和检测任务在同一事务保护下进行入库；
  • 系统继续执行后续任务，执行完成后对任务状态进行更新；
  • 系统周期性对超时未执行的任务进行加载，并进行检测，如果

已经执行，则更新任务状态

如果未执行，则触发任务执行

本地消息表就是基于该模式进行构建。

4.3. 对账模式

对账模式经常出现在与银行等金融机构对接的场景。

图片图片

业务对账思路非常简单：

  • 从不同的业务系统获取对账数据；
  • 按照规则进行双向对账，如果

一致，则说明系统一致

不一致，进行报警，人工介入进行处理

必须是双向对账，单向对账会出现数据丢失情况。

5. 小结

一致性是分布式系统面临的巨大挑战，根据不同场景可以将一致性分为：

  • 事务一致性。在一个事务内的所有操作，要么全部完成，要么全部不完成，即保证这些操作是对数据的一致更新，避免数据出现不一致的情况。主要通过使用事务保证来实现，例如：关系型数据库的ACID事务。
  • 副本一致性。各个副本之间的数据保持一致。当数据发生变化时，需要将这个变化同步到所有的副本中。主要使用副本同步技术来实现，例如MySQL的主从复制、MySQL 到 Redis的数据同步；

本文重点对事务一致性进行全方位的阐述，包括：

  • 技术视角，常见的解决方案：

MySQL 实现

2PC和XA协议

TCC 解决方案

  • 业务视角，将不同的事务进行分类，以便更好的解决：
  • 关键事务
  • 可补偿性事务
  • 可重试性事务

有了这些方案后，很多场景下仍需落地业务补充，常见方案包括：

  • 查询模型
  • 任务检查模式
  • 对账模型
    联系我们