消息机制（主讲人：强琦）在线视频

流计算的整个的消息机制

它是整个系统的处理的核心

它会影响系统的延迟

会影响整个容错机制 扩展性

以及服务的可用性

整个消息的过程

从分发到接收 到处理

相应的用户的work的代码

我们将这个消息框架

定义为shuffle的framerwork

对于消息的发送和接收

利用Zoomkeeper

来记录整个DAG

前面父亲结点和子结的

具体的物理地址

当然也可以利用队列

比如说Kafka

就像我们每一步都落地

每一步都存储这样的方式

来定义这个shuffle

当然前一种是主动push

对于解决消息的丢失

或者重发

那么前面也介绍到

其实有两种做法

一种做法是消息源头的重发

那么这种做法有什么好处

就是它每一步不用落地

就是每一个结点
跟每一个结点之间

是用Push过去的

中间结果不会进行任何的

可靠的存储

所以运行效率是比较高的

但是它会引发一个问题就是

当一个Topology比较大的时候

任意一个结点的档激或者超时

都会引起整体的重发和重做

那么进而增加雪崩的风险

第二种

是结点内部重发

但是它跟前者的优势

劣势刚好反转

它需要每一步的落地

也就是说

整个运行效率会比较低

系统也会产生大量的文件碎片

当然我们要说

中间的存储是可靠的

所以你就必须放在

一个持久化的存储里头

比如说盘古

这个对整个的系统运行的效率

以至对盘古的文件数

会大大增加

当然它的好处是

我重发重做只依赖于父亲结点

这是它的一个

很明显的好处

那么一种方法是

我的源头是可靠的存储

我只依赖于这一个点的可靠存储

源头存储既可以收集消息

又充当了消息重发的数据源

它中间结果不落地

雪崩

当然我在这里具体介绍一下雪崩

雪崩是指分布式系统中

最常见的异常的状态

是网络的抖动

在流计算处理系统中

消息跟踪的代价比较大

一般的跟踪机制

并不会在跟踪结果中

详细标识出是哪一个结点

发生故障

每法区别哪个结点发生

就是刚才我们说到的异或

异或只能判断说

我这个数据在整个DAG运行中

出现问题了

但它并不能标识出

在哪个结点出现故障

如果我们要跟踪

一个数据在某个结点出错的话

跟踪的信息本身

就会与数据量呈线性关系

整个的数据会完全线性膨胀

这个是不可接受的

在效率上

这种不加区分源头消息重发

会使得本身原来正常的结点

因为其他结点的故障

也会收到大量的重复数据

浪费宝贵的网络资源

使得整个网络的情况进一步恶化

然后这种情况会一步一步地变大

最终造成整个集群网络的瘫痪

针对以上问题

我们进行了消息命令流的

大量的阅检

就是我们不会跟踪具体

每一个Batch内的数据的跟踪

而是进而每一批发送

发送一个特殊的命令流

来跟踪这个命令流

因为任何一条数据的丢失

下游结点会知道

那么当它收到全部的数据

它才会去Acker这个命令流

所以我们无需对

Batch内的每一条数据进行跟踪

进而只跟踪每一个Batch内的

一个命令

来达到大量的跟踪消息的降低

那么我们要求每个信息

有唯一的batchid

并且

与前面源头结点的offset可重入

这是什么意思

就是我要求这个Batchid

再重新发一次

那么按照这个Batchid绑定的

这个offset
就是源头

比如Kafka的offset

一定是我用这个batchid

读到这个offset

一定读到的那个数据

一定是上一次读到那个数据

是完全一模一样的

这是一个很强的约束

而且还需要

这样子的Batchid去做消重

也作为一个唯一的版本

但是每次重试

我们会在Batchid后面加

新增一个attemptid

当然这样的问题

还是存在大量的通讯量

结点Crash后

整个表如何重建

也会产生这样的问题

内部重放

刚才我们反复提到了

它是收到消息后先落地

结点出现故障以后

它会在这个存储系统里头

上游结点去重放

那么上游结点定期去做Checkpoint

来尽量减少重放的这个代价

那么这一种方案有什么特点呢

就是它的方案极其简单

就是queue and work

但是依赖一个高效的

可靠的一个第三方的存储模块

它的特点是说

局部故障对全局影响非常好

非常小

雪崩的风险也极大地降低

这个也就详细介绍了

内部重放它的消息去重

是如何做到的

因为任意上下游两个结点

它相当于有一个本地的Queue

在Push消息的时候

会使用一个完全自增的一个id

这个id是从0开始赋值

当消息发送成功

接收方确定成功以后

去Acker这个消息

才可以进行下发下一个消息

否则一直会重试

所以接收端只要用这个递增id

每次去判断

这个递增id是否递增

就是之前的那个递增1

它有可以判断

我这个消息是重发的

还是正常接收

这样子的一个机制

每一次两个结点的发送消息

都会去持久化落地

这样就会造成大量的IO浪费

那么我们也会使用

读写分离技术

将结点接收信息

在上游信息进行持久化

和读区这个缓冲区队首消息

这两个进行异步化处理

以达到读写分离的解耦

读写分离的好处是什么

它可以做到网络抖动下

不影响到其他结点

局部故障也不放大

不会出现处理快的结点

一直在等慢的结点