netty相关概念介绍

  1. Channel：可以抽象的理解为netty中传输数据的管道，但并不对应一个tcp连接，只对应一次数据处理，比如在Http2多路复用的情况下，一个tcp连接就可能存在多个channel，比较常用的是NioSocketChannel，和Java NIO的NioSocketChannel同名，而netty的NioSocketChannel底层正是通过Java NIO的NioSocketChannel实现的，此外还有针对Linux系统的EpollSocketChannel，使用的就是Linux底层非常高效的方法：epoll
  2. EventLoop：EventLoop 是Netty中的事件循环，它负责处理网络事件、执行任务和管理相关的资源。每个 EventLoop 都绑定到一个线程，并且在该线程上运行，特殊情况下也可以是多个线程，总之就是处理一组事件。它使用事件循环机制来处理输入和输出的数据，以及执行用户自定义的任务。EventLoop 提供了一种异步执行任务的机制，以避免阻塞IO操作，每个Channel绑定了一个EventLoop
  3. EventLoopGroup：EventLoopGroup 是一个线程池，它包含一组 EventLoop。它负责管理和分配 EventLoop，用于处理客户端和服务器端的网络事件。EventLoopGroup 可以理解为一个线程池的集合，每个 EventLoop 都是一个独立的线程，用于处理具体的网络事件。
  4. ChannelPipeline：channel事件处理管道，也可以立即为channel事件处理流程，由多个ChannelHandler组成，当有IO读写事件时，如果是读事件，消息会先到达HeadContext（ChannelPipeline的头部），然后一直往下走，经过各种消息处理器，如果是写事件，当前消息处理器创建消息后，会往上走，到达HeadContext，然后发送出去，如果不是消息处理器创建的消息，而是channel直接创建的消息，就从管道尾部出发（TailContext），即入站消息从上往下，出战消息从下往上
  5. ChannelHandler（通道处理器）：可以理解为消息处理器，有两种：ChannelInboundHandler、ChannelOutboundHandler，前者是入站消息处理，当有读事件时，会生效，后者是出战消息处理，当由写消息时，会生效，此外还有一种特殊的双通道处理：ChannelDuplexHandler，读写事件，它都会生效
  6. Bootstrap：可以立即为启动器或者启动辅助器，用于配置Channel、EventLoopGroup、连接信息、连接参数以及其他相关的参数。
  7. ByteBuf：netty中高效的字节容器，支持内存池和零拷贝
  8. Codec：编解码器，一种特殊的ChannelHandler，因为netty本身只支持ByteBuf，所以需要编解码器将ByteBuf转为我们需要的类型，或者将我们生成的类型数据转为ByteBuf交给netty处理，它有两个主要的方法，decode，encode，前者负责解码，将ByteBuf转为我们需要的类型，后者负责编码，将我们生成的类型数据转为ByteBuf，也可以被单独分为两种，decoder(解码器)和encoder(编码器)，前者具有decode方法，实现ChannelInboundHandler，后者具有encode方法，实现ChannelOutboundHandler

服务端代码

依赖

1
2
3
4
5
<dependency>
  <groupId>io.netty</groupId>
  <artifactId>netty-all</artifactId>
  <version>4.1.97.Final</version>
</dependency>

代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
public class Http2UpgradeServer {

    public static void main(String[] args) throws CertificateException, InterruptedException, SSLException {

        // Create event loop groups
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();


        try {
            // Create and configure the server bootstrap
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)//设置处理客户端连接请求线程组以及处理客户端io（读写）请求线程组
                    .channel(NioServerSocketChannel.class)//指定服务器使用的网络通道类型,比如Linux系统的话，可以使用EpollSocketChannel网络通道
                    .option(ChannelOption.SO_BACKLOG, 128) //设置服务端一些属性，如队列最大小，还可以通过此方法设置其他参数值，比如超时时间
                    .childOption(ChannelOption.SO_KEEPALIVE, true)//设置客户端连接一些属性，
                    //每当有新的客户端连进来时，都是新建一个SocketChannel, 在这里设置SocketChannel初始化处理方法，对新的SocketChannel进行一些处理，主要是加入通道处理程序
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            HttpServerCodec httpServerCodec = new HttpServerCodec();//http消息解码以及编码器，由HttpRequestDecoder, HttpResponseEncoder组成
                            Http2FrameCodec http2FrameCodec = Http2FrameCodecBuilder.forServer().build();//http2帧消息解码以及编码器
                            ch.pipeline()
                                    .addLast(new CleartextHttp2ServerUpgradeHandler(
                                            httpServerCodec,
                                            new HttpServerUpgradeHandler(
                                                    httpServerCodec,
                                                    protocol -> protocol.equals("h2c") ? new Http2ServerUpgradeCodec(http2FrameCodec) : null,
                                                    1024),
                                            http2FrameCodec
                                    ))
                                   .addLast(new MessageHandler());
                        }
                    });

            // Bind and start the server
            ChannelFuture future = bootstrap.bind(8080).sync();
            System.out.println("Server started and listening on port 8080");

            // Wait until the server socket is closed
            future.channel().closeFuture().sync();
        } finally {
            // Shut down the event loop groups
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    @ChannelHandler.Sharable
    static class MessageHandler extends HttpObjectAggregator {

        public MessageHandler() {
            super(1024);
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof HttpObject) {
                handleHttpMessage(ctx, (HttpObject) msg);
            } else if (msg instanceof Http2Frame) {
                ctx.pipeline().addLast(new Http2MultiplexHandler(new Http2Handler())).remove(this);
            }
        }

        private void handleHttpMessage(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
            List<Object> out = new ArrayList<>();
            super.decode(ctx, msg, out);
            if (out.isEmpty()) {
                return;
            }
            assert out.size() == 1;
            FullHttpRequest fullHttpRequest = (FullHttpRequest) out.get(0);

            String content = fullHttpRequest.content().toString(CharsetUtil.UTF_8);
            System.out.println(fullHttpRequest.uri() + ":" + content);

            DefaultFullHttpResponse httpResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer(("success:" + content).getBytes()));
            httpResponse.headers()
                    .add("Content-Length", ("success:" + content).length())
                    .add("Content-Type", "text/html;charset=UTF-8");

            ctx.writeAndFlush(httpResponse);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            ctx.close();
        }
    }

    @ChannelHandler.Sharable
    static class Http2Handler extends ChannelInboundHandlerAdapter{
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof DefaultHttp2HeadersFrame) {
                ((DefaultHttp2HeadersFrame) msg).headers().forEach(map -> System.out.println(map.getKey() + ":" + map.getValue()));
            } else if (msg instanceof DefaultHttp2DataFrame) {
                System.out.println("----------------------" + ctx.channel().hashCode());

                DefaultHttp2DataFrame http2RequestDataFrame = (DefaultHttp2DataFrame) msg;

                String requestBody = http2RequestDataFrame.content().toString(CharsetUtil.UTF_8);
                System.out.println(http2RequestDataFrame.stream().id() + ":" + requestBody);

                String responseBody = "success:" + requestBody;
                Http2FrameStream requestStream = http2RequestDataFrame.stream();

                DefaultHttp2HeadersFrame headersFrame = new DefaultHttp2HeadersFrame(
                        new DefaultHttp2Headers().status("200")
                                .set(HttpHeaderNames.CONTENT_TYPE, "text/plain;charset=UTF-8")
                                .set(HttpHeaderNames.DATE, "Fri, 15 Sep 2023 03:00:01 GMT")
                                .set(HttpHeaderNames.CONTENT_LENGTH, String.valueOf(responseBody.length()))
                ).stream(requestStream);
                DefaultHttp2DataFrame dataFrame = new DefaultHttp2DataFrame(Unpooled.wrappedBuffer(responseBody.getBytes()), true).stream(requestStream);

                ctx.write(headersFrame);
                ctx.writeAndFlush(dataFrame);
            } else if (msg instanceof DefaultHttp2PingFrame) {
                if (((DefaultHttp2PingFrame) msg).ack()) {
                    DefaultHttp2PingFrame pingFrame = new DefaultHttp2PingFrame(1);
                    ctx.writeAndFlush(pingFrame);
                }
            }
        }
    }
}

代码解释

  1. CleartextHttp2ServerUpgradeHandler：http升级http2处理类，但并不是真正的升级处理类，有三个参数
    1. HttpServerCodec：http编解码器，由HttpRequestDecoder, HttpResponseEncoder组成，解析升级前的http消息，以及发送升级成功或失败后的响应
    2. HttpServerUpgradeHandler：真正的升级处理类，但它不仅能升级http为http2，还可以将http升级为其他协议，有三个参数
      1. SourceCodec：原始编码处理类（即http编码处理类），代码中使用的是HttpServerCodec，和CleartextHttp2ServerUpgradeHandler的是同一个对象，它在HttpServerUpgradeHandler中并没有真正的作用，因为它是为了CleartextHttp2ServerUpgradeHandler专门设计的
      2. UpgradeCodecFactory：升级工厂，生成升级处理编码类-UpgradeCodec的工厂，这里设计成工厂的目的是可以针对不同的升级协议，返回作用不同的升级处理编码类，这里的升级协议，取自upgrade消息头，升级到Http2用的升级处理编码类是Http2ServerUpgradeCodec，它实现了UpgradeCodec
      3. maxContentLength：http消息体的最大长度
    3. ChannelHandler：如果客户端发来的协议就是Http2的话，就会直接通过这个类处理，而不再经过上面的HttpServerCodec和HttpServerUpgradeHandler，同样的道理，如果发来的不是Http2协议，那么这个类就不会生效。这里虽然用的是ChannelHandler，除非自己实现Http2处理器，大部分情况下用的都是Http2ConnectionHandler及其子类，一般情况下使用Http2FrameCodec，它可以将消息处理为最方便的理解的各种http2的帧
  2. HttpObjectAggregator：间接继承了MessageToMessageDecoder，也一种编码器，代码中自定义的MessageHandler继承了HttpObjectAggregator，它用于合并消息，因为对于完整的http消息，netty提供的解码器只能将其分成两部分，即消息头和消息体，传到用户自定义的处理器时，也是分两部分的，而HttpObjectAggregator则可以将消息头和消息体合为一个完整的http消息，其只有一个必填的参数，就是最大消息体内容长度，其他参数可不填
  3. Http2Frame：http2帧，还有子类，Http2StreamFrame（stream帧，http2多路复用的关键），HTTP2包含很多不同的帧，有连接层面的帧，比如Setting帧，GoAway帧，也有很多请求层面的帧，用不同的流（stream）区分，比如Header帧，Data帧，连接层面的帧，相同的帧同时只会有一个，而请求层面的帧，同时可以有多个不同stream的相同的帧，而这也就是http2的多路复用
  4. Http2MultiplexHandler：多路复用处理器，虽然http2已经不受限于一个连接同时只能发起一个请求了，但是多个请求还是需要一个channel来处理，而Http2MultiplexHandler会针对不同stream的http2请求，生成Http2MultiplexHandlerStreamChannel，将他注册到父channel的事件处理器中，由生成的streamchannel处理具体的stream的http2请求，但是因为使用的是父channel的事件处理器，而netty一般的事件处理器都是单线程的，所以多个同父channel的stremachannel也是同步执行的，而且它们使用的handler也是同一个对象，所以在简单的使用场景中，似乎并没有太大的用处，可能比较适用于一些特殊或者复杂的场景

流程介绍

进入管道后，首先会进入CleartextHttp2ServerUpgradeHandler，它继承了ByteToMessageDecoder，属于解码器，所以生效方法是decode

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    int prefaceLength = CONNECTION_PREFACE.readableBytes();
    int bytesRead = Math.min(in.readableBytes(), prefaceLength);

    if (!ByteBufUtil.equals(CONNECTION_PREFACE, CONNECTION_PREFACE.readerIndex(),
                            in, in.readerIndex(), bytesRead)) {
        ctx.pipeline().remove(this);
    } else if (bytesRead == prefaceLength) {
        // Full h2 preface match, removed source codec, using http2 codec to handle
        // following network traffic
        ctx.pipeline()
        .remove(httpServerCodec)
        .remove(httpServerUpgradeHandler);

        ctx.pipeline().addAfter(ctx.name(), null, http2ServerHandler);
        ctx.pipeline().remove(this);

        ctx.fireUserEventTriggered(PriorKnowledgeUpgradeEvent.INSTANCE);
    }
}
1
2
3
4
5
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    ctx.pipeline()
    .addAfter(ctx.name(), null, httpServerUpgradeHandler)
    .addAfter(ctx.name(), null, httpServerCodec);
}

上面一个是编码器方法，一个是添加到管道触发事件，在handlerAdded方法中，会将http升级处理器和http消息编解码器加入通道，这里需要注意addAfter方法的特性，按照调用顺序，http消息编解码器会在http升级处理器之前，在decode方法中，会先判断是否有http2序言，”PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n”，具体参考http2协议，如果有http2序言，证明就是http2请求，无需升级，CleartextHttp2ServerUpgradeHandler会将自己、httpServerCodec和httpServerUpgradeHandler从管道删除，添加一个http2ServerHandler用于处理http2 ByteBuf消息

如果没有http2序言，那么CleartextHttp2ServerUpgradeHandler会将自己从管道中删除，那么当前管道中就只剩httpServerCodec和httpServerUpgradeHandler，按照顺序会先进入httpServerCodec，将ByteBuf消息转为netty中http的消息体，然后httpServerUpgradeHandler接收到http消息体，进行升级处理，它继承了HttpObjectAggregator，也是一种编码器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
protected void decode(ChannelHandlerContext ctx, HttpObject msg, List<Object> out)
throws Exception {

    if (!handlingUpgrade) {
        // Not handling an upgrade request yet. Check if we received a new upgrade request.
        if (msg instanceof HttpRequest) {
            HttpRequest req = (HttpRequest) msg;
            if (req.headers().contains(HttpHeaderNames.UPGRADE) &&
                shouldHandleUpgradeRequest(req)) {
                handlingUpgrade = true;
            } else {
                ReferenceCountUtil.retain(msg);
                ctx.fireChannelRead(msg);
                return;
            }
        } else {
            ReferenceCountUtil.retain(msg);
            ctx.fireChannelRead(msg);
            return;
        }
    }

    FullHttpRequest fullRequest;
    if (msg instanceof FullHttpRequest) {
        fullRequest = (FullHttpRequest) msg;
        ReferenceCountUtil.retain(msg);
        out.add(msg);
    } else {
        // Call the base class to handle the aggregation of the full request.
        super.decode(ctx, msg, out);
        if (out.isEmpty()) {
            // The full request hasn't been created yet, still awaiting more data.
            return;
        }

        // Finished aggregating the full request, get it from the output list.
        assert out.size() == 1;
        handlingUpgrade = false;
        fullRequest = (FullHttpRequest) out.get(0);
    }

    if (upgrade(ctx, fullRequest)) {
        // The upgrade was successful, remove the message from the output list
        // so that it's not propagated to the next handler. This request will
        // be propagated as a user event instead.
        out.clear();
    }

    // The upgrade did not succeed, just allow the full request to propagate to the
    // next handler.
}

这个方法简单来讲就是通过消息头（upgrade）判断是否需要升级，如果不需要升级，则直接退出，此时的消息是httpServerCodec解码出来的http消息，所以使用CleartextHttp2ServerUpgradeHandler是可以处理http消息的。如果需要升级，则使用父类HttpObjectAggregator的合并功能合并httpServerCodec产生的消息头和消息体，合并为FullHttpRequest，然后调用upgrade方法进行真正的升级处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
private boolean upgrade(final ChannelHandlerContext ctx, final FullHttpRequest request) {
    // Select the best protocol based on those requested in the UPGRADE header.
    final List<CharSequence> requestedProtocols = splitHeader(request.headers().get(HttpHeaderNames.UPGRADE));
    final int numRequestedProtocols = requestedProtocols.size();
    UpgradeCodec upgradeCodec = null;
    CharSequence upgradeProtocol = null;
    for (int i = 0; i < numRequestedProtocols; i ++) {
        final CharSequence p = requestedProtocols.get(i);
        final UpgradeCodec c = upgradeCodecFactory.newUpgradeCodec(p);
        if (c != null) {
            upgradeProtocol = p;
            upgradeCodec = c;
            break;
        }
    }

    if (upgradeCodec == null) {
        // None of the requested protocols are supported, don't upgrade.
        return false;
    }

    // Make sure the CONNECTION header is present.
    List<String> connectionHeaderValues = request.headers().getAll(HttpHeaderNames.CONNECTION);

    if (connectionHeaderValues == null || connectionHeaderValues.isEmpty()) {
        return false;
    }

    final StringBuilder concatenatedConnectionValue = new StringBuilder(connectionHeaderValues.size() * 10);
    for (CharSequence connectionHeaderValue : connectionHeaderValues) {
        concatenatedConnectionValue.append(connectionHeaderValue).append(COMMA);
    }
    concatenatedConnectionValue.setLength(concatenatedConnectionValue.length() - 1);

    // Make sure the CONNECTION header contains UPGRADE as well as all protocol-specific headers.
    Collection<CharSequence> requiredHeaders = upgradeCodec.requiredUpgradeHeaders();
    List<CharSequence> values = splitHeader(concatenatedConnectionValue);
    if (!containsContentEqualsIgnoreCase(values, HttpHeaderNames.UPGRADE) ||
        !containsAllContentEqualsIgnoreCase(values, requiredHeaders)) {
        return false;
    }

    // Ensure that all required protocol-specific headers are found in the request.
    for (CharSequence requiredHeader : requiredHeaders) {
        if (!request.headers().contains(requiredHeader)) {
            return false;
        }
    }

    // Prepare and send the upgrade response. Wait for this write to complete before upgrading,
    // since we need the old codec in-place to properly encode the response.
    final FullHttpResponse upgradeResponse = createUpgradeResponse(upgradeProtocol);
    if (!upgradeCodec.prepareUpgradeResponse(ctx, request, upgradeResponse.headers())) {
        return false;
    }

    // Create the user event to be fired once the upgrade completes.
    final UpgradeEvent event = new UpgradeEvent(upgradeProtocol, request);

    // After writing the upgrade response we immediately prepare the
    // pipeline for the next protocol to avoid a race between completion
    // of the write future and receiving data before the pipeline is
    // restructured.
    try {
        final ChannelFuture writeComplete = ctx.writeAndFlush(upgradeResponse);
        // Perform the upgrade to the new protocol.
        sourceCodec.upgradeFrom(ctx);
        upgradeCodec.upgradeTo(ctx, request);

        // Remove this handler from the pipeline.
        ctx.pipeline().remove(HttpServerUpgradeHandler.this);

        // Notify that the upgrade has occurred. Retain the event to offset
        // the release() in the finally block.
        ctx.fireUserEventTriggered(event.retain());

        // Add the listener last to avoid firing upgrade logic after
        // the channel is already closed since the listener may fire
        // immediately if the write failed eagerly.
        writeComplete.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
    } finally {
        // Release the event if the upgrade event wasn't fired.
        event.release();
    }
    return true;
}

这个方法简单来说就是先根据upgrade消息头的协议名生成http2升级编解码器，这一步是我们自己实现的，然后检查所需的消息头是否都具备，http升级http2必须的消息头包括HTTP2-Settings，upgrade，而且connection消息头的内容要包括HTTP2-Settings,upgrade，其中upgrade是http升级必须的消息头，而HTTP2-Settings只是http升级http2必须的消息头，检查完消息头后，调用upgradeCodec(Http2ServerUpgradeCodec类)相关方法进行升级，分别调用了upgradeCodec的prepareUpgradeResponse和upgradeTo，前者是根据HTTP2-Settings消息头生成Http2的Setting消息，后者是添加Http2ConnectionHandler，具体参考Http2ServerUpgradeCodec，然后httpServerUpgradeHandler会将自己从管道中删除，那么到这一步，管道中就只剩一个Http2ConnectionHandler了，所以后续客户端只能发送http2请求，否则只能重新建立连接，到这一步就完成升级了

总结CleartextHttp2ServerUpgradeHandler为什么能同时处理http1，http1升级，http2

  1. 处理http1：是因为在CleartextHttp2ServerUpgradeHandler的decode方法中，如果发现消息不是http2序言，那么删除掉自己，然后交给http编解码器和升级处理器，如果http中没有升级所需消息头，就会被当作普通http处理
  2. 处理http1升级：如果http有升级所需消息头，那么就可以升级
  3. 处理http2：是因为在CleartextHttp2ServerUpgradeHandler的decode方法中，如果发现消息是http2序言，那么就会将http编解码器、升级处理器还有自身都从管道中删除，并且添加http2处理器

netty读取和发送

netty的读取和发送离不开io.netty.channel.AbstractChannelHandlerContext，像代码中看到的writeAndFlush和channelRead方法，都需要经过此类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
private void invokeChannelRead(Object msg) {
    if (invokeHandler()) {
        try {
            // DON'T CHANGE
            // Duplex handlers implements both out/in interfaces causing a scalability issue
            // see https://bugs.openjdk.org/browse/JDK-8180450
            final ChannelHandler handler = handler();
            final DefaultChannelPipeline.HeadContext headContext = pipeline.head;
            if (handler == headContext) {
                headContext.channelRead(this, msg);
            } else if (handler instanceof ChannelDuplexHandler) {
                ((ChannelDuplexHandler) handler).channelRead(this, msg);
            } else {
                ((ChannelInboundHandler) handler).channelRead(this, msg);
            }
        } catch (Throwable t) {
            invokeExceptionCaught(t);
        }
    } else {
        fireChannelRead(msg);
    }
}

AbstractChannelHandlerContext的invokeChannelRead会调用处理器的channelRead方法，将消息提交给处理器处理，这里的channelRead一般是系统默认实现的解码器和自己实现的读取消息代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
private void invokeWrite0(Object msg, ChannelPromise promise) {
    try {
        // DON'T CHANGE
        // Duplex handlers implements both out/in interfaces causing a scalability issue
        // see https://bugs.openjdk.org/browse/JDK-8180450
        final ChannelHandler handler = handler();
        final DefaultChannelPipeline.HeadContext headContext = pipeline.head;
        if (handler == headContext) {
            headContext.write(this, msg, promise);
        } else if (handler instanceof ChannelDuplexHandler) {
            ((ChannelDuplexHandler) handler).write(this, msg, promise);
        } else {
            ((ChannelOutboundHandler) handler).write(this, msg, promise);
        }
    } catch (Throwable t) {
        notifyOutboundHandlerException(t, promise);
    }
}

当我们调用channel或者ctx的write(writeAndFlush)方法，会调用到AbstractChannelHandlerContext的invokeWrite0方法，并且它会调用处理器的write方法，将消息提交给处理器处理，这里的wirte方法一般是系统实现的编码器

另外还有一个比较重要的类，就是io.netty.channel.socket.nio.NioSocketChannel，这个类实在ServerBootstrap配置的，windows系统一般情况就是使用NioSocketChannel，这个类可以来理解为java.nio.channels.SocketChannel的封装类，NioSocketChannel的doWrite方法就是调用SocketChannel的write方法

关于TCP与HTTP

http是响应式的协议，只能一次请求，一次响应，http2变成同时多次请求，多次响应，但是从tcp协议的层面来看，tcp是传输层的协议，它本身可以交互，但并没有请求和响应的说法，tcp天然就支持http2的多路复用，所以http1.1和http2只是设计上的问题，并不存在网络传输技术的更改，而http2的多路复用，最终也需要导代码层面实现

本站总访问量：次本站总访问数：人
本站由 Peng Tong 使用 Stellar 1.29.1 主题创建。
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议，转载请注明出处。