Flink 中的一把锁

悠悠 2023-07-02 05:27 6阅读 0赞

![format_png][]

点击箭头处“蓝色字”，关注我们哦！！

**那把锁**

锁用于多线程安全场景下，在Flink中存在一把锁，被用于数据处理线程、定时器调用线程、checkpoint线程。在StreamTask中定义了一个Object对象lock，通过使用synchronized方式进行同步，在task的初始化过程中该对象传给了SystemProcessingTimeService、StreamInputProcessor、StreamTwoInputProcessor。

**数据处理线程**

这里所说的数据处理线程表示正常的数据处理流程，可以认为就是processElement处理过程，StreamInputProcessor/StreamTwoInputProcessor主要工作就是读取数据然后调用对应的operator处理读取到的数据也就是调用processElement方法：

StreamRecord<IN> record = recordOrMark.asRecord();
                synchronized (lock) {
                  numRecordsIn.inc();
                  streamOperator.setKeyContextElement1(record);
                  streamOperator.processElement(record);
                }

通过源码可以发现每次调用processElement之前都会使用synchronized锁住lock，然后才能进行后续的处理。

**定时器调用线程**

Flink中有一个很重要的功能那就是定时器，窗口触发需要定时器、用户自定义注册定时器需要定时器，但是定时器又可以按照时间属性分为两种：事件时间语义下watermark推进触发的定时器、处理时间语义下定时调度的定时器。

watermark也是作为一种StreamElement在管道中流动，在watermark向前推进时也是需要获得锁才能触发定时器：

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);
          }
        }

同样在处理时间触发的定时器也是需要获得锁才能执行：

//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过程中会存在锁竞争：

//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—**

![format_png 1][]

关注回复Flink

获取更多系列

原创不易，好看，就点个"在看"

![format_png 2][]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6L1VDNk0xSGYwU1NFcHIxbGJiaWF0aWF4UEpjOHk5SmNPZVJ5Sk9JaHNpYndTeFdQbUhDcUp3eldYOHhUTXo5TVlMcEhLQWtSZlkyZmNNcW92eXhyUjlLcEF3LzY0MA?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2pwZy9TN3RJdm81eFlNNjNhR0FoNlZmT0ZJMHBDVnFrVXRkbkJXWUYzVkFaOUVLaGNpYzhWeEtWT1JpY2MzQVFiN2d2ZkVOT1FnQ1dycFZoc2lhdXFhanR0Unc3US82NDA?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2dpZi9TN3RJdm81eFlNN01GRVJScTAzTTkwQUVIdTBJRUFSOFpEM21BVzhyZDRJQUZrSGxndEZnaWNCNll1Z014bXdCMk1TVmIyWllMc1kyaWJpYU9pYVdsNHVjUncvNjQw?x-oss-process=image/format,png