NIFI是可以部署成集群的，在多台机器上分布式部署提高数据吞吐能力。本文第五章，通过源码，来梳理NIFI的分布式如何实现。在此之前，我们先来学习分布式系统中很重要的一种协议

一、Two-phase Commit（2PC）-两段式提交协议

Two-phase Commit（2PC）两段式提交协议是一种分布式一致性（consensus）协议，常被用于分布式系统中，用来保证分布式事务的原子性（atomic），即分布式事务的所有参与者，要么都提交，要么都回滚，不会存在一部分参与者提交了，而另外的参与者回滚的情况。

数据的修改请求被分为两个阶段

第一阶段 Voting phase

1. coordinator向所有participant发送Prepare commit消息，并且等待participant的回复。
2. participant收到了coordinator的Prepare commit，执行持久化undo log、redo log等相关操作，直到可以提交（commit）或者回滚（rollback）。
3. 在participant收到Prepare commit后的本地操作，可能成功，也有可能失败。如果成功则回复OK，如果失败则回复not OK。

第二阶段 Completion phase

分为两种情况：

Commit情况：

1. coordinator如果收到所有participant的OK，那么coordinator会向所有participant发送Commit消息
2. 每个participant执行本地commit，释放事务持有的锁等相关的资源
3. 每个participant回复ACK给coordinator
4. coordiantor受到了所有participant的ACK后，完成分布式事务

Rollback情况：

1. coordinator如果收到任意一个participant的not OK消息（或者coordinator Prepare超时），那么coordinator会向所有participant发送Rollback消息
2. 每个participant通过undo log执行回滚操作，释放事务持有的锁等相关的资源
3. 每个participant回复ACK给coordinator
4. coordinator收到了所有participant的ACK后，本地undo分布式事务相关修改

在上一节 NIFI的 WAL 日志部分有提到，NIFI的数据，分为两部分，一个是在界面上可操作的配置信息部分，比如添加的处理器，新增的处理器组，两个处理器之间新增的连接，处理器的启停，在分布式情况下，肯定是需要在一个节点上的修改，能及时同步到其他节点。

另外一个就是flowFile部分，比如Web接口接到XML之后，拆分成了很多条。这部分要实现的是在一个节点的处理器生成的流文件，采用分布式负载均衡后，其他节点能分担某个节点的任务压力。

所以第五章也分为两部分来说。

二、集群协调器

还是以处理器的启动为例，在本系列的第一篇文章中，就已经讲到了启动处理器的入口：

在NIFI的源文件中按照上述路径，就能找到，除去前部处理异常的代码，剩下的如下：

第546行有一个判断，点进去可以看到：

它是用来检查当前的请求是否应该被复制到集群中，当前的NIFI实例不属于集群中的节点或者没有与集群建立连接的时候，返回false. 前一个是通过配置判断，后一个则是查看有没有集群协调器或者有没有与集群协调器进行通信。然后是检查请求头中的的 X-Request-Replicated  是否设置.

在最开始启动某个处理器的时候，上述方法返回true，才能实现将节点的启动信息，通过集群告知其他节点一起启动。也就是说，对NIFI界面上配置的修改，应该是在操作的时候，同步到集群其他节点的。我们继续跟进到547行的方法，会来到这里：

 红框这里，是判断是把请求只给到集群的协调器，还是直接给到每一个节点。

 

如果当前节点，已经被选做是集群的协调器， ReplicationTarget 返回 CLUSTER_NODES，代码上边，否则走下边。简单理解就是说，如果当前节点已经是协调器了，就可以直接把请求，给下边的参与节点了，否则需要把请求发给协调器，协调器再给到各个节点。

我们先跟到1102行进去，会来到这里，注意看类名：

这里173的判断除了抛出异常，没有做实质性的工作，跳过去，来到最后：

获取到所有处于连接状态的节点，

 加锁，防止不同的请求在同一时间，修改流，同时注意280行，在请求中加了个属性REPLICATION_INDICATOR_HEADER。还记得前边的判断吧。通过这个字段，NIFI节点就知道，当前的请求，是来自于集群协调器的，不必再进行转发。

最后来到：

 

将给节点发送请求打成任务丢入线程池，完成对所有节点的请求。

回到刚刚的分支，当需要给协调器发消息，需要找到协调器的位置，然后发送消息。不再赘述

小结

所以通过上边的代码整理可以看到：在集群模式下，当在一个NIFI节点的界面上，对流程进行修改的时候，该节点在后端，会先判断当前节点是否被选做了集群的协调器。如果是，则通过协调器就能将修改请求，发送到各个节点。如果不是，则当前节点知道协调器的位置，将修改请求转给协调器节点，然后重复一。协调器发出来的请求，在请求头中会带有标识 REPLICATION_INDICATOR_HEADER。各其他节点如果从请求中拿到的该字段有值，则不再对请求进行转发，转而执行该请求内容。

熟悉分布式的同学，会觉得博主的文档写的比较啰嗦，其实分布式这块儿，博主也是刚接触，很多概念也并不清楚，看代码看到了才会知道原来是这么回事。所以这个博客，也算是博主本人学习过程的一个记录。

三、协议实现-两阶段请求的发送

我们回到下面这个给所有节点发送请求的方法，来看具体的实现：

 方法比较长，我们一步步看，先是验证节点存在，并且是处于连接中的状态。

这里定义了两个回调，然后在构造响应 response 的时候传了进去。

这里根据注释，当请求是可修改请求的时候，并且performVerification（需要进行检验的时候）需要进行两段式的提交，让各个节点去检验请求能否被执行，所有节点都返回可以执行，再通知各个节点真正去执行。所有的操作都在performVerification方法中处理。点进去：

这里重温下 Function 的用法，其实是定义了一个方法，这个方法的参数是 NodeIdentifier类型，返回值是 NodeHttpRequest。注意这里只是定义，只有调用apply方法的时候，才会真正地放回这个对象。我们进入623行

 这里就调用了apply方法，返回了 NodeHttpRequest 对象，这个对象实现了 Runnable 方法，因此可以被加入到线程池当中。我们进到run方法看这个线程会干嘛：

 

 实际地发送请求。在run方法执行完之后，调用了回调函数的 onCompletion 方法。上边的代码中我把回调折叠了，我们现在来回头看回调函数是怎么定义的。

把所有的响应，都放在集合中，并判断当前集合的大小和应该发送的请求数是否相等。相等的话说明所有节点都已经给了响应。当已经收到所有节点都响应202.这时候两段式提交的第一阶段已经完成，并且所有节点返回了OK。

可以看到又回到了本节一开始部分的代码。 并且注意到 performVerification 传的是false.

再次回到这里，就要走448的代码。

 然后是第二阶段的请求提交，跟第一阶段的代码是一样的，只不过是请求头不一样了。

至此2阶段提交请求的发送流程已经都走完了。

四、协议实现-两阶段请求的响应

响应我们还回到处理器启动接口，进入 withWriteLock 方法：

 判断如果是两段式的请求，

如果是第一阶段的验证阶段，先把请求记录下来，然后响应202

接到第二阶段的请求，如果是正式的执行请求，就开始正式执行。如果是撤销请求，则撤销事务然后返回响应。

五、小结

这里结合官网图，就能清楚zookeeper，进一步说明一下。

 图上有行小字，客户端与接口的交互可以可能在任何一个节点上，就能理解上边提到的为什么要判断当前节点是不是协调器。能理解对其中一个节点的操作能通过协调器转发到集群其他节点，并通过分布式一致性协议（2pc）保持各个节点的配置都是相同的