网站建设销售人才简历,crm系统功能模块,手机报价网站大全,品牌营销策划服务背景 当flink消费kafka的消息时#xff0c;我们经常会用到FlinkKafkaConsumer进行水位线的发送#xff0c;本文就从源码看下FlinkKafkaConsumer.assignTimestampsAndWatermarks指定周期性水位线发送的流程 FlinkKafkaConsumer水位线发送 1.首先从Fetcher类开始#xff0c…背景 当flink消费kafka的消息时我们经常会用到FlinkKafkaConsumer进行水位线的发送本文就从源码看下FlinkKafkaConsumer.assignTimestampsAndWatermarks指定周期性水位线发送的流程 FlinkKafkaConsumer水位线发送 1.首先从Fetcher类开始创建Fetcher类的时候会构建一个周期性的水位线发送线程并启动 // if we have periodic watermarks, kick off the interval schedulerif (timestampWatermarkMode WITH_WATERMARK_GENERATOR autoWatermarkInterval 0) {PeriodicWatermarkEmitterT, KPH periodicEmitter new PeriodicWatermarkEmitter(checkpointLock,subscribedPartitionStates,watermarkOutputMultiplexer,processingTimeProvider,autoWatermarkInterval);periodicEmitter.start();}2.随后PeriodicWatermarkEmitter中注册处理时间定时器周期性执行 public void start() {timerService.registerTimer(timerService.getCurrentProcessingTime() interval, this);}Overridepublic void onProcessingTime(long timestamp) {synchronized (checkpointLock) {for (KafkaTopicPartitionState?, ? state : allPartitions) {// 这里当前算子任务消费的kafka 分区分别记录每个分区的水位值state.onPeriodicEmit();}//这里当前算子会把自己消费的kafka分区的所有水位线取最小值后当成当前算子任务自身的水位线发送出去注意这里是当前算子任务级别的watermarkOutputMultiplexer.onPeriodicEmit();}// schedule the next watermarktimerService.registerTimer(timerService.getCurrentProcessingTime() interval, this);}}3.对应state.onPeriodicEmit();记录每个kafka分区的水位线方法 Overridepublic void onPeriodicEmit(WatermarkOutput output) {final org.apache.flink.streaming.api.watermark.Watermark next wms.getCurrentWatermark();if (next ! null) {output.emitWatermark(new Watermark(next.getTimestamp()));}} 其中 WatermarkOutput output.emitWatermark(new Watermark(next.getTimestamp()))代码如下public DeferredOutput(OutputState state) {this.state state;}Overridepublic void emitWatermark(Watermark watermark) {state.setWatermark(watermark.getTimestamp());} 所以这里最终效果只是对应state(kafka分区[注意一个算子任务有可能消费好几个kafka分区])上设置了水位线/*** Returns true if the watermark was advanced, that is if the new watermark is larger than* the previous one.** pSetting a watermark will clear the idleness flag.*/public boolean setWatermark(long watermark) {this.idle false;final boolean updated watermark this.watermark;// 这里也可以看出来即使代码里面发送了更小值的水位线水位线也不会回退this.watermark Math.max(watermark, this.watermark);return updated;} 4.对应算子任务组合当前任务消费的所有分区水位线的方法 private void updateCombinedWatermark() {long minimumOverAllOutputs Long.MAX_VALUE;boolean hasOutputs false;boolean allIdle true;for (OutputState outputState : watermarkOutputs) {if (!outputState.isIdle()) {minimumOverAllOutputs Math.min(minimumOverAllOutputs, outputState.getWatermark());allIdle false;}hasOutputs true;}// if we dont have any outputs minimumOverAllOutputs is not valid, its still// at its initial Long.MAX_VALUE state and we must not emit that// 如果算子任务不消费任何分区它不会发出任何水位线这里是不是就是kafka消费者要小于kafka主题的原因所在if (!hasOutputs) {return;}if (allIdle) {// 如果当前算子任务处于空闲时间标识空闲以便后续算子可以继续推进underlyingOutput.markIdle();} else if (minimumOverAllOutputs combinedWatermark) {combinedWatermark minimumOverAllOutputs;underlyingOutput.emitWatermark(new Watermark(minimumOverAllOutputs));}}