基本组件：Handler、Pipeline 与 ByteBuf

# 1. Handler 与 Pipeline

Pipeline 好比一条流水线，Handler 好比流水线上一道一道的工序，上面流动的就是你要分批处理的数据

ChannelHandler 用来处理 Channel 上的各种事件，分为入站、出站两种。所有 ChannelHandler 被连成一串，就是 Pipeline

入站处理器通常是 ChannelInboundHandlerAdapter 的子类，主要用来读取客户端数据，写回结果
出站处理器通常是 ChannelOutboundHandlerAdapter 的子类，主要对写回结果进行加工

public class PipeLineServer {
    public static void main(String[] args) {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        // 在socketChannel的pipeline中添加handler
                        // pipeline中handler是带有head与tail节点的双向链表，的实际结构为
                        // head <-> handler1 <-> ... <-> handler4 <->tail
                        // Inbound主要处理入站操作，一般为读操作，发生入站操作时会触发Inbound方法
                        // 入站时，handler是从head向后调用的
                        socketChannel.pipeline().addLast("handler1" ,new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                System.out.println(Thread.currentThread().getName() + " Inbound handler 1");
                                // 父类该方法内部会调用fireChannelRead
                                // 将数据传递给下一个handler
                                super.channelRead(ctx, msg);
                            }
                        });
                        socketChannel.pipeline().addLast("handler2", new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                System.out.println(Thread.currentThread().getName() + " Inbound handler 2");
                                // 执行write操作，使得Outbound的方法能够得到调用
          socketChannel.writeAndFlush(ctx.alloc().buffer().writeBytes("Server...".getBytes(StandardCharsets.UTF_8)));
                                super.channelRead(ctx, msg);
                            }
                        });
                        // Outbound主要处理出站操作，一般为写操作，发生出站操作时会触发Outbound方法
                        // 出站时，handler的调用是从tail向前调用的
                        socketChannel.pipeline().addLast("handler3" ,new ChannelOutboundHandlerAdapter(){
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                System.out.println(Thread.currentThread().getName() + " Outbound handler 1");
                                super.write(ctx, msg, promise);
                            }
                        });
                        socketChannel.pipeline().addLast("handler4" ,new ChannelOutboundHandlerAdapter(){
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                System.out.println(Thread.currentThread().getName() + " Outbound handler 2");
                                super.write(ctx, msg, promise);
                            }
                        });
                    }
                })
                .bind(8080);
    }
}

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

运行结果如下：

nioEventLoopGroup-2-2 Inbound handler 1
nioEventLoopGroup-2-2 Inbound handler 2
nioEventLoopGroup-2-2 Outbound handler 2
nioEventLoopGroup-2-2 Outbound handler 1

1
2
3
4

通过 channel.pipeline().addLast(name, handler) 可以在 pipeline 中添加 handler，记得给 handler 取名字，这样可以调用 pipeline 的 addAfter、addBefore 等方法更灵活地向 pipeline 中添加 handler。

handler 需要放入通道的 pipeline 中，才能根据放入顺序来使用 handler：

pipeline 是结构是一个带有 head 与 tail 指针的双向链表，其中的每个节点为 handler
- 要通过 ctx.fireChannelRead(msg) 等方法，将当前 handler 的处理结果传递给下一个 handler
- ctx.fireChannelRead(msg) 已经被 super.channelRead() 或 super.write() 所封装
- 如果不直接或间接调用这个 ctx.fireChannelRead(msg)，那调用链就会断掉。
当有入站（Inbound）操作时，会从 head 开始向后调用 handler，直到 handler 不是处理 Inbound 操作为止
当有出站（Outbound）操作时，会从 tail 开始向前调用 handler，直到 handler 不是处理 Outbound 操作为止

结构如下：

调用顺序如下：

# 1.1 Inbound Handler

以示例代码的 handler 1、handler 2 为例，看一下 Inbound Handler。

入站处理器继承自 ChannelInboundHandlerAdapter，需要重写其中的 channelRead 来处理读取的消息 Object msg。第一个 handler 得到的 msg 就是原始 IO 读取到的数据，后面 handler 得到的 msg 是前面 handler 处理后传给他的 msg。每一个 handler 在处理完 msg 后，调用 super.channelRead(ctx, msg); 来交由下一个 inbound handler。

# 1.2 Outbound Handler

以示例代码的 handler 3、handler 4 为例，看一下 Inbound Handler。

出站处理器继承自 ChannelOutboundHandlerAdapter，需要重写其中的 write 来处理对外写的消息，在 handler 处理完之后，需要调用 super.write(ctx, msg, promise) 来交由上一个 outbound handler。

小细节：socketChannel.writeAndFlush VS ctx.writeAndFlush()

socketChannel.writeAndFlush()：当handler中调用该方法进行写操作时，会触发Outbound操作，此时是从tail向前寻找OutboundHandler

ctx.writeAndFlush()：当handler中调用该方法进行写操作时，会触发Outbound操作，此时是从当前handler向前寻找OutboundHandler

super.write() 是对 socketChannel.writeAndFlush()，这才是合理的，所以应该在 handler 中调用这个。

# 1.3 EmbeddedChannel

EmbeddedChannel 是一个用于调试的工具类，可以用于测试各个 handler。通过其构造函数按顺序传入需要测试 handler，然后调用对应的 Inbound 和 Outbound 方法即可。

public class TestEmbeddedChannel {
    public static void main(String[] args) {
        ChannelInboundHandlerAdapter h1 = new ChannelInboundHandlerAdapter() {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                System.out.println("1");
                super.channelRead(ctx, msg);
            }
        };

        ChannelInboundHandlerAdapter h2 = new ChannelInboundHandlerAdapter() {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                System.out.println("2");
                super.channelRead(ctx, msg);
            }
        };

        ChannelOutboundHandlerAdapter h3 = new ChannelOutboundHandlerAdapter() {
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                System.out.println("3");
                super.write(ctx, msg, promise);
            }
        };

        ChannelOutboundHandlerAdapter h4 = new ChannelOutboundHandlerAdapter() {
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                System.out.println("4");
                super.write(ctx, msg, promise);
            }
        };

        // 用于测试Handler的Channel
        EmbeddedChannel channel = new EmbeddedChannel(h1, h2, h3, h4);
        
        // 模拟入站操作，执行 Inbound 操作 
        channel.writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hello".getBytes(StandardCharsets.UTF_8)));
        // 模拟出站操作，执行 Outbound 操作
        channel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hello".getBytes(StandardCharsets.UTF_8)));
    }
}

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

# 2. ByteBuf

ByteBuf 是对 NIO 中的 ByteBuffer 的增强。

为了方便，先写一个调试工具方法，该方法可以帮助我们更为详细地查看 ByteBuf 中的内容：

private static void log(ByteBuf buffer) {
    int length = buffer.readableBytes();
    int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
    StringBuilder buf = new StringBuilder(rows * 80 * 2)
        .append("read index:").append(buffer.readerIndex())
        .append(" write index:").append(buffer.writerIndex())
        .append(" capacity:").append(buffer.capacity())
        .append(NEWLINE);
    appendPrettyHexDump(buf, buffer);
    System.out.println(buf.toString());
}

1
2
3
4
5
6
7
8
9
10
11

# 2.1 ByteBuf 的创建

使用 ByteBufAllocator.DEFAULT.buffer() 可以创建一个默认的 ByteBuf

public class ByteBufStudy {
    public static void main(String[] args) {
        // 创建ByteBuf
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16);
        ByteBufUtil.log(buffer);

        // 向buffer中写入数据
        StringBuilder sb = new StringBuilder();
        for(int i = 0; i < 20; i++) {
            sb.append("a");
        }
        buffer.writeBytes(sb.toString().getBytes(StandardCharsets.UTF_8));

        // 查看写入结果
        ByteBufUtil.log(buffer);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

ByteBuf 通过 ByteBufAllocator 选择 allocator 并调用对应的 buffer() 方法来创建的，默认使用直接内存作为 ByteBuf，容量为 256 个字节，可以指定初始容量的大小。

当 ByteBuf 的容量无法容纳所有数据时，ByteBuf 会进行自动扩容操作。

如果在 handler 中创建 ByteBuf，建议使用 ChannelHandlerContext ctx.alloc().buffer() 来创建。

# 2.2 直接内存 vs 堆内存

之前说过，堆内存的分配效率比较高，但读写效率比较低，而直接内存分配效率比较低，但读写效率比较高（因为 zero-copy）。

创建池化基于堆内存的 ByteBuf：ByteBufAllocator.DEFAULT.heapBuffer(16);
创建池化基于直接内存的 ByteBuf：yteBufAllocator.DEFAULT.directBuffer(16); 或 ByteBufAllocator.DEFAULT.buffer(16);

在选择时，需要考虑到：

直接内存创建和销毁的代价昂贵，但读写性能高（少一次内存复制），适合配合池化功能一起用
直接内存对 GC 压力小，因为这部分内存不受 JVM 垃圾回收的管理，但也要注意及时主动释放

# 2.3 池化 vs 非池化

这里的池化是指对分配的内存的池化，避免每次使用内存都是申请内存再释放内存。池化的最大意义在于可以重用 ByteBuf，优点有：

没有池化，则每次都得创建新的 ByteBuf 实例，这个操作对直接内存代价昂贵，就算是堆内存，也会增加 GC 压力
有了池化，则可以重用池中 ByteBuf 实例，并且采用了与 jemalloc 类似的内存分配算法提升分配效率
高并发时，池化功能更节约内存，减少内存溢出的可能

池化功能是否开启，可以通过下面的系统环境变量来设置：

-Dio.netty.allocator.type={unpooled|pooled}

4.1 以后，非 Android 平台默认启用池化实现，Android 平台启用非池化实现

4.1 之前，池化功能还不成熟，默认是非池化实现

看一下例子：

public class ByteBufStudy {
    public static void main(String[] args) {
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16);
        System.out.println(buffer.getClass());

        buffer = ByteBufAllocator.DEFAULT.heapBuffer(16);
        System.out.println(buffer.getClass());

        buffer = ByteBufAllocator.DEFAULT.directBuffer(16);
        System.out.println(buffer.getClass());
    }
}

1
2
3
4
5
6
7
8
9
10
11
12

打印结果：

// 使用池化的直接内存
class io.netty.buffer.PooledUnsafeDirectByteBuf
    
// 使用池化的堆内存    
class io.netty.buffer.PooledUnsafeHeapByteBuf
    
// 使用池化的直接内存    
class io.netty.buffer.PooledUnsafeDirectByteBuf

1
2
3
4
5
6
7
8

# 2.4 ByteBuf 的组成

ByteBuf 主要有以下几个组成部分：

最大容量与当前容量
- 在构造 ByteBuf 时，可传入两个参数，分别代表初始容量和最大容量，若未传入第二个参数（最大容量），最大容量默认为 Integer.MAX_VALUE
- 当 ByteBuf 容量无法容纳所有数据时，会进行扩容操作，若超出最大容量，会抛出 java.lang.IndexOutOfBoundsException 异常
读写操作不同于 ByteBuffer 只用 position 进行控制，ByteBuf 分别由读指针和写指针两个指针控制。进行读写操作时，无需进行模式的切换
- 读指针前的部分被称为废弃部分，是已经读过的内容
- 读指针与写指针之间的空间称为可读部分
- 写指针与当前容量之间的空间称为可写部分

# 2.5 ByteBuf 的写入

常用方法：

方法签名	含义	备注
writeBoolean(boolean value)	写入 boolean 值	用一字节 01\|00 代表 true\|false
writeByte(int value)	写入 byte 值
writeShort(int value)	写入 short 值
writeInt(int value)	写入 int 值	Big Endian，即 0x250，写入后 00 00 02 50
writeIntLE(int value)	写入 int 值	Little Endian，即 0x250，写入后 50 02 00 00
writeLong(long value)	写入 long 值
writeChar(int value)	写入 char 值
writeFloat(float value)	写入 float 值
writeDouble(double value)	写入 double 值
writeBytes(ByteBuf src)	写入 netty 的 ByteBuf
writeBytes(byte[] src)	写入 byte[]
writeBytes(ByteBuffer src)	写入 nio 的 ByteBuffer
int writeCharSequence(CharSequence sequence, Charset charset)	写入字符串

这些方法的未指明返回值的，其返回值都是 ByteBuf，意味着可以链式调用来写入不同的数据
网络传输中，默认习惯是 Big Endian，使用 writeInt(int value)

# 2.6 扩容

当 ByteBuf 中的容量无法容纳写入的数据时，会进行扩容操作：

buffer.writeLong(7);
ByteBufUtil.log(buffer);

1
2

结果：

// 扩容前
read index:0 write index:12 capacity:16
...

// 扩容后
read index:0 write index:20 capacity:20
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 00 00 00 05 06 00 00 00 00 00 00 00 |................|
|00000010| 00 00 00 07                                     |....            |
+--------+-------------------------------------------------+----------------+

1
2
3
4
5
6
7
8
9
10
11
12

扩容规则：

如何写入后数据大小未超过 512 字节，则选择下一个 16 的整数倍进行扩容
- 例如写入后大小为 12 字节，则扩容后 capacity 是 16 字节
如果写入后数据大小超过 512 字节，则选择下一个 2n
- 例如写入后大小为 513 字节，则扩容后 capacity 是 210=1024 字节（29=512 已经不够了）
扩容不能超过 maxCapacity，否则会抛出 java.lang.IndexOutOfBoundsException 异常

# 2.7 读取

读取主要是通过一系列 read 方法进行读取，读取时会根据读取数据的字节数移动读指针

如果需要重复读取，需要调用 buffer.markReaderIndex() 对读指针进行标记，并通过 buffer.resetReaderIndex() 将读指针恢复到 mark 标记的位置：

public class ByteBufStudy {
    public static void main(String[] args) {
        // 创建ByteBuf
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16, 20);

        // 向buffer中写入数据
        buffer.writeBytes(new byte[]{1, 2, 3, 4});
        buffer.writeInt(5);

        // 读取4个字节
        System.out.println(buffer.readByte());
        System.out.println(buffer.readByte());
        System.out.println(buffer.readByte());
        System.out.println(buffer.readByte());
        ByteBufUtil.log(buffer);

        // 通过mark与reset实现重复读取
        buffer.markReaderIndex();
        System.out.println(buffer.readInt());
        ByteBufUtil.log(buffer);

        // 恢复到mark标记处
        buffer.resetReaderIndex();
        ByteBufUtil.log(buffer);
    }
}

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

打印结果：

1
2
3
4
read index:4 write index:8 capacity:16
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 05                                     |....            |
+--------+-------------------------------------------------+----------------+
5
read index:8 write index:8 capacity:16

read index:4 write index:8 capacity:16
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 05                                     |....            |
+--------+-------------------------------------------------+----------------+

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

还有以 get 开头的一系列方法，这些方法不会改变读指针的位置

# 2.8 ByteBuf 的内存回收

由于 Netty 中有堆外内存（直接内存）的 ByteBuf 实现，堆外内存最好是手动来释放，而不是等 GC 垃圾回收。

UnpooledHeapByteBuf 使用的是 JVM 内存，只需等 GC 回收内存即可
UnpooledDirectByteBuf 使用的就是直接内存了，需要特殊的方法来回收内存
PooledByteBuf 和它的子类使用了池化机制，需要更复杂的规则来回收内存

Netty 这里采用了引用计数法来控制回收内存，每个 ByteBuf 都实现了 ReferenceCounted 接口：

每个 ByteBuf 对象的初始计数为 1
调用 release 方法计数减 1，如果计数为 0，ByteBuf 内存被回收
调用 retain 方法计数加 1，表示调用者没用完之前，其它 handler 即使调用了 release 也不会造成回收
当计数为 0 时，底层内存会被回收，这时即使 ByteBuf 对象还在，其各个方法均无法正常使用

# 2.8.1 释放规则

由于 handler pipeline 的存在，一般需要将 ByteBuf 传递给下一个 ChannelHandler，如果在每个 ChannelHandler 中都去调用 release ，就失去了传递性（如果在这个 ChannelHandler 内这个 ByteBuf 已完成了它的使命，那么便无须再传递）

基本规则是，哪个 handler 是最后使用者，哪个 handler 负责 release：

起点，对于 NIO 实现来讲，在 io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe.read 方法中首次创建 ByteBuf 放入 pipeline
入站 ByteBuf 处理原则
- 对原始 ByteBuf 不做处理，调用 ctx.fireChannelRead(msg) 向后传递，这时无须 release
- 将原始 ByteBuf 转换为其它类型的 Java 对象，这时 ByteBuf 就没用了，必须 release
- 如果不调用 ctx.fireChannelRead(msg) 向后传递，那么也必须 release
- 注意各种异常，如果 ByteBuf 没有成功传递到下一个 ChannelHandler，必须 release
- 假设消息一直向后传，那么 TailContext 会负责释放未处理消息（原始的 ByteBuf）
出站 ByteBuf 处理原则
- 出站消息最终都会转为 ByteBuf 输出，一直向前传，由 HeadContext flush 后 release
异常处理原则
- 有时候不清楚 ByteBuf 被引用了多少次，但又必须彻底释放，可以循环调用 release 直到返回 true

while (!buffer.release()) {}

当 ByteBuf 被传到了 pipeline 的 head 与 tail 时，ByteBuf 会被其中的方法彻底释放，但前提是 ByteBuf 被传递到了 head 与 tail 中。

# 2.8.2 TailConext 中释放 ByteBuf 的源码

protected void onUnhandledInboundMessage(Object msg) {
    try {
        logger.debug("Discarded inbound message {} that reached at the tail of the pipeline. Please check your pipeline configuration.", msg);
    } finally {
        // 具体的释放方法
        ReferenceCountUtil.release(msg);
    }
}

1
2
3
4
5
6
7
8

判断传过来的是否为 ByteBuf，是的话才需要释放：

public static boolean release(Object msg) {
    return msg instanceof ReferenceCounted ? ((ReferenceCounted)msg).release() : false;
}

1
2
3

# 2.9 切片（Slice）

切片操作：ByteBuf slice1 = buffer.slice(0, 5);

ByteBuf 的切片是零拷贝的体现之一。对原始 ByteBuf 进行切片成多个 ByteBuf，切片后的 ByteBuf 并没有发生内存复制，还是使用原始 ByteBuf 的内存，切片后的 ByteBuf 维护独立的 read、write 指针。

这里的“零拷贝”说的是从减少数据复制的角度来说的，并不是说的操作系统的零拷贝。

修改原 ByteBuf 中的值，也会影响切片后得到的 ByteBuf。

得到分片后的 buffer 后，要调用其 retain 方法，使其内部的引用计数加一。避免原 ByteBuf 释放，导致切片 buffer 无法使用

一个使用示例：

public class TestSlice {
    public static void main(String[] args) {
        // 创建ByteBuf
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16, 20);

        // 向buffer中写入数据
        buffer.writeBytes(new byte[]{1, 2, 3, 4, 5, 6, 7, 8, 9, 10});

        // 将buffer分成两部分
        ByteBuf slice1 = buffer.slice(0, 5);
        ByteBuf slice2 = buffer.slice(5, 5);

        // 需要让分片的buffer引用计数加一
        // 避免原Buffer释放导致分片buffer无法使用
        slice1.retain();
        slice2.retain();
        
        ByteBufUtil.log(slice1);
        ByteBufUtil.log(slice2);

        // 更改原始buffer中的值
        System.out.println("===========修改原buffer中的值===========");
        buffer.setByte(0,5);

        System.out.println("===========打印slice1===========");
        ByteBufUtil.log(slice1);
    }
}

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

注意：

切片后的 ByteBuf 在底层上是复用的原有内存区间，切片后的 max capacity 被固定为这个区间的大小，因此不能做追加 write。

# 2.10 duplicate

也是【零拷贝】的体现之一，就好比截取了原始 ByteBuf 所有内容，并且没有 max capacity 的限制，也是与原始 ByteBuf 使用同一块底层内存，只是读写指针是独立的。

# 2.11 copy

会将底层内存数据进行深拷贝，因此无论读写，都与原始 ByteBuf 无关

# 2.12 CompositeByteBuf

【零拷贝】的体现之一，可以将多个 ByteBuf 合并为一个逻辑上的 ByteBuf，避免拷贝。

示例：

// buf1
ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5);
buf1.writeBytes(new byte[]{1, 2, 3, 4, 5});
// buf2
ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5);
buf2.writeBytes(new byte[]{6, 7, 8, 9, 10});
// composite
CompositeByteBuf buf3 = ByteBufAllocator.DEFAULT.compositeBuffer();
// true 表示增加新的 ByteBuf 自动递增 write index, 否则 write index 会始终为 0
buf3.addComponents(true, buf1, buf2);

1
2
3
4
5
6
7
8
9
10

CompositeByteBuf 是一个组合的 ByteBuf，它内部维护了一个 Component 数组，每个 Component 管理一个 ByteBuf，记录了这个 ByteBuf 相对于整体偏移量等信息，代表着整体中某一段的数据。

优点，对外是一个虚拟视图，组合这些 ByteBuf 不会产生内存复制
缺点，复杂了很多，多次操作会带来性能的损耗

# 2.13 Unpooled

Unpooled 是一个工具类，类如其名，提供了非池化的 ByteBuf 创建、组合、复制等操作

这里仅介绍其跟【零拷贝】相关的 wrappedBuffer 方法，可以用来包装 ByteBuf：

ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5);
buf1.writeBytes(new byte[]{1, 2, 3, 4, 5});
ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5);
buf2.writeBytes(new byte[]{6, 7, 8, 9, 10});

// 当包装 ByteBuf 个数超过一个时, 底层使用了 CompositeByteBuf
ByteBuf buf3 = Unpooled.wrappedBuffer(buf1, buf2);
System.out.println(ByteBufUtil.prettyHexDump(buf3));

1
2
3
4
5
6
7
8

也可以用来包装普通字节数组，底层也不会有拷贝操作：

ByteBuf buf4 = Unpooled.wrappedBuffer(new byte[]{1, 2, 3}, new byte[]{4, 5, 6});
System.out.println(buf4.getClass());
System.out.println(ByteBufUtil.prettyHexDump(buf4));

1
2
3

# 2.14 💡 ByteBuf 优势总结

池化 - 可以重用池中 ByteBuf 实例，更节约内存，减少内存溢出的可能
读写指针分离，不需要像 ByteBuffer 一样切换读写模式
可以自动扩容
支持链式调用，使用更流畅
很多地方体现零拷贝，例如 slice、duplicate、CompositeByteBuf

# 3. 双向通信

# 3.1 一个关于读和写的误解

我最初在认识上有这样的误区，认为只有在 netty，nio 这样的多路复用 IO 模型时，读写才不会相互阻塞，才可以实现高效的双向通信，但实际上，Java Socket 是全双工的：在任意时刻，线路上存在A 到 B 和 B 到 A 的双向信号传输。即使是阻塞 IO，读和写是可以同时进行的，只要分别采用读线程和写线程即可，读不会阻塞写、写也不会阻塞读。

# 3.2 练习：echo server

编写 server：

new ServerBootstrap()
    .group(new NioEventLoopGroup())
    .channel(NioServerSocketChannel.class)
    .childHandler(new ChannelInitializer<NioSocketChannel>() {
        @Override
        protected void initChannel(NioSocketChannel ch) {
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                    ByteBuf buffer = (ByteBuf) msg;
                    System.out.println(buffer.toString(Charset.defaultCharset()));

                    // 建议使用 ctx.alloc() 创建 ByteBuf
                    ByteBuf response = ctx.alloc().buffer();
                    response.writeBytes(buffer);
                    ctx.writeAndFlush(response);

                    // 思考：需要释放 buffer 吗
                    // 思考：需要释放 response 吗
                }
            });
        }
    }).bind(8080);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

编写 client：

NioEventLoopGroup group = new NioEventLoopGroup();
Channel channel = new Bootstrap()
    .group(group)
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<NioSocketChannel>() {
        @Override
        protected void initChannel(NioSocketChannel ch) throws Exception {
            ch.pipeline().addLast(new StringEncoder());
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                    ByteBuf buffer = (ByteBuf) msg;
                    System.out.println(buffer.toString(Charset.defaultCharset()));

                    // 思考：需要释放 buffer 吗
                }
            });
        }
    }).connect("127.0.0.1", 8080).sync().channel();

channel.closeFuture().addListener(future -> {
    group.shutdownGracefully();
});

new Thread(() -> {
    Scanner scanner = new Scanner(System.in);
    while (true) {
        String line = scanner.nextLine();
        if ("q".equals(line)) {
            channel.close();
            break;
        }
        channel.writeAndFlush(line);
    }
}).start();

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

编辑

上次更新: 2023/11/18, 07:24:51

← 基本组件：EventLoop、Channel、Future 与 Promise 应用一：粘包与半包问题→