kafka数据源Flink Kafka Consumer分析

这篇文章主要讲解了"kafka数据源Flink Kafka Consumer分析"，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习"kafka数据源Flink Kafka Consumer分析"吧！

一、open()方法调用时机

FlinkKafkaConsumer继承自RichFunction，具有生命周期方法open()。那么flink是何时调用FlinkKafkaConsumer的open()方法呢？

StreamTask在调用算子程序之前，会执行beforeInvoke()方法，在该方法中会初始化算子的算子并且执行open()方法：

    operatorChain.initializeStateAndOpenOperators(createStreamTaskStateInitializer());

initializeStateAndOpenOperators()方法中循环对算子初始化：

       protected void initializeStateAndOpenOperators(StreamTaskStateInitializer streamTaskStateInitializer) throws Exception {                for (StreamOperatorWrapper operatorWrapper : getAllOperators(true)) {                        StreamOperator operator = operatorWrapper.getStreamOperator();                        operator.initializeState(streamTaskStateInitializer);                        operator.open();                }        }

kafka source对应的operator为StreamSource，其open()方法为

      public void open() throws Exception {                super.open();                FunctionUtils.openFunction(userFunction, new Configuration());        }

FunctionUtils的openFunction()即执行算子（要继承RichFunction）的open()方法：

 public static void openFunction(Function function, Configuration parameters) throws Exception{                if (function instanceof RichFunction) {                        RichFunction richFunction = (RichFunction) function;                        richFunction.open(parameters);                }        }

二、运行时上下文RuntimeContext何时赋值？

在 StreamTask.beforeInvoke() -> new OperatorChain() -> StreamOperatorFactoryUtil.createOperator()，在OperatorChain的构造函数中，通过工厂类StreamOperatorFactory来创建StreamOperator。kafka source对应的StreamOperatorFactory为SimpleOperatorFactory，createStreamOperator()方法中调用StreamOperator的setup()方法：

 public > T createStreamOperator(StreamOperatorParameters parameters) {                if (operator instanceof AbstractStreamOperator) {                        ((AbstractStreamOperator) operator).setProcessingTimeService(processingTimeService);                }                if (operator instanceof SetupableStreamOperator) {                        ((SetupableStreamOperator) operator).setup(                                parameters.getContainingTask(),                                parameters.getStreamConfig(),                                parameters.getOutput());                }                return (T) operator;        }

kafka source对应的StreamOperator为StreamSource，其实现了SetupableStreamOperator接口。其setup方法在父类AbstractUdfStreamOperator：

       public void setup(StreamTask containingTask, StreamConfig config, Output> output) {                super.setup(containingTask, config, output);                FunctionUtils.setFunctionRuntimeContext(userFunction, getRuntimeContext());        }

FunctionUtils.setFunctionRuntimeContext()来给算子设置RuntimeContext。设置的RuntimeContext在AbstractStreamOperator的setup()方法中，为StreamingRuntimeContext：

          this.runtimeContext = new StreamingRuntimeContext(                        environment,                        environment.getAccumulatorRegistry().getUserMap(),                        getMetricGroup(),                        getOperatorID(),                        getProcessingTimeService(),                        null,                        environment.getExternalResourceInfoProvider());

三、FlinkKafkaConsumer的run()方法

Flink调用FlinkKafkaConsumer的run()方法来生产数据。run()方法的处理逻辑：

①创建KafkaFetcher，来拉取数据

          this.kafkaFetcher = createFetcher(                                sourceContext,                                subscribedPartitionsToStartOffsets,                                watermarkStrategy,                                (StreamingRuntimeContext) getRuntimeContext(),                                offsetCommitMode,                                getRuntimeContext().getMetricGroup().addGroup(KAFKA_CONSUMER_METRICS_GROUP),                                useMetrics);

②KafkaFetcher的runFetchLoop()中创建KafkaConsumerThread线程来循环拉取kafka数据。KafkaConsumerThread通过KafkaConsumer拉取kafka数据，并交给Handover

                          if (records == null) {                                        try {                                                records = consumer.poll(pollTimeout);                                        }                                        catch (WakeupException we) {                                                continue;                                        }                                }                                try {                                        handover.produce(records);                                        records = null;                                }

KafkaFetcher通过Handover获取拉取的kafka数据

                   while (running) {                                // this blocks until we get the next records                                // it automatically re-throws exceptions encountered in the consumer thread                                final ConsumerRecords records = handover.pollNext();                                // get the records for each topic partition                                for (KafkaTopicPartitionState partition : subscribedPartitionStates()) {                                        List> partitionRecords =                                                records.records(partition.getKafkaPartitionHandle());                                        partitionConsumerRecordsHandler(partitionRecords, partition);                                }                        }

③通过SourceContext中的Output>来发送数据给下一个算子

             public void collect(T element) {                        synchronized (lock) {                                output.collect(reuse.replace(element));                        }                }

SourceContext在StreamSource的run()方法中通过StreamSourceContexts.getSourceContext()创建。Output>在OperatorChain的createOutputCollector()创建，为其返回值。

           for (StreamEdge outputEdge : operatorConfig.getNonChainedOutputs(userCodeClassloader)) {                        @SuppressWarnings("unchecked")                        RecordWriterOutput output = (RecordWriterOutput) streamOutputs.get(outputEdge);                        allOutputs.add(new Tuple2<>(output, outputEdge));                }

当有一个输出时，是RecordWriterOutput；多个时，是CopyingDirectedOutput或DirectedOutput

④单个输出RecordWriterOutput时，是通过成员属性RecordWriter实例来输出。RecordWriter通过StreamTask的createRecordWriterDelegate()创建，RecordWriterDelegate为RecordWriter的代理类，内部持有RecordWriter实例：

      public static  RecordWriterDelegate>> createRecordWriterDelegate(                        StreamConfig configuration,                        Environment environment) {                List>>> recordWrites = createRecordWriters(                        configuration,                        environment);                if (recordWrites.size() == 1) {                        return new SingleRecordWriter<>(recordWrites.get(0));                } else if (recordWrites.size() == 0) {                        return new NonRecordWriter<>();                } else {                        return new MultipleRecordWriters<>(recordWrites);                }        }        private static  List>>> createRecordWriters(                        StreamConfig configuration,                        Environment environment) {                List>>> recordWriters = new ArrayList<>();                List outEdgesInOrder = configuration.getOutEdgesInOrder(environment.getUserClassLoader());                for (int i = 0; i < outEdgesInOrder.size(); i++) {                        StreamEdge edge = outEdgesInOrder.get(i);                        recordWriters.add(                                createRecordWriter(                                        edge,                                        i,                                        environment,                                        environment.getTaskInfo().getTaskName(),                                        edge.getBufferTimeout()));                }                return recordWriters;        }

outEdgesInOrder来源于StreamGraph中的StreamNode的List outEdges。

创建RecordWriter时，根据StreamEdge的StreamPartitioner outputPartitioner的isBroadcast()方法判断是BroadcastRecordWriter还是ChannelSelectorRecordWriter：

     public RecordWriter build(ResultPartitionWriter writer) {                if (selector.isBroadcast()) {                        return new BroadcastRecordWriter<>(writer, timeout, taskName);                } else {                        return new ChannelSelectorRecordWriter<>(writer, selector, timeout, taskName);                }        }

outputPartitioner是根据上下游节点并行度是否一致来确定：

                 if (partitioner == null && upstreamNode.getParallelism() == downstreamNode.getParallelism()) {                                partitioner = new ForwardPartitioner