• 获取数据 交给 Operator 处理

继承

  1. public final class StreamTaskNetworkInput<T> implements StreamTaskInput<T> { ... }

核心方法

获取数据

  1. @Override
  2.     public InputStatus emitNext(DataOutput<T> output) throws Exception {
  4.         while (true) {
  5.             // get the stream element from the deserializer
  6.             // 从反序列化器获取流元素
  7.             if (currentRecordDeserializer != null) {
  8.                 DeserializationResult result;
  9.                 try {
  10.                     // TODO 当前记录反序列化器 获取下一条数据
  11.                     result = currentRecordDeserializer.getNextRecord(deserializationDelegate);
  12.                 } catch (IOException e) {
  13.                     throw new IOException(
  14.                             String.format("Can't get next record for channel %s", lastChannel), e);
  15.                 }
  16.                 // TODO 缓冲区已消耗
  17.                 if (result.isBufferConsumed()) {
  18.                     currentRecordDeserializer.getCurrentBuffer().recycleBuffer();
  19.                     currentRecordDeserializer = null;
  20.                 }
  22.                 // TODO 是完整记录
  23.                 if (result.isFullRecord()) {
  24.                     processElement(deserializationDelegate.getInstance(), output);
  25.                     return InputStatus.MORE_AVAILABLE;
  26.                 }
  27.             }
  29.             // TODO 从 InputGate 中获取 下一个 Buffer
  30.             Optional<BufferOrEvent> bufferOrEvent = checkpointedInputGate.pollNext();
  32.             // 如果是 Buffer, 则设置 RecordDeserializer 处理下一条记录,否则释放 RecordDeserializer
  33.             if (bufferOrEvent.isPresent()) {
  34.                 // return to the mailbox after receiving a checkpoint barrier to avoid processing of
  35.                 // data after the barrier before checkpoint is performed for unaligned checkpoint
  36.                 // mode
  37.                 if (bufferOrEvent.get().isBuffer()) {
  38.                     processBuffer(bufferOrEvent.get());
  39.                 } else {
  40.                     processEvent(bufferOrEvent.get());
  41.                     return InputStatus.MORE_AVAILABLE;
  42.                 }
  43.             } else {
  44.                 if (checkpointedInputGate.isFinished()) {
  45.                     checkState(
  46.                             checkpointedInputGate.getAvailableFuture().isDone(),
  47.                             "Finished BarrierHandler should be available");
  48.                     return InputStatus.END_OF_INPUT;
  49.                 }
  50.                 return InputStatus.NOTHING_AVAILABLE;
  51.             }
  52.         }
  53.     }

不同的数据走不同的 处理模式

  1.  // TODO 不同的数据走不同的 处理模式
  2.     private void processElement(StreamElement recordOrMark, DataOutput<T> output) throws Exception {
  3.         // 如果是 WaterMark 走这边
  4.         if (recordOrMark.isRecord()) {
  5.             // output  发出记录
  6.             output.emitRecord(recordOrMark.asRecord());
  8.         } else if (recordOrMark.isWatermark()) {
  9.             statusWatermarkValve.inputWatermark(
  10.                     recordOrMark.asWatermark(), flattenedChannelIndices.get(lastChannel), output);
  12.         } else if (recordOrMark.isLatencyMarker()) {
  13.             output.emitLatencyMarker(recordOrMark.asLatencyMarker());
  15.         } else if (recordOrMark.isStreamStatus()) {
  16.             statusWatermarkValve.inputStreamStatus(
  17.                     recordOrMark.asStreamStatus(),
  18.                     flattenedChannelIndices.get(lastChannel),
  19.                     output);
  20.         } else {
  21.             throw new UnsupportedOperationException("Unknown type of StreamElement");
  22.         }
  23.     }

  • output 根据 Watermark 发送数据 ? 需要交给 DataOutput 的 emitRecord 方法

    1.   interface DataOutput<T> {
    3.       void emitRecord(StreamRecord<T> streamRecord) throws Exception;
    5.       void emitWatermark(Watermark watermark) throws Exception;
    7.       void emitStreamStatus(StreamStatus streamStatus) throws Exception;
    9.       void emitLatencyMarker(LatencyMarker latencyMarker) throws Exception;
    10.   }

  • 通过子类具体实现 触发 emitRecord 方法 —-> OneInputStreamTask 的 emitRecord 方法

    1.       // TODO 一个流的 处理逻辑
    2.       @Override
    3.       public void emitRecord(StreamRecord<IN> record) throws Exception {
    4.           numRecordsIn.inc();
    5.           operator.setKeyContextElement1(record);
    6.           operator.processElement(record);
    7.       }

  • 这里的 operator.processElement(record); 通过 Input 的子类 触发了 WindowOperator 的 ProcessElement 计算

    1. // TODO 处理元素
    2.   @Override
    3.   public void processElement(StreamRecord<IN> element) throws Exception { ...}