Flink Mailbox模型

Flink在1.10版本中完成了对StreamTask的运行机制进行了重构，引入了Mailbox机制。本文对这次改动做了哪些修改，以及对比之前的实现有哪些好处进行一些简要的介绍。

通过MailBoxProcessor优先处理特殊事件(checkpoint，窗口的timer)，以及处理InputChannel中的数据

Flink线程模型的变化

flink之前是通过一个CheckPointLock加锁来实现多线程之间的互斥操作，之后引入Actor模型的mailbox机制来取代现有的多线程模型。变为了单线程(mailboxprocessor)+阻塞队列(mailbox)的形式。flink重构核心线程模型，所有算子的Task都是StreamTask，Source算子由于历史原因是SourceStreamTask（兼容的核心思想是以两个不同的线程来独立运行，SourceFunction对应的事件生成在一个线程上，而Mailbox是另一个线程，并且两者以检查点锁来保持互斥）

这样针对遗留的Source循环还是以独立的一套机制运行，而绝大部分算子的task则运行在mailbox线程上。

个人理解：mailbox是task级别的，每个task都有一个mailboxprocessor线程处理特殊事件和inputChannel数据

Flink处理模型

Flink将任务代码转换为DAG后，每个节点有各自的并发度p，节点间依靠网络传输消息。每个Task的每个并发都是独立的线程，可以简单的理解为

class Task implements Runnable {public void run() {while(true) {// 从输入缓冲区中读取数据，经过处理逻辑后继续写到输出缓冲区processInput();}}
}

但是除了常规的处理逻辑，还有其他线程可能会访问Task线程数据，比如

checkpoint：checkpoint的同步阶段
processTimeTimer：timer触发时，比如窗口结束触发计算就是使用timer实现（eventTimeTimer由watermark消息触发，所以包含在Task线程中）
Async IO：异步IO在结果返回时
上面这些逻辑需要与Task线程的处理逻辑互斥，否则就可能会有并发问题，造成数据错乱。最简单的解决办法就是-加锁，在Flink-1.10以前也是这么做的。

Before 1.10

每个Task都有一把锁，名曰checkpointLock，对于所有互斥操作都需要抢到锁后再进行处理。这样虽然看起来简单方便，但也造成了很多问题…

checkpointLock四处乱飞，可读性较差，对Flink开发人员不够友好。
在SourceContext中将checkpointLock直接暴露给用户代码，造成了一定的不稳定性。如果没有正确处理好，可能无法保证"exactly-once"语义。

Flink-1.10: mailbox-based approach

如果能将互斥的操作全部放在同一个线程中执行，是不是就不需要加锁了，天然串行执行。
所以在1.10版本中引入了Mailbox机制，外层抽象为类型Executor的接口，所有的互斥操作全部作为Runnable提交到MailboxExecutor，构建为Mail后加入Mailbox队列，MailboxProcessor在Task-Thread中运行，主要分为3种情况

有特殊事件，即Mailbox不为空，优先处理完特殊事件。主要有checkpoint、timer
无特殊事件且有network.buffer不为空，执行processInput处理数据
无事可做，执行await等待唤醒

调用关系图
MailboxProcessor主要逻辑

while (processMail(localMailbox)) {mailboxDefaultAction.runDefaultAction(); //defaultAction == processInput
}

唤醒机制

在很多时刻，Mailbox队列中没有待处理事件，且没有数据流入，这时线程会进入等待，这里使用了ReentrantLock.newCondition().await()/signal() 替代了Object.wait()/notify()。
在Mailbox加入新任务时，执行signal()唤醒线程继续处理。如果有新的输入数据时，会在Mailbox中加入一类特殊的ControlMail，不执行具体逻辑，目的是为了唤醒线程继续工作。
这样可以保证可以在第一时间响应且不会浪费资源。

不足之处：LegacySourceFunction

原本API中的SourceFunction只提供了run方法，即Task-Thread永远在SourceFunction.run()内部死循环，这与现在的Mailbox机制不能很好的融合。现在暂时对SourceTask进行了兼容，每个SourceTask包含两个线程，一个运行MailboxProcessor，另一个执行SourceFunction，通过对关键操作加锁保证逻辑正确。

我们可以做些什么

在引入Mailbox机制后，所有的任务逻辑都会收敛在Task-Thread中执行，也就是MailboxProcessor，lock不会堆外暴露（现阶段SourceFunction除外）。只需要记录MailboxProcessor中实际运行时间的占比（即除去await时间），就可以作为Task的负载指标，越接近1时代表已经接近处理能力的极限，需要对任务进行扩容。
这样相比阶段使用bufferpool指标进行判断，会更加直观，而且在机器资源充足的情况下，负载与流量基本是线性的关系，对配置优化更有知道意义。