一、两阶段提交2PC
在分布式系统中，可以使用两阶段提交来实现事务性从而保证数据的一致性，两阶段提交分为：预提交阶段与提交阶段，通常包含两个角色：协调者与执行者，协调者用于用于管理所有执行者的操作，执行者用于执行具体的提交操作，具体的操作流程：
1. 首先协调者会送预提交(pre-commit)命令有的执行者
2. 执行者执行预提交操作然后发送一条反馈(ack)消息给协调者
3. 待协调者收到所有执行者的成功反馈，则发送一条提交信息(commit)给执行者
4. 执行者执行提交操作
如果在流程2中部分预提交失败，那么协调者就会收到一条失败的反馈，则会发送一条rollback消息给所有执行者，执行回滚操作，保证数据一致性；但是如果在流程4中，出现部分提交成功部分提交失败，那么就会造成数据的不一致，因此后面也提出了3PC或者通过其他补偿机制来保证数据最终一致性，接下看看flink 是如何做到2PC，保证数据的一致性。
二、Flink中两阶段提交
1. jobMaster 会周期性的发送执行checkpoint命令(start checkpoint)；
2.当source端收到执行指令后会产生一条barrier消息插入到input消息队列中，当处理到barrier时会执行本地checkpoint, 并且会将barrier发送到下一个节点，当checkpoint完成之后会发送一条ack信息给jobMaster ；
3. 当DAG图中所有节点都完成checkpoint之后，jobMaster会收到来自所有节点的ack信息，那么就表示一次完整的checkpoint的完成；
4. JobMaster会给所有节点发送一条callback信息，表示通知checkpoint完成消息，这个过程是异步的，并非必须的，方便做一些其他的事情，例如kafka offset提交到kafka。
对比Flink整个checkpoint机制调用流程可以发现与2PC非常相似，JobMaster相当于协调者，所有的处理节点相当于执行者，start-checkpoint消息相当于pre-commit消息，每个处理节点的checkpoint相当于pre-commit过程，checkpoint ack消息相当于执行者反馈信息，最后callback消息相当于commit消息，完成具体的提交动作。那么我们应该怎么去使用这种机制来实现2PC呢？
Flink提供了CheckpointedFunction与CheckpointListener这样两个接口，CheckpointedFunction中有snapshotState方法，每次checkpoint触发执行方法，通常会将缓存数据放入状态中，可以理解为是一个hook，这个方法里面可以实现预提交，CheckpointListener中有notifyCheckpointComplete方法，checkpoint完成之后的通知方法，这里可以做一些额外的操作，例如FlinkKafakConsumerBase 使用这个来完成kafka offset的提交，在这个方法里面可以实现提交操作。
在2PC中提到如果对应流程2预提交失败，那么本次checkpoint就被取消不会执行，不会影响数据一致性，那么如果流程4提交失败了，在flink中可以怎么处理的呢？ 我们可以在预提交阶段(snapshotState)将事务的信息保存在state状态中，如果流程4失败，那么就可以从状态中恢复事务信息，并且在CheckpointedFunction的initializeState方法中完成事务的提交，该方法是初始化方法只会执行一次，从而保证数据一致性。