那把锁
锁用于多线程安全场景下，在Flink中存在一把锁，被用于数据处理线程、定时器调用线程、checkpoint线程。在StreamTask中定义了一个Object对象lock，通过使用synchronized方式进行同步，在task的初始化过程中该对象传给了SystemProcessingTimeService、StreamInputProcessor、StreamTwoInputProcessor。
数据处理线程
这里所说的数据处理线程表示正常的数据处理流程，可以认为就是processElement处理过程，StreamInputProcessor/StreamTwoInputProcessor主要工作就是读取数据然后调用对应的operator处理读取到的数据也就是调用processElement方法：
代码语言：javascript复制StreamRecord<IN> record = recordOrMark.asRecord();
synchronized (lock) {
numRecordsIn.inc();
streamOperator.setKeyContextElement1(record);
streamOperator.processElement(record);
}通过源码可以发现每次调用processElement之前都会使用synchronized锁住lock，然后才能进行后续的处理。
定时器调用线程
Flink中有一个很重要的功能那就是定时器，窗口触发需要定时器、用户自定义注册定时器需要定时器，但是定时器又可以按照时间属性分为两种：事件时间语义下watermark推进触发的定时器、处理时间语义下定时调度的定时器。
watermark也是作为一种StreamElement在管道中流动，在watermark向前推进时也是需要获得锁才能触发定时器：
代码语言：javascript复制public void handleWatermark(Watermark watermark) {
try {
synchronized (lock) {
watermarkGauge.setCurrentWatermark(watermark.getTimestamp());
operator.processWatermark(watermark);
}
} catch (Exception e) {
throw new RuntimeException("Exception occurred while processing valve output watermark: ", e);
}
}同样在处理时间触发的定时器也是需要获得锁才能执行：
代码语言：javascript复制 //SystemProcessingTimeServuce中
public void run() {
synchronized (lock) {
try {
if (serviceStatus.get() == STATUS_ALIVE) {
target.onProcessingTime(timestamp);
}
} catch (Throwable t) {
TimerException asyncException = new TimerException(t);
exceptionHandler.handleAsyncException("Caught exception while processing timer.", asyncException);
}
}
}定时触发为什么需要锁？在processElement中可能会操作状态、在定时回调onTimer中也可能会操作状态，那么状态就是作为共享数据，为了保证数据的一致性，所以这里加了锁。
checkpoint线程
checkpoint是由jobmaster协调完成的，会定时向source端发送barrier标记然后在数据流中流动，checkpoint是为了对状态某一个时间点的备份，同样与processElement存在状态数据的竞争，为了保证数据的一致性，在checkpoint过程中会存在锁竞争：
代码语言：javascript复制//StreamTask中performCheckpoint方法
synchronized (lock) {
if (isRunning) {
operatorChain.prepareSnapshotPreBarrier(checkpointMetaData.getCheckpointId());
// Step (2): Send the checkpoint barrier downstream
operatorChain.broadcastCheckpointBarrier(
checkpointMetaData.getCheckpointId(),
checkpointMetaData.getTimestamp(),
checkpointOptions);
// Step (3): Take the state snapshot. This should be largely asynchronous, to not
// impact progress of the streaming topology
checkpointState(checkpointMetaData, checkpointOptions, checkpointMetrics);
return true;
}
else {
.....
}
}—END—
