@Override
protected void initChannel(Channel channel) {
    ChannelPipeline pipeline = channel.pipeline();

    /*
     * 设置reactiveBridge通道处理器
     * 第二个参数是连接操作提供者，会用来创建连接操作对象。
     * 对于HttpServerConfig而言，没有实现TransportConfig中的方法，而后者的默认实现是返回null。
     */
    ChannelOperations.addReactiveBridge(channel, config.channelOperationsProvider(), connectionObserver);

    // 追加通道初始化操作
    config.defaultOnChannelInit()
            // 添加配置中的初始化器
          .then(config.doOnChannelInit)
            // 执行通道初始化
          .onChannelInit(connectionObserver, channel, remoteAddress);

    // 从pipeline中移除该handler，因为初始化器执行一次就够了。
    pipeline.remove(this);

    if (log.isDebugEnabled()) {
        log.debug(format(channel, "Initialized pipeline {}"), pipeline.toString());
    }
}

public static void addReactiveBridge(Channel ch, OnSetup opsFactory, ConnectionObserver listener) {
    requireNonNull(ch, "channel");
    requireNonNull(opsFactory, "opsFactory");
    requireNonNull(listener, "listener");
    // 注册通道处理器
    ch.pipeline()
      .addLast(NettyPipeline.ReactiveBridge, new ChannelOperationsHandler(opsFactory, listener));
}

configureHttp11Pipeline(
        channel.pipeline(),
        accessLogEnabled,
        accessLog,
        compressPredicate(compressPredicate, minCompressionSize),
        cookieDecoder,
        cookieEncoder,
        false,
        decoder,
        formDecoderProvider,
        forwardedHeaderHandler,
        httpMessageLogFactory,
        idleTimeout,
        observer,
        mapHandle,
        maxKeepAliveRequests,
        methodTagValue,
        metricsRecorder,
        minCompressionSize,
        readTimeout,
        requestTimeout,
        uriTagValue);

@SuppressWarnings("deprecation")
static void configureHttp11Pipeline(ChannelPipeline p,
        boolean accessLogEnabled,
        @Nullable Function<AccessLogArgProvider, AccessLog> accessLog,
        @Nullable BiPredicate<HttpServerRequest, HttpServerResponse> compressPredicate,
        ServerCookieDecoder cookieDecoder,
        ServerCookieEncoder cookieEncoder,
        boolean channelOpened,
        HttpRequestDecoderSpec decoder,
        HttpServerFormDecoderProvider formDecoderProvider,
        @Nullable BiFunction<ConnectionInfo, HttpRequest, ConnectionInfo> forwardedHeaderHandler,
        HttpMessageLogFactory httpMessageLogFactory,
        @Nullable Duration idleTimeout,
        ConnectionObserver listener,
        @Nullable BiFunction<? super Mono<Void>, ? super Connection, ? extends Mono<Void>> mapHandle,
        int maxKeepAliveRequests,
        @Nullable Function<String, String> methodTagValue,
        @Nullable ChannelMetricsRecorder metricsRecorder,
        int minCompressionSize,
        @Nullable Duration readTimeout,
        @Nullable Duration requestTimeout,
        @Nullable Function<String, String> uriTagValue) {
    HttpDecoderConfig decoderConfig = new HttpDecoderConfig();
    decoderConfig.setMaxInitialLineLength(decoder.maxInitialLineLength())
                 .setMaxHeaderSize(decoder.maxHeaderSize())
                 .setMaxChunkSize(decoder.maxChunkSize())
                 .setValidateHeaders(decoder.validateHeaders())
                 .setInitialBufferSize(decoder.initialBufferSize())
                 .setAllowDuplicateContentLengths(decoder.allowDuplicateContentLengths());
    p.addBefore(NettyPipeline.ReactiveBridge,
                NettyPipeline.HttpCodec,
                new HttpServerCodec(decoderConfig))
     .addBefore(NettyPipeline.ReactiveBridge, // 添加在ReactiveBridge之前
                NettyPipeline.HttpTrafficHandler,
                // 创建通道处理器
                new HttpTrafficHandler(compressPredicate, cookieDecoder, cookieEncoder, formDecoderProvider,
                        forwardedHeaderHandler, httpMessageLogFactory, idleTimeout, listener, mapHandle, maxKeepAliveRequests,
                        readTimeout, requestTimeout));

}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {

            HttpServerOperations ops;

                // 创建HTTP服务的操作对象
                ops = new HttpServerOperations(Connection.from(ctx.channel()),
                        listener,
                        request,
                        compress,
                        connectionInfo,
                        cookieDecoder,
                        cookieEncoder,
                        formDecoderProvider,
                        httpMessageLogFactory,
                        false,
                        mapHandle,
                        readTimeout,
                        requestTimeout,
                        secure,
                        timestamp);

            // 绑定到通道对象中
            ops.bind();

}

default Connection bind() {
    channel().attr(ReactorNetty.CONNECTION)
             .set(this);
    return this;
}

@Override
@SuppressWarnings("FutureReturnValueIgnored")
public final void channelRead(ChannelHandlerContext ctx, Object msg) {
    if (msg == null || msg == Unpooled.EMPTY_BUFFER || msg instanceof EmptyByteBuf) {
        return;
    }
    try {
        // 这里主要考虑返回值类型是HttpServerOperations
        ChannelOperations<?, ?> ops = ChannelOperations.get(ctx.channel());
        if (ops != null) {
            ops.onInboundNext(ctx, msg);
        }
        else {

            ReferenceCountUtil.release(msg);
        }
    }
    catch (Throwable err) {

    }
}

@Nullable
public static ChannelOperations<?, ?> get(Channel ch) {
    return Connection.from(ch)
                     .as(ChannelOperations.class);
}

@Override
protected void onInboundNext(ChannelHandlerContext ctx, Object msg) {
    if (msg instanceof HttpRequest) {

        try {
            // 接收请求
            listener().onStateChange(this, HttpServerState.REQUEST_RECEIVED);
        }

        return;
    }

}

@Override
public void onStateChange(Connection connection, State newState) {
    if (newState == State.DISCONNECTING) {
        if (connection.channel().isActive() && !connection.isPersistent()) {
            connection.dispose();
        }
    }

    // 处理连接状态
    childObs.onStateChange(connection, newState);
}

DisposableServer startHttpServer() {
    HttpServer server = this.httpServer;
    // 下面这个条件默认是true
    if (this.routeProviders.isEmpty()) {
        // 设置请求处理器
        server = server.handle(this.handler);
    }
    else {
        server = server.route(this::applyRouteProviders);
    }
    if (this.lifecycleTimeout != null) {
        return server.bindNow(this.lifecycleTimeout);
    }
    /*
     * 执行bind操作，HttpServer是reactor-netty提供的，所以离开了spring-boot范畴，
     * 最后会一直执行到netty层面的ServerBootstrap#bind()
     */
    return server.bindNow();
}

public final HttpServer handle(
        BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler) {
    Objects.requireNonNull(handler, "handler");
    return childObserve(new HttpServerHandle(handler));
}

static final class HttpServerHandle implements ConnectionObserver {

    final BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler;

    HttpServerHandle(BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler) {
        this.handler = handler;
    }

    @Override
    @SuppressWarnings("FutureReturnValueIgnored")
    public void onStateChange(Connection connection, State newState) {
        if (newState == HttpServerState.REQUEST_RECEIVED) {
            try {
                if (log.isDebugEnabled()) {
                    log.debug(format(connection.channel(), "Handler is being applied: {}"), handler);
                }
                // 封装连接
                HttpServerOperations ops = (HttpServerOperations) connection;
                // 调用apply方法，处理连接
                Publisher<Void> publisher = handler.apply(ops, ops);
                Mono<Void> mono = Mono.deferContextual(ctx -> {
                    ops.currentContext = Context.of(ctx);
                    return Mono.fromDirect(publisher);
                });
                if (ops.mapHandle != null) {
                    mono = ops.mapHandle.apply(mono, connection);
                }
                mono.subscribe(ops.disposeSubscriber());
            }
            catch (Throwable t) {
                log.error(format(connection.channel(), ""), t);
                //"FutureReturnValueIgnored" this is deliberate
                connection.channel()
                          .close();
            }
        }
    }
}