一文搞定 Flink 消费消息的全流程

FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("canal_monitor_order_astable", new SimpleStringSchema(), properties);
		consumer.setStartFromEarliest();
		
env.addSource(consumer).flatMap(...).print()

......
/*
			这个方法就是用户代码所真正被执行的入口。比如我们写的什么 new MapFunction() 的逻辑，最终就是在这里被执行的
			 */
			// run the invokable
			invokable.invoke();
......

......
run();
.....

@Override
	// source task 获取数据的入口方法
	protected void run() throws Exception {
		headOperator.run(getCheckpointLock(), getStreamStatusMaintainer());
	}

......
	// 生成上下文之后，接下来就是把上下文交给 SourceFunction 去执行,用户自定义的 run 方法开始正式运行
			userFunction.run(ctx);
......

private void emit(T record, int targetChannel) throws IOException, InterruptedException {
		// 最底层的抽象是 MemorySegment，用于数据传输的是 Buffer，将 java 对象转化为 buffer 是这个
		// Flink 把对象调用该对象所属的序列化器序列化为字节数组
		serializer.serializeRecord(record);
		
		if (copyFromSerializerToTargetChannel(targetChannel)) {
			serializer.prune();
		}
	}

......
if (cause != null) {
						ErrorResponse msg = new ErrorResponse(
							new ProducerFailedException(cause),
							reader.getReceiverId());

						// 真正往 netty 的 nio 通道里写入.
						// 在这里，写入的是一个 RemoteInputChannel，对应的就是下游节点的 InputGate 的 channels。
						ctx.writeAndFlush(msg);
					}
......

// nio 通道的另一端( 下游 )需要读入 buffer
	// 上游的算子写入，下游的算子读取，这也是反压的原理
	// 为什么叫 decodeMsg，主要上游传过来的是 byte 数组，这个将 byte 数组 转化为 record
	private void decodeMsg(Object msg) throws Throwable {
		final Class<?> msgClazz = msg.getClass();

		// ---- Buffer --------------------------------------------------------
		if (msgClazz == NettyMessage.BufferResponse.class) {
			NettyMessage.BufferResponse bufferOrEvent = (NettyMessage.BufferResponse) msg;

			RemoteInputChannel inputChannel = inputChannels.get(bufferOrEvent.receiverId);
			if (inputChannel == null) {
				bufferOrEvent.releaseBuffer();

				cancelRequestFor(bufferOrEvent.receiverId);

				return;
			}

			decodeBufferOrEvent(inputChannel, bufferOrEvent);

		} else if (msgClazz == NettyMessage.ErrorResponse.class) {
			// ---- Error ---------------------------------------------------------
			NettyMessage.ErrorResponse error = (NettyMessage.ErrorResponse) msg;

			SocketAddress remoteAddr = ctx.channel().remoteAddress();

			if (error.isFatalError()) {
				notifyAllChannelsOfErrorAndClose(new RemoteTransportException(
					"Fatal error at remote task manager '" + remoteAddr + "'.",
					remoteAddr,
					error.cause));
			} else {
				RemoteInputChannel inputChannel = inputChannels.get(error.receiverId);

				if (inputChannel != null) {
					if (error.cause.getClass() == PartitionNotFoundException.class) {
						inputChannel.onFailedPartitionRequest();
					} else {
						inputChannel.onError(new RemoteTransportException(
							"Error at remote task manager '" + remoteAddr + "'.",
							remoteAddr,
							error.cause));
					}
				}
			}
		} else {
			throw new IllegalStateException("Received unknown message from producer: " + msg.getClass());
		}
	}

......
/*
			这个方法就是用户代码所真正被执行的入口。比如我们写的什么 new MapFunction() 的逻辑，最终就是在这里被执行的
			 */
			// run the invokable
			invokable.invoke();
......

......
run();
.....

	@Override
	protected void run() throws Exception {
		// cache processor reference on the stack, to make the code more JIT friendly
		final StreamInputProcessor<IN> inputProcessor = this.inputProcessor;
		//处理输入的消息
		while (running && inputProcessor.processInput()) {
			// all the work happens in the "processInput" method
		}
	}

//            程序首先获取下一个 buffer
			// 主要是尝试获取 buffer，然后赋值给当前的反序列化器
			// 处理 barrier 的逻辑，被包含在了getNextNonBlocked 中
			final BufferOrEvent bufferOrEvent = barrierHandler.getNextNonBlocked();
			if (bufferOrEvent != null) {
				if (bufferOrEvent.isBuffer()) {
					currentChannel = bufferOrEvent.getChannelIndex();
					currentRecordDeserializer = recordDeserializers[currentChannel];
					currentRecordDeserializer.setNextBuffer(bufferOrEvent.getBuffer());
				}
				else {
					// Event received
					final AbstractEvent event = bufferOrEvent.getEvent();
					if (event.getClass() != EndOfPartitionEvent.class) {
						throw new IOException("Unexpected event: " + event);
					}
				}
			}

// 这里就是真正的，用户的代码即将被执行的地方
						// now we can do the actual processing
						StreamRecord<IN> record = recordOrMark.asRecord();
						synchronized (lock) {
							numRecordsIn.inc();
							//set KeyContext setCurrentKey
							streamOperator.setKeyContextElement1(record);
							streamOperator.processElement(record);
						}
						return true;