一文搞懂 Flink 中的锁

// 这里就是真正的，用户的代码即将被执行的地方
						// now we can do the actual processing
						StreamRecord<IN> record = recordOrMark.asRecord();
						//处理每条 record lock
						// 所以如果是 window 由 processElement 导致的 window fire 也会被 lock 住
						synchronized (lock) {
							numRecordsIn.inc();
							//throught KeySelector set KeyContext setCurrentKey
							streamOperator.setKeyContextElement1(record);
							//处理数据
							streamOperator.processElement(record);
						}

public void handleWatermark(Watermark watermark) {
			try {
				// event time lock
				synchronized (lock) {
					watermarkGauge.setCurrentWatermark(watermark.getTimestamp());//gauge
					//处理 watermark 的入口
					operator.processWatermark(watermark);
				}
			} catch (Exception e) {
				throw new RuntimeException("Exception occurred while processing valve output watermark: ", e);
			}
		}

@Override
		public void run() {
			// process timer lock
			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);
				}
			}
		}

private final Object lock = new Object();