這篇文章主要講解了“kafka數(shù)據(jù)源Flink Kafka Consumer分析”，文中的講解內(nèi)容簡單清晰，易于學(xué)習(xí)與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學(xué)習(xí)“kafka數(shù)據(jù)源Flink Kafka Consumer分析”吧！

成都創(chuàng)新互聯(lián)公司秉承專業(yè)、誠信、服務(wù)、進(jìn)取的價值觀，堅持以客戶為中心、客戶至上的服務(wù)理念，以“關(guān)注企業(yè)需求，實現(xiàn)企業(yè)價值”為導(dǎo)向，努力為企業(yè)提供全面優(yōu)質(zhì)的互聯(lián)網(wǎng)應(yīng)用服務(wù)。服務(wù)包括域名注冊、網(wǎng)絡(luò)空間、企業(yè)郵箱、網(wǎng)站建設(shè)、網(wǎng)站優(yōu)化、網(wǎng)絡(luò)營銷解決方案和咨詢服務(wù)，以幫助企業(yè)客戶應(yīng)用互聯(lián)網(wǎng)。

一、open()方法調(diào)用時機

FlinkKafkaConsumer繼承自RichFunction，具有生命周期方法open()。那么flink是何時調(diào)用FlinkKafkaConsumer的open()方法呢？

StreamTask在調(diào)用算子程序之前，會執(zhí)行beforeInvoke()方法，在該方法中會初始化算子的算子并且執(zhí)行open()方法：

	operatorChain.initializeStateAndOpenOperators(createStreamTaskStateInitializer());

initializeStateAndOpenOperators()方法中循環(huán)對算子初始化：

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

kafka source對應(yīng)的operator為StreamSource，其open()方法為

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

FunctionUtils的openFunction()即執(zhí)行算子（要繼承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的構(gòu)造函數(shù)中，通過工廠類StreamOperatorFactory來創(chuàng)建StreamOperator。kafka source對應(yīng)的StreamOperatorFactory為SimpleOperatorFactory，createStreamOperator()方法中調(diào)用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對應(yīng)的StreamOperator為StreamSource，其實現(xiàn)了SetupableStreamOperator接口。其setup方法在父類AbstractUdfStreamOperator：

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

	}

FunctionUtils.setFunctionRuntimeContext()來給算子設(shè)置RuntimeContext。設(shè)置的RuntimeContext在AbstractStreamOperator的setup()方法中，為StreamingRuntimeContext：

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

三、FlinkKafkaConsumer的run()方法

Flink調(diào)用FlinkKafkaConsumer的run()方法來生產(chǎn)數(shù)據(jù)。run()方法的處理邏輯：

①創(chuàng)建KafkaFetcher，來拉取數(shù)據(jù)

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

②KafkaFetcher的runFetchLoop()中創(chuàng)建KafkaConsumerThread線程來循環(huán)拉取kafka數(shù)據(jù)。KafkaConsumerThread通過KafkaConsumer拉取kafka數(shù)據(jù)，并交給Handover

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

				try {
					handover.produce(records);
					records = null;
				}

KafkaFetcher通過Handover獲取拉取的kafka數(shù)據(jù)

			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>來發(fā)送數(shù)據(jù)給下一個算子

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

SourceContext在StreamSource的run()方法中通過StreamSourceContexts.getSourceContext()創(chuàng)建。Output>在OperatorChain的createOutputCollector()創(chuàng)建，為其返回值。

		for (StreamEdge outputEdge : operatorConfig.getNonChainedOutputs(userCodeClassloader)) {
			@SuppressWarnings("unchecked")
			RecordWriterOutput output = (RecordWriterOutput) streamOutputs.get(outputEdge);

			allOutputs.add(new Tuple2<>(output, outputEdge));
		}

當(dāng)有一個輸出時，是RecordWriterOutput；多個時，是CopyingDirectedOutput或DirectedOutput

④單個輸出RecordWriterOutput時，是通過成員屬性RecordWriter實例來輸出。RecordWriter通過StreamTask的createRecordWriterDelegate()創(chuàng)建，RecordWriterDelegate為RecordWriter的代理類，內(nèi)部持有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。

創(chuàng)建RecordWriter時，根據(jù)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是根據(jù)上下游節(jié)點并行度是否一致來確定：

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