Netty网络框架（二）-阿里云开发者社区

Netty网络框架（一）https://developer.aliyun.com/article/1469519

（二）应用篇

1、HTTP

1) 0.9版本

GET /index.html

服务端只能返回html格式，传输过程只能处理文字。

2) 1.0版本

支持任何格式的内容，包括图像、视频、二进制等等

引入了POST命令、HEAD命令

增加了请求头、状态码，以及权限、缓存等

GET / HTTP/1.0
User-Agent:Mozilla/1.0
Accept: */*

HTTP/1.0 200 OK
Content-Type: text/plain
Content-Encoding: gzip
<html>
  <body> hello world </body>
</html>

a、 Content-Type

服务端通知客户端，当前数据的格式

示例： text/html 、 image/png 、 application/pdf 、 video/mp4

前面是一级类型，后面是二级类型，用斜杠分隔；还可以增加其他参数，如编码格式。

Content-Type: text/plain; charset=utf-8

b、Content-Encoding

表示数据压缩的方式，gzip、compress、deflate

对应客户端的字段为 Accept-Encoding，代表接收哪些压缩方式

c、缺点和问题

每个TCP连接只能发送一个请求，发送完毕连接关闭，使用成本很高，性能较差。

Connection: keep-alive   - 非标准字段

3) 1.1版本

GET / HTTP/1.1
User-Agent: PostmanRuntime/7.24.1
Accept: */*
Cache-Control: no-cache
Postman-Token: 636ce8a6-7eab-451a-8638-4534a3578095
Host: cn.bing.com
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
HTTP/1.1 200 OK
Cache-Control: private, max-age=0
Content-Length: 45786
Content-Type: text/html; charset=utf-8
Content-Encoding: br
Vary: Accept-Encoding
P3P: CP="NON UNI COM NAV STA LOC CURa DEVa PSAa PSDa OUR IND"
Set-Cookie: SRCHD=AF=NOFORM; domain=.bing.com; expires=Wed, 22-Jun-2022 07:03:23 GMT; path=/
Set-Cookie: SRCHUID=V=2&GUID=5C28FF778A2C4A00B32F5408147038BF&dmnchg=1; domain=.bing.com; expires=Wed, 22-Jun-2022 07:03:23 GMT; path=/
Set-Cookie: SRCHUSR=DOB=20200622; domain=.bing.com; expires=Wed, 22-Jun-2022 07:03:23 GMT; path=/
Set-Cookie: _SS=SID=23AE726877396796143D7C99761766C7; domain=.bing.com; path=/
Set-Cookie: _EDGE_S=F=1&SID=23AE726877396796143D7C99761766C7; path=/; httponly; domain=bing.com
Set-Cookie: _EDGE_V=1; path=/; httponly; expires=Sat, 17-Jul-2021 07:03:23 GMT; domain=bing.com
Set-Cookie: MUID=1CB3B15BCBD2637E2C01BFAACAFC6214; samesite=none; path=/; secure; expires=Sat, 17-Jul-2021 07:03:23 GMT; domain=bing.com
Set-Cookie: MUIDB=1CB3B15BCBD2637E2C01BFAACAFC6214; path=/; httponly; expires=Sat, 17-Jul-2021 07:03:23 GMT
X-MSEdge-Ref: Ref A: 71D6F4B3BA9C448EB453341AC182C7BC Ref B: BJ1EDGE0311 Ref C: 2020-06-22T07:03:23Z
Date: Mon, 22 Jun 2020 07:03:23 GMT

a、持久连接，含义为默认不关闭tcp连接，可以被多个请求复用。大多时候，浏览器对同一个域名，允许同时建立6个连接。

b、管道机制，支持客户端发送多个请求，管理请求的顺序的。服务器还是按照接受请求的顺序，返回对应的响应结果。

c、Content-Length, 用来区分数据包的重要字段

d、支持PUT、DELETE、PATCH等命令

缺点和问题

当部分请求耗时较长时，仍会阻塞后续请求的处理速度，这种现象叫做“队头阻塞”/"线头阻塞"。

4) 2.0版本

解决队头阻塞的问题，使用的是多路复用的方式。

说了这么多上代码来操作一下吧！！

我们的编写思路是这样的

编写初始化服务端
编写自定义初始化器和自定义处理器
启动postman 查看我们设置的 http 的响应结果

我们这里有三个类

HttpServer 初始化服务端
MyHttpHandler 自定义处理器
MyHttpInitializer 自定义初始化

首先是 server

我们需要在初始化服务端的时候设置主从线程模型（Netty中常用）
设置启动参数和阻塞队列的长度等设置
设置初始化

public class HttpServer {
    public static void main(String[] args) {
        //可以自定义线程的数量
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        // 默认创建的线程数量 = CPU 处理器数量 *2
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .handler(new LoggingHandler())
                //当前连接被阻塞的时候，BACKLOG代表的事 阻塞队列的长度
                .option(ChannelOption.SO_BACKLOG, 128)
                //设置连接保持为活动状态
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childHandler(new MyHttpInitializer());
        try {
        //设置为异步启动 异步 关闭
            ChannelFuture future = serverBootstrap.bind(9988).sync();
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
          //netty的优雅关闭 指 等一切执行完毕之后 慢慢的关闭
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

编写初始化

继承ChannelInitializer泛型为Channel，用来进行设置出站解码器和入站编码器
使用 codec netty封装好的解码器，这样我们就不用每次定义解码和编码

public class MyHttpInitializer extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        //先对应请求解码，后对响应解码
        //pipeline.addLast("decoder", new HttpRequestDecoder());
        //pipeline.addLast("encoder", new HttpRequestEncoder());
        //当然每次我们都要解码编码很麻烦，netty也有为我们提供对应的解决方案，建议直接使用这个 不会出错
        pipeline.addLast("codec", new HttpServerCodec());
        //压缩数据
        pipeline.addLast("compressor", new HttpContentCompressor());
        //聚合完整的信息 参数代表可以处理的最大值 此时的是 512 kb
        pipeline.addLast("aggregator", new HttpObjectAggregator(512 * 1024));
        pipeline.addLast(new MyHttpHandler());
    }
}

有了初始化，我们还需要一个做事的那就是处理器 Handler

netty帮我们封装了返回完整http响应的类 DefaultFullHttpResponse
我们只需要在读取的时候设置协议，状态码和响应信息，
配置响应头的类型和长度就可以完成对请求的响应

/*
 * 泛型需要设置为 FullHttpRequest
 * 筛选 message 为此类型的消息才处理
 * */
public class MyHttpHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest fullHttpRequest) throws Exception {
        //DefaultFullHttpResponse 是一个默认的完整的http响应
        DefaultFullHttpResponse response = new DefaultFullHttpResponse(
                HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
                Unpooled.wrappedBuffer("hello http netty demo".getBytes())
        );
        //    我们还需要设置响应头
        // 设置请求 响应头字段 可以使用 HttpHeaderNmaes
        // 设置字段时 可以使用
        HttpHeaders headers = response.headers();
        headers.add(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN + ";charset=UTF-8");
        headers.add(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
        ctx.write(response);
    }
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }
}

启动结果

postman 差看到的请求响应参数
可以看到我们设置的 http1.1 协议
类型 text/plain 长度 47
响应给客户端的内容 hello http netty demo

小结

了解 http 各个版本的解决了什么问题，优缺点，优劣性
手动编写一个服务端的响应，用postman 查看响应头我们设置的内容
体验到 netty强大的封装带给我们的便利性

2、WebSocket

websocket是由浏览器发起的

协议标识符 http://127.0.0.1:8080 ws://127.0.0.1:7777

GET ws://127.0.0.1:7777 HTTP/1.1
Host: 127.0.0.1
Upgrade: websocket    # 升级为ws   
Connection: Upgrade   # 此链接需要升级
Sec-WebSocket-key: client-random-string ...  # 标识加密相关信息

HTTP/1.1 101
Upgrade: websocket
Connection: Upgrade

响应码 101 代表本次协议需要更改为websocket

连接建立后，支持文本信息及二进制信息。

Websocket实现的原理：
通过http协议进行连接的建立（握手和回答），建立连接后不再使用http，而tcp自身是支持双向通信的，所以能达到“全双工”的效果。

通信使用的单位叫帧 frame
客户端：发送时将消息切割成多个帧
服务端：接收时，将关联的帧重新组装

【客户端】

var ws = new WebSocket("ws://127.0.0.1:7777/hello");
ws.onopen = function(ev){
     ws.send("hello"); //建立连接后发送数据
}

设计一个样式
左右两个各有一个文本框，中间放一个发送按钮。
左侧文本框用来发送数据，右侧文本框用来显示数据。

Websocket 应用demo

服务端代码

public class WebSocketServer {
    public static void main(String[] args) {
        //可以自定义线程的数量
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        // 默认创建的线程数量 = CPU 处理器数量 *2
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .handler(new LoggingHandler())
                //当前连接被阻塞的时候，BACKLOG代表的事 阻塞队列的长度
                .option(ChannelOption.SO_BACKLOG, 128)
                //设置连接保持为活动状态
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childHandler(new WebSocketInitialzer());
        try {
            ChannelFuture future = serverBootstrap.bind(7777).sync();
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服务端初始化器

public class WebSocketInitialzer extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        //增加编解码器 的另一种方式
        pipeline.addLast(new HttpServerCodec());
        //    块方式写的处理器 适合处理大数据
        pipeline.addLast(new ChunkedWriteHandler());
        //聚合
        pipeline.addLast(new HttpObjectAggregator(512 * 1024));
        /*
         * 这个时候 我们需要声明我们使用的是 websocket 协议
         * netty为websocket也准备了对应处理器  设置的是访问路径
         * 这个时候我们只需要访问 ws://127.0.0.1:7777/hello 就可以了
         * 这个handler是将http协议升级为websocket  并且使用 101 作为响应码
         * */
        pipeline.addLast(new WebSocketServerProtocolHandler("/hello"));
        pipeline.addLast(new WebSocketHandler());
    }
}

服务端处理器

通信使用的单位叫帧 frame
客户端：发送时将消息切割成多个帧
服务端：接收时，将关联的帧重新组装

/*
 * 泛型 代表的是处理数据的单位
 * TextWebSocketFrame : 文本信息帧
 * */
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame) throws Exception {
        //可以直接调用text 拿到文本信息帧中的信息
        System.out.println("msg:" + textWebSocketFrame.text());
        Channel channel = ctx.channel();
        //我们可以使用新建一个对象 将服务端需要返回的信息放入其中 返回即可
        TextWebSocketFrame resp = new TextWebSocketFrame("hello client from websocket server");
        channel.writeAndFlush(resp);
    }
}

websocket前端编写

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<script>
    var socket;
    //    判断当前浏览器是否支持websocket
    if (!window.WebSocket) {
        alert("不支持websocket")
    } else {
        <!-- 创建websocket 连接对象-->
        socket = new WebSocket("ws://127.0.0.1:7777/hello");
        //设置开始连接的方法
        socket.onopen = function (ev) {
            var tmp = document.getElementById("respText");
            tmp.value = "连接已经开启";
        }
        //设置关闭连接的方法
        socket.onclose = function (ev) {
            var tmp = document.getElementById("respText");
            tmp.value = tmp.value + "\n" + "连接已经关闭";
        }
        //设置接收数据的方法
        socket.onmessage = function (ev) {
            var tmp = document.getElementById("respText");
            tmp.value = tmp.value + "\n" + ev.data;
        }
    }
    function send(message) {
        if (!window.socket) {
            return
        }
        /*
        * 判断socket的状态
        * connecting 正在连接 closing 正在关闭
        * closed 已经关闭或打开连接失败
        * open 连接可以 已经正常通信
        * */
        if (socket.readyState == WebSocket.OPEN) {
            socket.send(message);
        } else {
            alert("连接未开启");
        }
    }
</script>
<!--防止表单自动提交-->
<form onsubmit="return false">
    <textarea name="message" style="height: 400px;width: 400px"></textarea>
    <input type="button" value="发送" onclick="send(this.form.message.value)">
    <textarea id="respText" style="height: 400px;width: 400px"></textarea>
</form>
</body>
</html>

【客户端】

var ws = new WebSocket("ws://127.0.0.1:7777/hello");
ws.onopen = function(ev){
     ws.send("hello"); //建立连接后发送数据
}

设计一个样式
左右两个各有一个文本框，中间放一个发送按钮。
左侧文本框用来发送数据，右侧文本框用来显示数据。

演示效果

启动服务发送消息

小结

websocket 一般用于做可复用连接，http一般做短链接
websocket解决了http连接只能客户端发起的

（三）原理篇

1、ByteBuf

NIO中ByteBuffer的缺点：

A 长度固定，无法动态的扩容和缩容，缺乏灵活性
B 使用一个position记录读写的索引位置，在读写模式切换时需手动调用flip方法，增加了使用的复杂度。
C 功能有限，使用过程中往往需要自行封装

1）分类

按照内存的位置，分为堆内存缓冲区 heap buffer、直接内存缓冲区direct buffer、复合内存缓冲区composite buffer。

A heap buffer

将数据存储到JVM的堆空间中，实际使用字节数组byte[]来存放。
优点：数据可以快速的创建和释放，并且能够直接访问内部数组
缺点：在读写数据时，需要将数据复制到直接缓冲区再进行网络传输。

B direct buffer

不在堆中，而是使用了操作系统的本地内存。
优点：在使用Socket进行数据传输过程中，减少一次拷贝，性能更高。
缺点：释放和分配的空间更昂贵，使用时需要更谨慎。

C composite buffer

将两个或多个不同内存的缓冲区合并
优点：可以统一进行操作

应用场景：在通信线程使用缓冲区时，往往使用direct buffer，而业务消息使用缓冲区时，往往使用heap buffer，在解决http包，请求头+请求体特性不同而选择不同位置存储时，可以将两者拼接使用

D 池化的概念

对于内存空间分配和释放的复杂度和效率，netty通过内存池的方式来解决。
内存池，可以循环利用ByteBuf，提高使用率。但是管理和维护较复杂。

Unpooled正是非池化缓冲区的工具类。

主要区别在于，池化的内存由netty管理，非池化的内存由GC回收。

E 回收方式

回收方式为引用计数，具体规则为，通过记录被引用的次数，判断当前对象是否还会被使用。
当对象被调用时，引用计为+1，当对象被释放时，引用计为-1，当引用次数为0时，对象可以回收。

弊端：可能引发内存泄漏。
当对象不可达，JVM会通过GC回收掉，但此时引用计数可能不为0，对象无法归还内存池，会导致内存泄漏。netty只能通过对内存缓冲区进行采样，来检查。

2）工作原理

和ByteBuffer不同在于，增加了一个指针，通过两个指针记录读模式和写模式时的索引位置，读指针叫做readerIndex，写指针叫做writerIndex。

A 读写分离

当执行clear()方法时，索引位置清空回初始位置，但数据保持不变。

mark和reset方法在ByteBuf中同样适用，如markReaderIndex和resetReaderIndex。

B 深浅拷贝

浅拷贝，拷贝的是对对象的引用，并没有创建新对象，新对象和原对象之间互相影响。
深拷贝，拷贝的是整个对象，和原对象之间完全独立。

duplicate和slice方法，达成全部浅拷贝和部分浅拷贝。
copy，部分深拷贝，部分代表的是可读空间。

3）扩容机制

A ByteBuffer的存储

ByteBuffer在put数据时，会校验剩余空间是否不足，如果不足，会抛出异常。

ByteBuffer buffer = ByteBuffer.allocate(8);
buffer.put("yu".getBytes());
----------------------------------------------------
    public final ByteBuffer put(byte[] src) {
        return put(src, 0, src.length);
    }
    
    // 额外接收偏移量（存储数据的起始位置）  和数据长度
    public ByteBuffer put(byte[] src, int offset, int length) {
        // 校验参数的有效性
        checkBounds(offset, length, src.length);
        // 如果要存储数据的长度 > 剩余可用空间  抛出buffer越界的异常
        if (length > remaining())
            throw new BufferOverflowException();
        // 如果剩余空间足够  计算存储的结束位置 = 偏移量 + 数据长度    
        int end = offset + length;
        for (int i = offset; i < end; i++)
            this.put(src[i]);
        return this;
    }

如果要手动对ByteBuffer扩容，可以在put之前，先校验剩余空间是否足够，如果不足够，创建一个新的ByteBuffer，新的容量确保足够，旧的buffer数据拷贝到新的buffer中，然后继续存储数据。

B ByteBuf的存储和扩容

当写数据时，先判断是否需要扩容，如果当前空间较小（<4M），以64作为基数倍增（10 -> 64 -> 128 -> 256）, 如果当前空间较大（>4M）, 每次扩容都增加4M，这种方式叫做"步进式"。

查看源码，以AbstractByteBuf子类为依据查看，最重要的子类之一，ByteBuf的公共属性和功能都在此中实现。
ByteBuf buf = Unpooled.buffer(10);
System.out.println("capacity: " + buf.capacity());
for (int i = 0; i < 11; i++) {
    buf.writeByte(i);
}
----------------------------------------------------   
[ByteBuf类]
public abstract ByteBuf writeByte(int value);
按住Ctrl+Alt快捷键
[AbstractByteBuf子类]
---------------------------------------------------- 
    @Override
    public ByteBuf writeByte(int value) {
        // 确保可写空间足够
        ensureWritable0(1);
        // 写入数据
        _setByte(writerIndex++, value);
        return this;
    }
    
    // 参数为 最小写入数据的大小
    final void ensureWritable0(int minWritableBytes) {
        final int writerIndex = writerIndex();
        // 目标容量 = 当前写操作索引 + 最小写入数据大小
        final int targetCapacity = writerIndex + minWritableBytes;
        // 容量足够  不需扩容
        if (targetCapacity <= capacity()) {
            ensureAccessible();
            return;
        }
        // 容量不足时 如果目标容量 超出最大容量  抛出异常
        if (checkBounds && targetCapacity > maxCapacity) {
            ensureAccessible();
            throw new IndexOutOfBoundsException(String.format(
                    "writerIndex(%d) + minWritableBytes(%d) exceeds maxCapacity(%d): %s",
                    writerIndex, minWritableBytes, maxCapacity, this));
        }
    
    // 扩容逻辑
        // 获取可写空间大小
        final int fastWritable = maxFastWritableBytes();
        // 如果 可写空间 >= 所需空间   新的容量=写操作索引+可写空间大小 
        // 如果 可写空间 < 所需空间   计算要扩容的新容量大小 calculateNewCapacity方法
        int newCapacity = fastWritable >= minWritableBytes ? writerIndex + fastWritable
                : alloc().calculateNewCapacity(targetCapacity, maxCapacity);
        // Adjust to the new capacity.
        
        // 计算完成后 生成新的ByteBuffer
        capacity(newCapacity);
    }
    
    // 获取可写空间大小
    public int maxFastWritableBytes() {
        return writableBytes();
    }
    
[AbstractByteBufAllocator子类]
---------------------------------------------------- 
    // 计算要扩容的新容量大小
    @Override
    public int calculateNewCapacity(int minNewCapacity, int maxCapacity) {
        // 校验参数有效性
        checkPositiveOrZero(minNewCapacity, "minNewCapacity");
        if (minNewCapacity > maxCapacity) {
            throw new IllegalArgumentException(String.format(
                    "minNewCapacity: %d (expected: not greater than maxCapacity(%d)",
                    minNewCapacity, maxCapacity));
        }
        // 扩容方式的分界点 以4M大小为界
        final int threshold = CALCULATE_THRESHOLD; // 4 MiB page
        if (minNewCapacity == threshold) {
            return threshold;
        }
        // If over threshold, do not double but just increase by threshold.、
        // 如果所需容量大于4M  按照步进的方式扩容 
        //   举例： 比如 minNewCapacity = 5M  
        if (minNewCapacity > threshold) {
            // newCapacity = 5 / 4 * 4 = 4M  确保是4的倍数
            int newCapacity = minNewCapacity / threshold * threshold;
            if (newCapacity > maxCapacity - threshold) {
                newCapacity = maxCapacity;
            } else {
                // newCapacity = 4 + 4 = 8M;
                newCapacity += threshold;
            }
            return newCapacity;
        }
        // Not over threshold. Double up to 4 MiB, starting from 64.
        // 如果所需容量大于4M  按照64的倍数扩容  找到最接近所需容量的64的倍数
        int newCapacity = 64;
        while (newCapacity < minNewCapacity) {
            newCapacity <<= 1;
        }
        
        // 保障在最大可接受容量范围内
        return Math.min(newCapacity, maxCapacity);
    }

4）优势

A 池化的方式提高内存使用率

B 提出了复合型缓冲区的整合方案

C 增加了索引，使读写分离，使用更便捷

D 解决了ByteBuffer长度固定的问题，增加了扩容机制

E 用引用计数的方式进行对象回收

2、Channel

1）Channel

channel是通讯的载体，对应通讯的一端，在BIO中对应Socket，NIO中对应SocketChannel，Netty中对应NioSocketChannel，ServerSocket同理。
channelhandler是通道的处理器，一个channel往往有多个handler
channelpipeline是handler的容器，装载并管理handler的顺序（本质是双向链表）

如图，channel创建时，会对应创建一个channelpipeline，pipeline首先会记录一个头部的处理器handler，当pipeline进行分发时，先分发给头部，然后依次执行，执行handler全部执行完成。

同时，channel创建后，会注册进EventLoop之中，EventLoop会监听事件的发生。不同的事件调用handler不同的处理方法，让流程运转起来。

channel生命周期，对应四种状态，分别为：
A) ChannelUnregistered 已创建但还未被注册到监听器中
B) ChannelRegistered 已注册到监听器EventLoop中
C) ChannelActive 连接完成处于活跃状态，此时可以接收和发送数据
D) ChannelInactive 非活跃状态，代表连接未建立或者已断开

channelhandler生命周期，对应三种状态，分别为：
A) handlerAdded 把handler添加到pipeline之中
B) handlerRemoved 从pipeline中移除
C) exceptionCaught 在处理过程中有错误产生

创建channel源码分析

以服务端启动为例
ChannelFuture future = serverBootstrap.bind(8888).sync();
参数设置
serverBootstrap.channel(NioServerSocketChannel.class)
【AbstractBootstrap】 启动对象的父类
------------------------------------------------------------------
    public ChannelFuture bind(int inetPort) {
        return bind(new InetSocketAddress(inetPort));
    }
    
    public ChannelFuture bind(SocketAddress localAddress) {
        validate();
        return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress"));
    }
    
    private ChannelFuture doBind(final SocketAddress localAddress) {
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
        .......
    }
    
    
    final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            channel = channelFactory.newChannel();
            init(channel);
        } 
        .......
    }
   
   
 【ReflectiveChannelFactory】  工厂实现类
 ------------------------------------------------------------------   
    public T newChannel() {
        try {
            return constructor.newInstance();
        } catch (Throwable t) {
            throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
        }
    }

在启动对象调用bind()或connect()方法时，会创建channel
本质上通过反射，使用工厂的反射实现类创建对应的实例，此时实例对象的类型是通过channel参数来设置的

2）ChannelHandler

类层次关系图

入站和出站：
从服务端的角度，数据从客户端发送到服务端，称之为入站，当数据处理完成返回给客户端，称之为出站。是相对的概念。

从客户端的角度，数据从服务端发送给客户端，称之为入站，当数据返回给服务端，称之为出站。

不论是入站还是出站，handler从一端开始，到另一端结束，以责任链的模式依次执行。

责任链模式——"击鼓传花"，当请求被不同的接收者处理时，每个接收者都包含对下一个接收者的引用，一个接收者处理完成后，将依次向下传递。

适配器模式——出国时要使用的电源转换器（美国/日本110V 中国220V电压），作为两个不兼容的接口之间的桥梁，将类的接口转换为需要的另外一种接口。

ChannelDuplexHandler是除了入站和出站handler之外的，另一个常用子类。
它同时实现了ChannelInboundHandler和ChannelOutboundHandler接口，如果需要既处理入站事件又处理出站事件，可以继承此类。

serverBootstrap.handler(new LoggingHandler(LogLevel.INFO))
 
 ------------------------------------------------------------------
 public class LoggingHandler extends ChannelDuplexHandler{}
 
 ------------------------------------------------------------------
 public class ChannelDuplexHandler extends ChannelInboundHandlerAdapter implements ChannelOutboundHandler {}
 
  ------------------------------------------------------------------
  public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler {}

ChannelHandlerAdapter
提供了额外的isSharable()方法，用来判断handler是否可以被共享到多个pipeline之中。默认情况不共享，如果需要共享，在继承了适配器的handler上，增加注解@Sharable

@Sharable
public class LoggingHandler extends ChannelDuplexHandler {}

ChannelInboundHandler
最重要的方法是channelRead()，在使用时，需要显式的释放ByteBuf相关的内存。使用ReferenceCountUtil是引用计数的工具类。

@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // netty中的缓冲区  叫做ByteBuf  -- 对ByteBuffer的封装
        ByteBuf buf = (ByteBuf) msg;
        // 释放ByteBuf内存
        ReferenceCountUtil.release(msg);
    }

为了减少对资源内存的管理，使用SimpleChannelInboundHandler，使用其channelRead0()方法，可以自动释放资源，使用更便利。

SimpleChannelInboundHandler源码
------------------------------------------------
@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        boolean release = true;
        try {
            if (acceptInboundMessage(msg)) {
                @SuppressWarnings("unchecked")
                I imsg = (I) msg;
                channelRead0(ctx, imsg);
            } else {
                release = false;
                ctx.fireChannelRead(msg);
            }
        } finally {
            if (autoRelease && release) {
                ReferenceCountUtil.release(msg);
            }
        }
    }

3）ChannelPipeline

pipeline中维护入站和出站链路，两条链路的执行顺序。

handler只负责处理自身的业务逻辑，对通道而言，它是无状态的。通道的信息会保存到handlerContext处理器上下文中，它是连接pipeline和handler之间的中间角色。

pipeline管理的是由handlerContext包裹的handler，也就是说，当添加handler时，先将其转为handlerContext，然后添加到pipeline的双向链表中。头结点叫做HeadContext，尾节点叫做TailContext。

ch.pipeline().addLast(new NettyServerHandler());
 
 [DefaultChannelPipeline]
 ----------------------------------------------------------------
    public final ChannelPipeline addLast(ChannelHandler... handlers) {
        return addLast(null, handlers);
    }
    
    public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
        ObjectUtil.checkNotNull(handlers, "handlers");
        for (ChannelHandler h: handlers) {
            if (h == null) {
                break;
            }
            addLast(executor, null, h);
        }
        return this;
    }
    
    
    // 关键逻辑
    public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
        final AbstractChannelHandlerContext newCtx;
        synchronized (this) {
            // 检查当前handler是否支持共享，如果不支持，又被添加到其他pipeline中，会报错
            checkMultiplicity(handler);
        // 将handler封装为context
            newCtx = newContext(group, filterName(name, handler), handler);
            // 将context添加到链表尾部
            addLast0(newCtx);
            // If the registered is false it means that the channel was not registered on an eventLoop yet.
            // In this case we add the context to the pipeline and add a task that will call
            // ChannelHandler.handlerAdded(...) once the channel is registered.
            // 判断当前通道的注册状态，如果是未注册，执行此逻辑
            if (!registered) {
                // 添加一个任务，当通道被注册后，能够回调handlerAdded方法
                newCtx.setAddPending();
                callHandlerCallbackLater(newCtx, true);
                return this;
            }
            // 如果已被注册  执行调用handlerAdded方法
            EventExecutor executor = newCtx.executor();
            if (!executor.inEventLoop()) {
                callHandlerAddedInEventLoop(newCtx, executor);
                return this;
            }
        }
        callHandlerAdded0(newCtx);
        return this;
    }
    
    
    
    private static void checkMultiplicity(ChannelHandler handler) {
        if (handler instanceof ChannelHandlerAdapter) {
            ChannelHandlerAdapter h = (ChannelHandlerAdapter) handler;
            if (!h.isSharable() && h.added) {
                throw new ChannelPipelineException(
                        h.getClass().getName() +
                        " is not a @Sharable handler, so can't be added or removed multiple times.");
            }
            h.added = true;
        }
    }
    
    
    private AbstractChannelHandlerContext newContext(EventExecutorGroup group, String name, ChannelHandler handler) {
        return new DefaultChannelHandlerContext(this, childExecutor(group), name, handler);
    }
    
    // 尾节点会提前声明并创建
    final AbstractChannelHandlerContext tail;
    
    //  prev -> tail   在其中插入newctx
    //  prev -> newctx -> tail   放到倒数第二个位置中  tail节点是保持不变的
    //  依次更改 新节点的前后指针   以及prev节点的后置指针和tail节点的前置指针
    private void addLast0(AbstractChannelHandlerContext newCtx) {
        AbstractChannelHandlerContext prev = tail.prev;
        newCtx.prev = prev;
        newCtx.next = tail;
        prev.next = newCtx;
        tail.prev = newCtx;
    }
    
    // 构造器中已经提前创建了头尾节点
    protected DefaultChannelPipeline(Channel channel) {
        this.channel = ObjectUtil.checkNotNull(channel, "channel");
        succeededFuture = new SucceededChannelFuture(channel, null);
        voidPromise =  new VoidChannelPromise(channel, true);
        tail = new TailContext(this);
        head = new HeadContext(this);
        head.next = tail;
        tail.prev = head;
    }

使用pipeline模式的优点：
A) 解耦，让处理器逻辑独立，可以被多个channel共享
B) channel相关信息，交给context维护
C) 具有极大的灵活性，使用处理器可以方便的添加或删除，或者更改它们的顺序

Netty网络框架（三）https://developer.aliyun.com/article/1469522

Netty网络框架（二）

（二）应用篇

1、HTTP

1) 0.9版本

2) 1.0版本

3) 1.1版本

4) 2.0版本

说了这么多上代码来操作一下吧！！

小结

2、WebSocket

Websocket 应用demo

演示效果

小结

（三）原理篇

1、ByteBuf

1）分类

A heap buffer

B direct buffer

C composite buffer

D 池化的概念

E 回收方式

2）工作原理

A 读写分离

B 深浅拷贝

3）扩容机制

A ByteBuffer的存储

B ByteBuf的存储和扩容

4）优势

2、Channel

1）Channel

2）ChannelHandler

3）ChannelPipeline

热门文章

最新文章

相关电子书

Netty网络框架（二）

（二） 应用篇

1、HTTP

1) 0.9版本

2) 1.0版本

3) 1.1版本

4) 2.0版本

说了这么多 上代码来操作一下吧！！

小结

2、WebSocket

Websocket 应用demo

演示效果

小结

（三）原理篇

1、ByteBuf

1）分类

A heap buffer

B direct buffer

C composite buffer

D 池化的概念

E 回收方式

2）工作原理

A 读写分离

B 深浅拷贝

3）扩容机制

A ByteBuffer的存储

B ByteBuf的存储和扩容

4）优势

2、Channel

1）Channel

2）ChannelHandler

3）ChannelPipeline

热门文章

最新文章

相关电子书

（二）应用篇

说了这么多上代码来操作一下吧！！