修炼内功，一文梳理分布式事务及相关算法，剖析 Flink 端到端的一致性

kk大数据

发布于 2021-07-30 16:53:46

6000

发布于 2021-07-30 16:53:46

文章被收录于专栏：kk大数据 kk大数据kk大数据

一、临界知识对我们学习的巨大帮助

临界知识这个概念，是我上个月读《好好学习：个人知识管理精进指南》这本书学到的概念，真的有被启发到，现在觉得它对于我们深刻了解世界有着非常大的作用。

所谓临界知识，是我们经过深度思考后发现的，对于认识世界具有普遍指导意义的规律或定律，比如我们经常会看到复利模型、概率论、边际收益、二八法则这些基础概念，它们都是临界知识。通常一个临界知识，对不同的领域都具有指导意义和应用价值。

当然在编程世界中，也有很多临界知识：

比如最经典的就是设计模式里面的 KISS 原则、SOLID 原则，这些都是前人在无数的编程实践中提炼的最核心的概念，然而我很久之前可能还只停留在 Cat、Dog、Animal 等层次上（捂脸）；
再比如分布式系统中的 CAP 和 Base 理论，可能我们都知道它的内容，但是有可能仅限于字面上的了解，而未曾想过各个分布式系统是如何权衡与取舍，做出折中选择的；
再比如在 Flink 计算框架中，可以对内存中的数据设置 TTL 时间，但它本质上和 FIFO、LIFO 一样是一种数据淘汰的策略。假如说以后有幸可以参与底层框架设计，这就可以作为一种临界知识，指导我们的架构设计。

那我们今天就来看看分布式系统中的一些临界知识与核心概念。

二、集中式系统与分布式系统

1、集中式系统

就是由一台或者多台计算机组成中心节点，数据集中存储与这个中心节点中，并且所有功能都由这个中心节点来处理。

比如很多年前的单体 Mysql、Tomcat服务。提高性能只能提升物理主机的性能，代价昂贵。

所以阿里巴巴发起了去 IOE（IBM 小型机、Oracle 数据库、EMC 高端存储）的行动。单体架构带来的企业成本越来越高，提升单机性能的性价比越来越低；稳定性和可用性很难达标。

2、分布式系统

分布式服务，就是一群独立的服务器集合共同对外提供服务。但是对于用户来说是透明的，就相当于是一台超级计算机在提供服务。

分布式系统，可以使用廉价的机器，横向扩展性能。计算机越多，CPU、内存、存储资源等也就越多，能够处理的并发访问量也越大。

3、分布式系统存在的问题

分布式系统在功能上，确实要比单机系统要强大的多，但是分布式系统的设计实现和维护的难度大大提升了。下面总结了一些分布式系统常见的问题与解决方案：

通信异常。网络不可用（消息延迟或者丢失），是普遍性的问题，会导致分布式系统内无法顺利进行一次网络通信进行沟通协调。可能造成多节点数据丢失和状态不一致，或者造成数据乱序。解决方案：重试机制。
网络分区。整个网络环境，由于网络问题，被切分成多个孤立的网络区域，造成网络孤岛。解决方案：把数据状态不是最新的给下线掉。
节点故障/宕机。服务器节点出现宕机或者“僵死”的现象，这是常态。解决方案：数据多副本存储。

4、异常处理黄金原则

任何在设计阶段考虑到的异常情况一定会在系统实际运行中发生，但在系统实际运行遇到的异常却很有可能在设计阶段未考虑到，所以，除非需求指标允许，在系统设计阶段不能放过任何异常情况。

通常，一个设计精良的分布式系统，都会对异常做充分的处理，比如 Kafka，Spark 等知名框架。

三、中心化与去中心化

1、中心化

中心化的设计理念比较简单，分布式集群中的节点按照角色分工，大体上分为两种角色：

Master的角色主要负责任务分发并监督 Slave 的健康状态，可以动态的将任务均衡到 Slave 上，以致 Slave 节点不至于“忙死”或”闲死”的状态。
Worker的角色主要负责任务的执行工作并维护和Master的心跳，以便Master可以分配任务给Slave。

在 Hadoop 1.x 的设计中，JobTracker 和 TaskTracker 便是一种中心化的设计，它具有单点故障的问题。

中心化思想设计存在的问题：

一旦 Master 出现了问题，则群龙无首，整个集群就会崩溃。为了解决这个问题，大多数 Master/Slave 架构模式都采用了主备 Master 的设计方案，可以是热备或者冷备，也可以是自动切换或手动切换，而且越来越多的新系统都开始具备自动选举切换 Master 的能力,以提升系统的可用性。

2、去中心化

在去中心化设计里，通常没有 Master/Slave 的概念，所有的角色都是一样的，地位是平等的。

全球互联网就是一个典型的去中心化的分布式系统，联网的任意节点设备 down 机，都只会影响很小范围的功能。

去中心化设计的核心设计在于整个分布式系统中不存在一个区别于其他节点的”管理者”，因此不存在单点故障问题。

但由于不存在” 管理者”节点所以每个节点都需要跟其他节点通信才得到必须要的机器信息，而分布式系统通信的不可靠性，则大大增加了上述功能的实现难度。

实际上，真正去中心化的分布式系统并不多见。

反而动态中心化分布式系统正在不断涌出。在这种架构下，集群中的管理者是被动态选择出来的，而不是预置的，并且集群在发生故障的时候，集群的节点会自发的举行"会议"来选举新的"管理者"去主持工作。最典型的案例就是ZooKeeper及Go语言实现的Etcd。

四、分布式一致性

当一个计算集群中有多台服务器一起工作时，各台服务器状态由于网络、硬件等原因，会产生数据不一致的情况，这是分布式系统最常遇到的问题。

分布式一致性的级别有很多种，不同的分布式系统的实现，可以采取不同的一致性级别。

强一致性。写操作完成后，读操作一定能获取到最新的数据。在分布式场景中，很难实现，下文提到的 Paxos 算法，Quorum 机制，ZAB 协议等都是解决方案。
弱一致性。不承诺立即可以读到最新写入的值，也不承诺多久之后数据能达到一致，但会尽可能地保证某个事件级别（比如秒级别）后，数据能够达到一致状态。
读写一致性。用户读取自己写入结果的一致性，保证用户永远能够第一时间看到自己更新的内容。比如我们发一条朋友圈，朋友圈的内容是不是第一时间被朋友看到不重要，但是一定要显示在自己的列表上。
单调读一致性。本次读到的数据不能比上次读的旧。
因果一致性。如果节点 A 在更新完某个数据后通知了节点 B ，那么节点 B 之后对该数据的访问和修改都是基于 A 更新后的值。于此同时，和节点 A 无因果关系的 C 的数据访问，则没有这种限制。
最终一致性。是所有分布式一致性模型当中最弱的，不考虑中间的任何状态，只保证经过一段时间之后，最终系统内数据正确。它最大程度上保证了系统的并发能力，也因此，在高并发场景下，它也是使用最广的一种模型。

通常，在 OLTP 系统中，大多数都是强一致性；而在大数据的系统，比如 HDFS ，则是最终一致性。

五、分布式事务 - 2PC

2PC（two-phase commit protocol，两阶段提交协议），2PC 是一个非常经典的强一致、中心化的原子提交协议。

这里所说的中心化是指协议中有两类节点：一个是中心化协调者节点（coordinator）和 N 个参与者节点（participant）。

2PC 的执行过程

第一阶段：请求/表决阶段

1、在分布式事务发起者向分布式事务协调者发送请求的时候，事务协调者向所有参与者发送事务预处理请求（vote request）。

2、这个时候参与者会开启本地事务并开始执行本地事务，执行完后不会 commit，而是向事务协调者报告是否可以处理本事务。

第二阶段：提交/执行/回滚阶段

分布式事务协调者收到所有参与者反馈后，所有参数者节点均响应可以提交，则通知参与者和发起者执行 commit，否则 rollback。

2PC 的优点

2PC 原理简单，实现方便，很多分布式技术的分布式事务方案，都是基于 2PC 进行了改进，或者提供了补偿方案来设计。

2PC 的缺点

性能问题（参与者同步阻塞）

从流程上面可以看出，最大的缺点就是在执行过程中节点都处于阻塞状态。

各个操作数据库的节点都占用着资源，只有当所有节点准备完毕，事务协调者才会通知进行全局 commit/rollback，参与者进行本地事务 commit/rollback 之后才会释放资源，对性能影响比较大。

协调者单点故障问题

协调者是整个 2PC 的核心，一旦事务协调者出现故障，会导致参与者收不到 commit/rollback 的通知，从而导致参与者节点一直处于事务无法完成的中间状态。

数据不一致

在第二阶段的时候，如果发生局部网络问题或者局部参与者机器故障等问题，一部分参与者执行了 Commit ，而发生故障的参与者收不到 commit/rollback 消息，那么就会导致节点间数据不一致。

过于保守

必须收到所有参与者的正反馈才提交事务：如果有任意一个事务参与者的响应没有收到，则整个事务回滚失败。

六、分布式事务 - 3PC

3PC 的执行过程

总体来说，3 PC 相较于 2PC 来说，多了第一个询问阶段，即询问所有节点是否做好准备提交事务，避免某节点宕机，导致所有节点都占用资源的问题。

通过第一阶段的确认，第二阶段所有事务参与者执行事务成功的概率大大增加。

第一阶段：CanCommit（提交询问）

分布式事务协调者询问所有参与者是否可以进行事务操作，参与者根据自身健康情况，是否可以执行事务操作响应（y/n)。

第二阶段：PreCommit（预提交）

同 2PC 的第一阶段，只是加入了超时的概念。如果协调者收到的预提交响应为拒绝或者超时，则执行中断事务操作，通知各参与者中断事务

第三阶段：DoCommit（最终提交）

同 2PC 的第二阶段，同样加入了超时的概念。如果协调者收到执行者反馈超时，则发送中断指令。

并且参与者在一定时间内，未收到协调者的指令，则会自动提交本地事务。

3PC 的优点

降低了二阶段的同步阻塞范围。在第二阶段，只要参与者收到 preCommit 请求，就会执行事务，此后不管能不能（超时）收到协调者的 doCommit 请求，都会执行事务提交，不会出现阻塞问题
解决了单点问题。进入阶段三会出现两种情况，1：协调者出问题；2：协调者与参与者之间网络故障。都会导致参与者无法收到 DoCommit 请求，但参与者在超时之后都会提交事务。

七、Flink 如何保证事务一致性

好，到目前为止，我们已经掌握了如何保证分布式一致性的 2 种方法（2PC 和 3PC），这就是临界知识。但是它毕竟是一个理论，如何把它运用到实践中？我们可以参考著名的分布式计算框架 Flink，它是如何利用 2PC 来保证数据一致性的。

1、Flink 如何保证内部 Exactly Once

首先什么是 Exactly Once，也就是数据恰好被计算一次，不多计算也不少计算。

那么，什么情况下，数据会被多计算或者少计算？答案是一些 Operator 算子挂掉了。它消费了 Kafka 数据，但是在计算过程中发生了问题，计算结果没有被及时保存下来，这就造成了少计算的问题。

所以当应用程序故障时，为了保证数据消费 Exactly Once， Flink 是通过周期性进行 Checkpoint 机制来解决这个问题。每次做 Checkpoint 的时候，会把当前消费 Kafka 的 offset，计算结果等写入到状态后端中。任务挂了的时候，只需要从最近的一次成功的Checkpoint 中，拿到 offset 和计算结果，从这个地方接着开始消费和计算就行了。

举个例子：假设我们设置 1 分钟一次 Checkpoint，第 10 次 Checkpoint的时候，partition0 offset 消费到了 50000，PV 统计结果为（app1,30000），（app2,35000）。

又接着消费了 10s，offset 已经消费到了 50100，PV 结果为（app1,30010）（app2,35010），任务挂了，该怎么办？

很简单，只需要从最近的一次 Checkpoint 的 offset 50000 处接着消费，PV 值也要从（app1,30000），（app2,35000）开始算即可。

当然是不能从 offset = 50100，PV 结果为（app1,30010）（app2,35010）开始算的，因为此时还没有进行 Checkpoint，这个状态根本没有保存下来。

2、Flink 如何保证端到端的 Exactly Once

上面只是 Flink 内部使用 Checkpoint 来保证数据一致性，那么 Flink 的结果最终是要写入到 Sink 端的存储中的。但是 Flink 的 Sink 算子可能是多个，同时往外部存储里面写，该如何保证 Flink 和外部存储之间的 Exactly Once？

这就要使用上面我们说的二阶段提交了，Flink 将二阶段提交的逻辑放在 Checkpoint 的过程之中。

实现类为：TwoPhaseCommitSinkFunction

Flink 的 JobManager 对应到 2PC 的协调者，Operator 实例对应到 2PC 的参与者。

TwoPhaseCommitSinkFunction定义了如下 5 个抽象方法：

// 处理每一条数据
protected abstract void invoke(TXN transaction, IN value, Context context) throws Exception;
// 开始一个事务，返回事务信息的句柄
protected abstract TXN beginTransaction() throws Exception;
// 预提交（即提交请求）阶段的逻辑
protected abstract void preCommit(TXN transaction) throws Exception;
// 正式提交阶段的逻辑
protected abstract void commit(TXN transaction);
// 取消事务，Rollback 相关的逻辑
protected abstract void abort(TXN transaction);

这五个方法在什么时候会被执行呢？

（1）任务执行的第一个阶段

TwoPhaseCommitSinkFunction 对应的所有并行度在本次事务中 invoke 全部成功（往 Kafka 发送数据）；等到 Flink 开始 Checkpoint 时，会执行 snapshotState() 方法，这个方法会对本次事务进行预提交 preCommit() 。如果 invoke() 和 preCommit() 全部成功了，才表示第一个阶段成功了。

如果在第一个阶段中，有机器故障或者 invoke() 失败或者 preCommit() 失败，都可以理解为 2PC 的第一个阶段返回了 No，即投票失败，会执行 2PC 第二阶段的 rollback，对应到 TwoPhaseCommitSinkFunction 中，就是 abort 方法。

在第一个阶段结束时，数据会被写入到外部存储。如果外部存储的事务隔离级别为读已提交时（Read Committed），并不能读取到我们的写入的数据，因为没有执行 commit 操作。

（2）任务执行的第二个阶段

当所有的实例做快照完成，并且都执行完 preCommit 时，会把快照完成的消息发送给 JobManager，JobManager 收到后会认为本次 Checkpoint 完成了，会向所有的实例发送 Checkpoint 完成的通知（Notify Checkpoint Completed），当 Sink 算子收到这个通知之后，就会执行 commit 方法正式提交。

此时外部存储就可以读取到我们提交的数据了。