整体设计

Mailbox线程模型通过引入阻塞队列配合一个Mailbox线程的方式，可以轻松修改StreamTask内部状态的修改。Checkpoint、ProcessingTime Timer的相关操作（Runnable任务），会以Mail的形式保存到Mailbox内的阻塞队列中。StreamTask在invoke阶段的runMailboxLoop时期，就会轮询Mailbox来处理队列中保存的Mail，Mail处理完毕后才会对DataStream上的数据元素执行处理逻辑。

MailboxProcessor的能力就是负责拉取、处理Mail，以及执行MailboxDefaultAction（默认动作，即processInput()方法中对DataStream上的普通消息的处理逻辑，包括：处理Event、barrier、Watermark等）。

/*** 开始轮询Mailbox内的Mail，Checkpoint和ProcessingTime Timer的触发事件会以Runnable的形式（作为Mail）添加到Mailbox的队列中，等待“Mailbox线程”去处理*/
public void runMailboxLoop() throws Exception {// 获取最新的TaskMailbox：主要用于存储提交的Mail，并提供获取接口。// TaskMailbox有2个队列：//        1.queue：阻塞队列，通过ReentrantLock控制队列中的读写操作//        2.batch：非阻塞队列，调用createBatch()方法会将queue中的Mail转存到batch中，这样读操作就能通过tryTakeFromBatch()方法从batch队列中批量获取Mail，且只能被Mailbox线程消费final TaskMailbox localMailbox = mailbox;// 检查当前线程是否为Mailbox线程，即StreamTask运行时所在的主线程Preconditions.checkState(localMailbox.isMailboxThread(),"Method must be executed by declared mailbox thread!");// 确认Mailbox的状态：必须为OPENassert localMailbox.getState() == TaskMailbox.State.OPEN : "Mailbox must be opened!";// 创建MailboxController实例：可以控制Mailbox的循环、临时暂停和恢复MailboxDefaultAction（默认动作）final MailboxController defaultActionContext = new MailboxController(this);/*** 核心：事件循环* processMail()方法会检测Mailbox中是否还有Mail需要处理，新Mail会（在ReentrantLock的保护下）被添加到queue队列并转存到batch队列中。* MailboxProcessor处理完batch队列中的全部Mail后（执行作为Mail的Runnable#run()方法），才会进入到while循环内，执行MailboxDefaultAction的默认动作，* 即调用StreamTask#processInput()方法，对读取到的数据（Event、Barrier、Watermark等）进行处理*/while (processMail(localMailbox)) {mailboxDefaultAction.runDefaultAction(defaultActionContext); // lock is acquired inside default action as needed}
}

可以看出，对Mail和MailboxDefaultAction的处理，是由唯一的Mailbox线程负责的。

processMail

在while轮询时，首先会processMail

/*** 处理Mailbox中的Mail：Checkpoint、ProcessingTime Timer的触发事件，会以Runnable的形式作为Mail保存在Mailbox的queue队列中，* 并在ReentrantLock的保护下，将queue队列中的新Mail转移到batch队列中。MailboxProcessor会根据queue队列、batch队列内的Mail情况，* 决定处理Mail or processInput。只有当TaskMailbox内的所有Mail全都处理完毕后，MailboxProcessor才会去processInput()*/
private boolean processMail(TaskMailbox mailbox) throws Exception {/*** 新Mail写入queue队列，TaskMailbox会将queue队列中的新Mail转移到batch队列中，MailboxProcessor会根据queue队列、batch队列内的Mail情况，* 判断执行Mail的run() or processInput()。只有当TaskMailbox内的所有Mail全部处理完成后，MailboxProcessor才会去processInput()。*/if (!mailbox.createBatch()) {return true;}Optional<Mail> maybeMail;/*** 能走到这，说明queue队列中的Mail已被全部转移至batch队列。现在要从batch队列中获取到Mail并执行（它作为Runnable的run()方法），* 直到batch队列中的所有Mail全都处理完毕*/while (isMailboxLoopRunning() && (maybeMail = mailbox.tryTakeFromBatch()).isPresent()) {maybeMail.get().run();}/**如果默认操作处于Unavailable状态，那就先阻塞住，直到它重新回归available状态*/while (isDefaultActionUnavailable() && isMailboxLoopRunning()) {mailbox.take(MIN_PRIORITY).run();}// 返回Mailbox是否还在Loopreturn isMailboxLoopRunning();
}

很核心的一个点就是Mailbox要去createBatch，TaskMailboxImpl提供了具体的实现逻辑。Mailbox引入了2个队列，新Mail被add到Mailbox内的queue队列中（此过程受ReentrantLock保护）。同时为了减少读取queue队列时的同步开销，Mailbox还构建了一个batch队列专门用来后续消费（避免加锁操作）。

/*** 对Deque<Mail>队列的读写，通过ReentrantLock加以保护*/
private final ReentrantLock lock = new ReentrantLock();/*** Internal queue of mails.* 使用Deque（内部队列）保存所有的Mail*/
@GuardedBy("lock")
private final Deque<Mail> queue = new ArrayDeque<>();
/*** 为了减少读取queue队列所造成的同步开销，TaskMailbox会创建一个batch队列，queue队列中的Mail会被转移到batch队列中，* 有效避免了后续消费时的加锁操作*/
private final Deque<Mail> batch = new ArrayDeque<>();@Override
public boolean createBatch() {checkIsMailboxThread();/*** 如果queue队列中没有新Mail，那就要看batch队列是否为空。* 1.如果batch也是空的（Mailbox里已经没有任何Mail了，需要去processInput()了），那processMail()也会return true，* MailboxProcessor就会进入到while循环内部，执行processInput()来处理DataStream上的数据；* 2.如果batch不空，说明MailboxProcessor还需要继续processMail()，即取出Mail执行它（作为Runnable）的run()方法；* 由此可见，Mailbox中的batch队列中的Mail最终一定会被Mailbox线程消耗殆尽（轮询、处理），然后才会去processInput()*/if (!hasNewMail) { // 只要queue队列里还有Mail，hasNewMail就为truereturn !batch.isEmpty();}/**能走到这说明queue队列中仍有新Mail，接下来需要将它的新Mail向batch队列转移，该过程受ReentrantLock保护*/final ReentrantLock lock = this.lock;// 获取锁lock.lock();try {Mail mail;/**每次循环都将queue队列中的First Mail，转移到batch队列中，直至queue队列被消耗殆尽。此时一定return true*/while ((mail = queue.pollFirst()) != null) {batch.addLast(mail);}// 此时queue队列内的所有Mail都被转移到batch队列中了，queue中没有新Mail了hasNewMail = false;// 此时根据batch队列是否为空，MailboxProcessor会判断执行Mail的run() or processInput()return !batch.isEmpty();} finally {// 最终释放锁lock.unlock();}
}

如果Mailbox内的queue队列中仍有新Mail，那就在ReentrantLock的加持下将queue内的Mail全都转移到batch队列中；如果Mailbox内的queue队列中没有新Mail，那就看batch队列的情况了。决断权交给外层的MailboxProcessor，总的来看：

• 如果batch队列中有Mail，MailboxProcessor会从Mailbox内的batch队列中逐个pollFirst，然后执行（它作为Runnable#run()方法），直到batch队列中的所有Mail全都被“消耗殆尽”为止
• 如果batch队列中没有Mail，MailboxProcessor此时就没有Mail可处理了，那就直接processInput

1.Checkpoint Tigger

对Checkpoint的触发，是通过MailboxExecutor向Mailbox提交Mail的

/*** 触发执行StreamTask中的Checkpoint操作：异步的通过MailboxExecutor，将“执行Checkpoint”的请求封装成Mail后，* 提交到TaskMailbox中，最终由MailboxProcessor来处理*/
@Override
public Future<Boolean> triggerCheckpointAsync(CheckpointMetaData checkpointMetaData,CheckpointOptions checkpointOptions,boolean advanceToEndOfEventTime) {// 通过MailboxExecutor，将“触发执行Checkpoint”的具体逻辑封装成Mail，提交到Mailbox中，后期会被MailboxProcessor执行return mailboxProcessor.getMainMailboxExecutor().submit(// 触发Checkpoint的具体逻辑() -> triggerCheckpoint(checkpointMetaData, checkpointOptions, advanceToEndOfEventTime),"checkpoint %s with %s",checkpointMetaData,checkpointOptions);
}

triggerCheckpoint操作会被封装成Mail，添加到Mailbox中等待被处理。

@Override
public void execute(@Nonnull final RunnableWithException command,final String descriptionFormat,final Object... descriptionArgs) {try {mailbox.put(new Mail(command, priority, actionExecutor, descriptionFormat, descriptionArgs));} catch (IllegalStateException mbex) {throw new RejectedExecutionException(mbex);}
}

当然，Checkpoint的完成操作，也是同样的套路。

2.ProcessingTime Timer Trigger

/*** 借助Mailbox线程模型，由MailboxExecutor负责将"ProcessingTime Timer触发的消息"封装成Mail提交到TaskMailbox中，后续由MailboxProcessor处理*/
public ProcessingTimeService getProcessingTimeService(int operatorIndex) {Preconditions.checkState(timerService != null, "The timer service has not been initialized.");MailboxExecutor mailboxExecutor = mailboxProcessor.getMailboxExecutor(operatorIndex);// 通过MailboxExecutor将Mail提交到Mailbox中等待处理return new ProcessingTimeServiceImpl(timerService, callback -> deferCallbackToMailbox(mailboxExecutor, callback));
}private ProcessingTimeCallback deferCallbackToMailbox(MailboxExecutor mailboxExecutor, ProcessingTimeCallback callback) {return timestamp -> {mailboxExecutor.execute(() -> invokeProcessingTimeCallback(callback, timestamp),"Timer callback for %s @ %d",callback,timestamp);};
}

processInput

StreamInputProcessor会对输入的数据进行处理、输出，包含：StreamTaskInput + OperatorChain + DataOutput。每次processInput都相当于是在处理一个有界流（外层MailboxProcessor在不断地的轮询），处理完DataStream上的StreamRecord后，会返回InputStatus的枚举值，根据InputStatus值来决定下一步该“何去何从”。

/*** StreamTask的执行逻辑：处理输入的数据，返回InputStatus状态，并根据InputStatus决定是否需要结束当前Task。* 该方法会通过MailboxProcessor调度、执行（作为MailboxProcessor的默认动作），底层调用StreamInputProcessor#processInput()方法*/
protected void processInput(MailboxDefaultAction.Controller controller) throws Exception {/*** 核心：借助StreamInputProcessor完成数据的读取，并交给算子处理，处理完毕后会返回InputStatus。* 每次触发，相当于处理一个有界流，在外层Mailbox拉取Mail才是while循环无限拉取*/InputStatus status = inputProcessor.processInput();/*** case 1：上游如果还有数据 && RecordWriter是可用的，立即返回。意为：继续处理！*/if (status == InputStatus.MORE_AVAILABLE && recordWriter.isAvailable()) {return;}/*** case 2：当状态为END_OF_INPUT，说明本批次的有界流数据已经处理完毕，* 通过MailboxCollector来告诉Mailbox*/if (status == InputStatus.END_OF_INPUT) {controller.allActionsCompleted();return;}/*** case 3：当前有界流中没有数据，但未来可能会有。此时处理线程会被挂起：直到有新的可用数据到来 && RecordWriter可用* 此时会先临时暂停对MailboxDefaultAction的处理，等啥时候又有新数据了，再重新恢复MailboxDefaultAction的处理。*/CompletableFuture<?> jointFuture = getInputOutputJointFuture(status);// 通过MailboxCollector让Mailbox线程暂时停止对MailboxDefaultAction的处理MailboxDefaultAction.Suspension suspendedDefaultAction = controller.suspendDefaultAction();// 等啥时候又有了input、output，RecordWriter也变得可用了以后，再重新继续执行默认操作jointFuture.thenRun(suspendedDefaultAction::resume);
}

MailboxController是MailboxDefaultAction和Mailbox之间交互的桥梁，在StreamTask处理DataStream元素的过程中，会利用MailboxController将处理状态及时通知给Mailbox。如果这批有界流处理完毕，就会通过MailboxController通知Mailbox（本质就是向Mailbox发送一个Mail），进行下一轮的处理。

private void sendControlMail(RunnableWithException mail, String descriptionFormat, Object... descriptionArgs) {mailbox.putFirst(new Mail(mail,Integer.MAX_VALUE /*not used with putFirst*/,descriptionFormat,descriptionArgs));
}

兼容SourceStreamTask

作为DataStream Source是专门用来生产无界流数据的，并不能穿插兼顾Mailbox内Mail的检测。如果仅有一个线程生产无界流数据的话，那将永远无法检测Mailbox内的Mail。作为StreamTask的子类，SourceStreamTask会额外启动另一个独立的LegacySourceFunctionThread线程来执行SourceFunction中的循环（生产无界流），Mailbox线程（主线程）依然负责处理Mail和默认操作。

/*** 专门为Source源生产数据的线程*/
private class LegacySourceFunctionThread extends Thread {private final CompletableFuture<Void> completionFuture;LegacySourceFunctionThread() {this.completionFuture = new CompletableFuture<>();}@Overridepublic void run() {try {// CheckpointLock保证线程安全headOperator.run(getCheckpointLock(), getStreamStatusMaintainer(), operatorChain);completionFuture.complete(null);} catch (Throwable t) {// Note, t can be also an InterruptedExceptioncompletionFuture.completeExceptionally(t);}}public void setTaskDescription(final String taskDescription) {setName("Legacy Source Thread - " + taskDescription);}CompletableFuture<Void> getCompletionFuture() {return isFailing() && !isAlive() ? CompletableFuture.completedFuture(null) : completionFuture;}
}

负责为Source生产无界流数据的LegacySourceFunctionThread线程启动后，不管是启动成功 or 出现异常，都会封装对应的Mail并发送给Mailbox，而Mailbox线程的processMail一直在等待处理Mail。

/*** SourceStreamTask中，一个Thread负责专门生产无界流，另一个MailBox Thread处理Checkpoint、ProcessingTime Timer等事件Mail*/
@Override
protected void processInput(MailboxDefaultAction.Controller controller) throws Exception {/*** 通过MailboxDefaultAction.Controller告诉Mailbox：让MailboxThread先暂停处理MailboxDefaultAction。* TaskMailbox收到该消息后，就会在processMail()中一直等待并处理Mail（在MailboxThread中会一直处理Mail）*/controller.suspendDefaultAction();/**启动LegacySourceFunctionThread线程：专门生产Source无界流数据的，和MailboxThread线程一起运行*/sourceThread.setTaskDescription(getName());sourceThread.start();/**LegacySourceFunctionThread线程启动后，会通知Mailbox，Mailbox会在processMail()中一直等待并处理mail（不会返回，即Mailbox线程会一直处理mail）*/sourceThread.getCompletionFuture().whenComplete((Void ignore, Throwable sourceThreadThrowable) -> {/**LegacySourceFunctionThread线程启动过程中发生的任何异常、以及启动成功，都会以Mail的形式发送给Mailbox*/if (isCanceled() && ExceptionUtils.findThrowable(sourceThreadThrowable, InterruptedException.class).isPresent()) {mailboxProcessor.reportThrowable(new CancelTaskException(sourceThreadThrowable));} else if (!isFinished && sourceThreadThrowable != null) {mailboxProcessor.reportThrowable(sourceThreadThrowable);} else {mailboxProcessor.allActionsCompleted();}});
}

Mailbox主线程和LegacySourceFunctionThread线程线程都在运行，通过CheckpointLock锁来保证线程安全。