[PHP知识库]记录一种新的流式数据计算架构的设计想法

灵感来源：

因为个人工作需要，做数据采集以及采集数据的处理计算，当时想着学习flink来解决数据实时处理计算的需求，但发现flink想要投入到生产环境的话还是过于复杂了，所以想是否可以基于springboot微服务来实现一个轻量级的、更灵活的实时数据流处理计算框架。特别记录一下个人的设计思路。

（尚在不断完善中。。。纯粹是个人灵感，手动记录一下，浅薄之处请各位大佬多多赐教，不要嘲笑，如果有哪位朋友基于此思路有具体代码实现，可以私聊我，我看看你是怎么实现的，参考学习一下，谢谢）

基于微服务实现流式数据处理思路：
1. 每个数据计算处理模块都是一个微服务项目，该项目内负责最基础的算子处理逻辑，即 一个算子任务=一个微服务处理逻辑，而算子任务的并行度=微服务处理模块的部署数量。

2. 上游算子如何感知到下游算子？

个人有两种思路：

（1）基于微服务服务注册中心的思路，下游算子主动向上游算子进行http请求注册，然后上游的每一个算子内部维护一个队列，内部记录下游算子的ip地址和端口号，用来请求发送数据。这种方式有一个缺点，就是在下游算子项目启动前配置好上游算子的所有机器ip地址和端口号。

（2）同样得，像eurake、nacos做集中的服务注册中心，然后每个上游算子定时主动和服务注册中心拉取最新的下游算子访问地址列表，这样的话不用在启动配置里写死了，方便动态扩展。

注意，每一类算子向服务注册中心注册的时候要带有一个共同的名字，代表着当前同一类算子任务的分组，这样上游算子才能依据下游算子任务组的名称获得正确的地址列表。（其实就和微服务注册道理一样）

第一种方式很不方便算子并行度动态扩展，所以pass掉，推荐第二种方式。

3. 上游算子如何将数据路由到同一个下游算子机器上？

有两种方式

（1）每条数据都制定一个字符串key值，通过这个key来进行hash分区，从缓存的下游算子队列中获取指定的下游算子机器。所以，这个缓存下游算子访问地址的队列中元素初始顺序（下标）一定不能发生改变，哪怕下游某个机器挂掉、或者与上游算子服务器直接出现网络分区，最多我们会给这个地址打上不可用的标识，但一定不能删除队列元素，避免打乱队列中的初始顺序。

（2）参考redis集群模式下的数据分区方式，我们固定16384个数据槽位，初始启动后，将这16384个数据槽位均分在每个算子任务集群机器上，这样就不在乎队列中的顺序改变了，我们将通过map结构来存储槽位对应的机器访问地址。

而数据的key分区操作，也将通过这16384个槽位进行分区，而且，同样得，我们也可以模仿redis集群解决数据倾斜的办法，如果真的有大量的数据倾斜到同一台机器，那么我们就针对这台机器上的槽位再次手动划分，划分到上线的新算子机器上，这样就可以迅速解决数据倾斜问题，而且这样的分区操作，实际上影响到的只有一台机器上的key，其他机器没有影响。（至于重分区的过程中上游算子的数据传输处理需要仔细考虑）

所以基本还是采用第二种方法比较靠谱。

4. 一旦下游算子出现问题，导致消息堆积，如何做到反压？

数据在上下游算子之间的传递依靠http协议，每个算子内部维护一个同步阻塞定长队列用来缓存上游传来的数据内容，当这个队列满容量的时候，通过http阻塞上游算子的数据传输行为执行，这样就会阻塞住上游算子，进而向整个上游传递阻塞，进而做到反压（http相对简单一些，而且还可以做到超时阻塞，如果不设置超时时间，则可以永久阻塞）。

5. 可以通过eurake来作为服务注册中心，通过数据key来指定数据要前往的下游算子服务器，因为eurake的自动服务发现，倒是可以实现算子并行度动态扩展，但需要考虑扩展后发生的重新hash如何解决
一般来说，算子每接收一条数据就会进行一次计算处理，得到一个处理结果要么将这个处理结果发送至下游继续处理，要么产生一个中间结果处理状态，这个状态也是一个计算统计结果，flink中将这些状态提供了内存、hdfs、以及rocksdb三种存储方式，同样的，我们可以采用本地内存、或者是集群redis、甚至是mysql来存储我们的中间状态。

6. 算子运行恢复数据状态、算子并行度调整所带来的rehash问题，如果有其中某个算子的计算处理逻辑存在问题，我们可能需要关闭这一批算子重新上线代码，这个时候我们就需要考虑状态恢复的解决办法，避免数据重复计算、或数据丢失，也能避免我们需要所有数据全部重新处理。?如果算子任务的并行度发生了变化，我们还需要处理算子中的缓存队列中所有数据key的rehash问题。

这里是最难解决的地方，目前来说没有特别好的思路。

首先就是算子运行状态恢复问题：我们在算子内部维护了一份队列用来缓存上游下发的数据，同时也可以阻塞上游处理流程，但是一旦机器宕机，这部分缓存的尚未处理计算的数据就会丢失，那么
或许我们可以将同步队列数据缓存一份到redis中，但是这也有个问题，这样做就意味着redis中将存储这整个计算集群机器上的所有缓存队列中的数据以及所有的算子状态数据等，在特别大数据量处理的情况下，redis是否可行存在较大问题。

要么就参考flink中的分布式快照算法，定期进行checkpoint，并将checkpoint保存到存储中间件中，但还是会比较复杂。
或许使用kafka？将上游数据写入直接写入kafka？也不太行，这种方式没法阻塞上游算子，而且需要为每一个上下游算子机器的数据传输之间建立一个topic，还需要记录每一个已消费的offset，玩意数据处理计算了，但是offset没记录下来，就会导致数据重复处理。

其次就是数据的重新hash问题，算子缓存队列中的数据需要重新分区，分发的重新分区后的机器上，同时还要保证必须是在算子队列中的数据重新分区完成后，才能继续正常接收上游发送的数据，至于算子内部的中间键控状态数据，如果集中在redis中存储的话，倒容易解决，如果是存在本地机器，那么就还得对这部分键控状态数据进行重新分发。

7. 这个想法还是在和阿里的一位面试官的面试中得到的启发，当我讲述完我构思的这套数据实时处理架构的思路的时候，阿里面试官问了我一句，既然能够做到算子并行度的动态扩展调整，那么能不能做到算子任务的动态插拔，也就是说当整个计算流程正在执行中的时候，能够添加一类新的算子任务（添加一个新的数据计算维度）到某个运行中的算子任务下（这个提问真的瞬间感觉不愧是阿里大佬，真的绝了）

个人实现思路比较简单，既然要实现算子的动态插拔，那么就要有一个映射关系，也就是当前算子的下游算子，个人想法是可以通过zookeeper来实现，当前算子名称注册挂载zookeeper上作为一个节点，然后下游算子任务分组名称就挂载这个节点下面作为子节点，而且是有序子节点，这样的话，当前算子节点只需要取到下游算子任务分组名，然后遍历进行数据分发即可

8. 分布式环境下所带来的 数据乱序处理问题，因为是分布式环境，下游算子会收到来自多个上游算子发送的数据，对于下游算子来说，接收到来自上游算子的数据时序很有可能都是乱序的，那么此时我们就需要考虑，如何保证数据的有序处理。也就是flink中的时间乱序问题。（怎么说呢，乱序问题flink实际上也没有更好的解决方法，通过watermark机制来尽量控制乱序带来的问题，但并不能完全解决）

首先，我们也需要向flink中一样，制定好每一条数据的时间语义，如果是处理时间语义，那就很好解决了，如果要依据接收数据请求的时间排序的话，我们加上一个公平锁就行了，但如果是事件时间语义，就比较麻烦，这种情况下暂时没有好的解决方法，只能参考flink，等收集到一定时间范围内的数据之后再做处理计算，这一块比较复杂，还没有好的实现思路。

9. 待补充。。。

?