PermanentBlobService permanentBlobService,        MetricGroup taskManagerMetricGroup,        ExecutorService ioExecutor,        ScheduledExecutor scheduledExecutor,        FatalErrorHandler fatalErrorHandler,        WorkingDirectory workingDirectory)         throws Exception {    ...     final ShuffleEnvironment<?, ?> shuffleEnvironment =            createShuffleEnvironment(                    taskManagerServicesConfiguration,                    taskEventDispatcher,                    taskManagerMetricGroup,                    ioExecutor,                    scheduledExecutor);     final int listeningDataPort = shuffleEnvironment.start();

       NettyBufferPool bufferPool,        ResultPartitionProvider partitionProvider,        TaskEventPublisher taskEventPublisher,        NettyConfig nettyConfig,         boolean connectionReuseEnabled) {     this .server = new NettyServer (nettyConfig);     this .client = new NettyClient (nettyConfig);     this .bufferPool = checkNotNull(bufferPool);     this .partitionRequestClientFactory =    eqf.baijinpai.cn         new PartitionRequestClientFactory (                    client, nettyConfig.getNetworkRetries(), connectionReuseEnabled);     this .nettyProtocol =             new NettyProtocol (                    checkNotNull(partitionProvider), checkNotNull(taskEventPublisher));

    // Add the server port number to the name in order to distinguish     // multiple clients running on the same host.     String name =            NettyConfig.CLIENT_THREAD_GROUP_NAME + " (" + config.getServerPortRange() + ")" ;     EpollEventLoopGroup epollGroup =             new EpollEventLoopGroup (                    config.getClientNumThreads(), NettyServer.getNamedThreadFactory(name));    bootstrap.group(epollGroup).channel(EpollSocketChannel.class);    config.getTcpKeepIdleInSeconds()            .ifPresent(idle -> bootstrap.option(EpollChannelOption.TCP_KEEPIDLE, idle));    config.getTcpKeepInternalInSeconds()            .ifPresent(                    interval -> bootstrap.option(EpollChannelOption.TCP_KEEPINTVL, interval));    config.getTcpKeepCount()owqbh.baijinpai.cn

初始化过程重要设置 EventLoopGroup 和 channel，可以用 epoll 的话就用 epoll，否则就用 nio。设置好这些后就是设置了一些通道参数（连接超时时间、Bufffer 池等）。

到这里 client 的初始化其实并没有结束，还需要设置 Handler 流水线，这些工作是在 Task 启动时执行了。

server 初始化  

server 的初始化过程是先创建并初始化了 ServerBootstrap。之后同样也是设置 EventLoopGroup 和 channel，以及通道相关的各种参数。

public ChannelHandler[] getServerChannelHandlers() {     PartitionRequestQueue queueOfPartitionQueues = new PartitionRequestQueue ();     PartitionRequestServerHandler serverHandler =             new PartitionRequestServerHandler (                    partitionProvider, taskEventPublisher, queueOfPartitionQueues);     return new ChannelHandler [] {        messageEncoder,         new NettyMessage .NettyMessageDecoder(),

lmrpa.baijinpai.cn

  Integer port = portsIterator.next();    LOG.debug( "Trying to bind Netty server to port: {}" , port);    bootstrap.localAddress(config.getServerAddress(), port);     try {        bindFuture = bootstrap.bind().syncUninterruptibly();    } catch (Exception e) {         // syncUninterruptibly() throws checked exceptions via Unsafe         // continue if the exception is due to the port being in use, fail early         // otherwise         if (isBindFailure(e)) {            LOG.debug( "Failed to bind Netty server" , e);        } else {             throw e;        }    }a.baijinpai.cn } if (bindFuture == null ) {     throw new BindException (             "Could not start rest endpoint on any port in port range "

for (InputGate inputGate : inputGates) {    recoveredFutures.add(inputGate.getStateConsumedFuture());    inputGate            .getStateConsumedFuture()            .thenRun(                    () ->                            mainMailboxExecutor.execute(                                    inputGate::requestPartitions,                                     "Input gate request partitions" )); } private void internalRequestPartitions () {     for (InputChannel inputChannel : inputChannels()) {         try {            inputChannel.requestSubpartitions();        } catch (Throwable t) {  zffar.baijinpai.cn            inputChannel.setError(t);             return ;

public void requestSubpartitions () throws IOException, InterruptedException {     if (partitionRequestClient == null ) {        LOG.debug(                 "{}: Requesting REMOTE subpartitions {} of partition {}. {}" ,                 this ,                consumedSubpartitionIndexSet,                partitionId,                channelStatePersister);         // Create a client and request the partition         try {            partitionRequestClient =                    connectionManager.createPartitionRequestClient(connectionId);        } catch (IOException e) {             // IOExceptions indicate that we could not open a connection to the remote             // TaskExecutor             throw new PartitionConnectionException (partitionId, e);        }te786.baijinpai.cn

public ChannelHandler[] getClientChannelHandlers() {     NetworkClientHandler networkClientHandler = new CreditBasedPartitionRequestClientHandler ();     return new ChannelHandler [] {        messageEncoder,         new NettyMessageClientDecoderDelegate (networkClientHandler),        networkClientHandler   s5iup.baijinpai.cn

if (reader.isRegisteredAsAvailable()) {         return ;    }    ResultSubpartitionView. AvailabilityWithBacklog availabilityWithBacklog =            reader.getAvailabilityAndBacklog();     if (!availabilityWithBacklog.isAvailable()) {         int backlog = availabilityWithBacklog.getBacklog();         if (backlog > 0 && reader.needAnnounceBacklog()) {            announceBacklog(reader, backlog);m5q19.baijinpai.cn        }         return ;  eqbeb.baijinpai.cn    }     // Queue an available reader for consumption. If the queue is empty,     // we try trigger the actual write. Otherwise this will be handled by     // the writeAndFlushNextMessageIfPossible calls.     boolean triggerWrite = availableReaders.isEmpty();wilvx.baijinpai.cn

public BufferAndAvailability getNextBuffer () throws IOException {     BufferAndBacklog next = subpartitionView.getNextBuffer();     if (next != null ) {  bpokt.baijinpai.cn        if (next.buffer().isBuffer() && --numCreditsAvailable < 0 ) {      xelzr.baijinpai.cn       throw new IllegalStateException ( "no credit available" );        }         final Buffer. DataType nextDataType = getNextDataType(next);   41n96.baijinpai.cn      return new BufferAndAvailability (                next.buffer(), nextDataType, next.buffersInBacklog(), next.getSequenceNumber());

assert Thread.holdsLock(inputChannelsWithData);     if (channelsWithEndOfPartitionEvents.get(channel.getChannelIndex())) {         return false ;   mylvcy.baijinpai.cn    }     final boolean alreadyEnqueued =  idlfr.baijinpai.cn            enqueuedInputChannelsWithData.get(channel.getChannelIndex());     if (alreadyEnqueued            && (!priority || inputChannelsWithData.containsPriorityElement(channel))) {         // already notified / prioritized (double notification), ignore   te786.baijinpai.cn         return false ;  zczat.baijinpai.cn    }     // 当前 inputChannel 加入到 inputChannelsWithData    inputChannelsWithData.add(channel, priority, alreadyEnqueued);     if (!alreadyEnqueued) {