StreamMap.processElement  方法中，调用完 map 方法之后，就会调用  output.collect  方法将数据输出，这里的 output 就是 RecordWriterOutput。在 RecordWriterOutput 中，会调用 RecordWriter 的 emit 方法。

private <X> void pushToRecordWriter (StreamRecord<X> record) {    serializationDelegate.setInstance(record);     try {        recordWriter.emit(serializationDelegate);    } catch (IOException e) {         throw new UncheckedIOException (e.getMessage(), e);

这里的 serializationDelegate 是用来对 record 进行序列化的。RecordWriter 有两个实现类，一个是 ChannelSelectorRecordWriter，另一个是 BroadcastRecordWriter。ChannelSelectorRecordWriter 需要先调用 ChannelSelector 选择对应的 subparition，然后进行写入。BroadcastRecordWriter 则是写到所有的 subparition。

public void emitRecord (ByteBuffer record, int targetSubpartition) throws IOException {    totalWrittenBytes += record.remaining();     BufferBuilder buffer = appendUnicastDataForNewRecord(record, targetSubpartition);     while (record.hasRemaining()) {         // full buffer, partial record        finishUnicastBufferBuilder(targetSubpartition);        buffer = appendUnicastDataForRecordContinuation(record, targetSubpartition);    }     if (buffer.isFull()) {         // full buffer, full record        finishUnicastBufferBuilder(targetSubpartition);    }

private MemorySegment requestMemorySegment ( int targetChannel) {     MemorySegment segment = null ;     synchronized (availableMemorySegments) {        checkDestroyed();         if (!availableMemorySegments.isEmpty()) {            segment = availableMemorySegments.poll();        } else if (isRequestedSizeReached()) {             // Only when the buffer request reaches the upper limit(i.e. current pool size),             // requests an overdraft buffer.            segment = requestOverdraftMemorySegmentFromGlobal();        }         if (segment == null ) {             return null ;        }         if (targetChannel != UNKNOWN_CHANNEL) {             if (++subpartitionBuffersCount[targetChannel] == maxBuffersPerChannel) {                unavailableSubpartitionsCount++;

private void addToSubpartition (        BufferBuilder buffer,         int targetSubpartition,         int partialRecordLength,         int minDesirableBufferSize)  q8m.ttlyf.cn         throws IOException {     int desirableBufferSize =            subpartitions[targetSubpartition].add(                    buffer.createBufferConsumerFromBeginning(), partialRecordLength);

public DataInputStatus emitNext (DataOutput<T> output) throws Exception {     while ( true ) {         // get the stream element from the deserializer    ablzyu.ttlyf.cn           if (currentRecordDeserializer != null ) {            RecordDeserializer.DeserializationResult result;             try {                result = currentRecordDeserializer.getNextRecord(deserializationDelegate);            } catch (IOException e) {                 throw new IOException (                        String.format( "Can't get next record for channel %s" , lastChannel), e);            }             if (result.isBufferConsumed()) {                currentRecordDeserializer = null ;            }             if (result.isFullRecord()) {                 final boolean breakBatchEmitting =                        processElement(deserializationDelegate.getInstance(), output);                 if (canEmitBatchOfRecords.check() && !breakBatchEmitting) {                     continue ;                }  hnbhvd.ttlyf.cn                 return DataInputStatus.MORE_AVAILABLE;            }        }        Optional<BufferOrEvent> bufferOrEvent = checkpointedInputGate.pollNext();         if (bufferOrEvent.isPresent()) {             // return to the mailbox after receiving a checkpoint barrier to avoid processing of             // data after the barrier before checkpoint is performed for unaligned checkpoint             // mode             if (bufferOrEvent.get().isBuffer()) {                processBuffer(bufferOrEvent.get());            } else {                 DataInputStatus status = processEvent(bufferOrEvent.get(), output);                 if (status == DataInputStatus.MORE_AVAILABLE && canEmitBatchOfRecords.check()) {                     continue ;                }                 return status;dcydqe.ttlyf.cn            }        } else {             if (checkpointedInputGate.isFinished()) {                checkState(                        checkpointedInputGate.getAvailableFuture().isDone(),                         "Finished BarrierHandler should be available" );      m.junpin001.com             return DataInputStatus.END_OF_INPUT;

public Optional<BufferAndAvailability> getNextBuffer () throws IOException {    checkError();     if (!toBeConsumedBuffers.isEmpty()) {         return getBufferAndAvailability(toBeConsumedBuffers.removeFirst());    }     ResultSubpartitionView subpartitionView = this .subpartitionView;     if (subpartitionView == null ) {         // There is a possible race condition between writing a EndOfPartitionEvent (1) and         // flushing (3) the Local         // channel on the sender side, and reading EndOfPartitionEvent (2) and processing flush         // notification (4). When         // they happen in that order (1 - 2 - 3 - 4), flush notification can re-enqueue         // LocalInputChannel after (or         // during) it was released during reading the EndOfPartitionEvent (2).         if (isReleased) {             return Optional.empty();wap.gz-lzd.com        }         // this can happen if the request for the partition was triggered asynchronously         // by the time trigger   eyisoa.ttlyf.cn         // would be good to avoid that, by guaranteeing that the requestPartition() and         // getNextBuffer() always come from the same thread         // we could do that by letting the timer insert a special "requesting channel" into the         // input gate's queue   ar83xxm.ttlyf.cn        subpartitionView = checkAndWaitForSubpartitionView();    }     BufferAndBacklog next = subpartitionView.getNextBuffer();     // ignore the empty buffer directly     while (next != null && next.buffer().readableBytes() == 0 ) {        next.buffer().recycleBuffer();        next = subpartitionView.getNextBuffer();        numBuffersIn.inc();dsmtcy.ttlyf.cn    }     if (next == null ) {         if (subpartitionView.isReleased()) {             throw new CancelTaskException (                     "Consumed partition " + subpartitionView + " has been released." );        } else {             return Optional.empty();cs.ttlyf.cn        }xfnz.ttlyf.cn      }     Buffer buffer = next.buffer();     if (buffer instanceof FullyFilledBuffer) {        List<Buffer> partialBuffers = ((FullyFilledBuffer) buffer).getPartialBuffers();         int seq = next.getSequenceNumber();         for (Buffer partialBuffer : partialBuffers) {            toBeConsumedBuffers.add(     mwdvqy.ttlyf.cn                new BufferAndBacklog (                            partialBuffer,  junpin001.com                            next.buffersInBacklog(),                            buffer.getDataType(),                            seq++));  jrszb.ttLyf.cn        }