Future&Promise
Netty中的Future与jdk中的Future同名，但是是两个接口，netty的Future继承自jdk的Future,而Promise又对netty Future进行了扩展
- jdk Future只能同步等待任务结束(或成功、或失败)才能得到结果
- netty Future可以同步等待任务结束得到结果，也可以异步方式得到结果，但是要等到任务结束
- netty Promise不仅有netty Future的功能，而且脱离了任务独立存在，只作为两个线程间传递结果的容器
| 功能/名称 | jdk Future | netty Future | Promise |
|---|---|---|---|
| cancel | 取消任务 | - | - |
| isCanceled | 任务是否取消 | - | - |
| isDone | 任务是否完成，不能区分成功失败 | - | - |
| get | 获取任务结果，阻塞等待 | - | - |
| getNow | - | 获取任务结果，非阻塞，还未产生结果时返回 null | - |
| await | - | 等待任务结束，如果任务失败，不会抛异常，而是通过 isSuccess 判断 | - |
| sync | - | 等待任务结束，如果任务失败，抛出异常 | - |
| isSuccess | - | 判断任务是否成功 | - |
| cause | - | 获取失败信息，非阻塞，如果没有失败，返回null | - |
| addLinstener | - | 添加回调，异步接收结果 | - |
| setSuccess | - | - | 设置成功结果 |
| setFailure | - | - | 设置失败结果 |
JDK Future
@Slf4j
public class Fuature1 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService pool = Executors.newFixedThreadPool(5);
Future<? extends Integer> future = pool.submit((Callable<? extends Integer>) () -> {
log.info("waiting ...");
TimeUnit.SECONDS.sleep(3);
return 60;
});
log.info("---");
Integer result = future.get();
log.info("result:{}",result);
}
}
Netty Future
@Slf4j
public class Future2 {
public static void main(String[] args) {
NioEventLoopGroup loopGroup = new NioEventLoopGroup();
EventLoop eventLoop = loopGroup.next();
Future<? extends Integer> future = eventLoop.submit((Callable<? extends Integer>) () -> {
log.info("task runing ...");
TimeUnit.SECONDS.sleep(3);
return 80;
});
future.addListener(new FutureListener<Integer>(){
@Override
public void operationComplete(Future<Integer> integerFuture) throws Exception {
Integer result = integerFuture.getNow();
log.info("result:{}",result);
}
});
log.info("waiting ...");
}
}
Promise
Promise相当于一个容器，可以用于存放各个线程中的结果，然后让其他线程去获取该结果
@Slf4j
public class NettyPromise {
public static void main(String[] args) throws ExecutionException, InterruptedException {
NioEventLoopGroup group=new NioEventLoopGroup();
Promise<Integer> promise=new DefaultPromise<>(group.next());
new Thread(()->{
try {
log.info("calc ...");
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
promise.setFailure(e);
}
promise.setSuccess(100);
}).start();
log.info("result:{}",promise.get());#### Future&Promise
Netty中的Future与jdk中的Future同名，但是是两个接口，netty的Future继承自jdk的Future,而Promise又对netty Future进行了扩展
* jdk Future只能同步等待任务结束(或成功、或失败)才能得到结果
* netty Future可以同步等待任务结束得到结果，也可以异步方式得到结果，但是要等到任务结束
* netty Promise不仅有netty Future的功能，而且脱离了任务独立存在，只作为两个线程间传递结果的容器
| 功能/名称 | jdk Future | netty Future | Promise |
| ------------ | ------------------------------ | ------------------------------------------------------------ | ------------ |
| cancel | 取消任务 | - | - |
| isCanceled | 任务是否取消 | - | - |
| isDone | 任务是否完成，不能区分成功失败 | - | - |
| get | 获取任务结果，阻塞等待 | - | - |
| getNow | - | 获取任务结果，非阻塞，还未产生结果时返回 null | - |
| await | - | 等待任务结束，如果任务失败，不会抛异常，而是通过 isSuccess 判断 | - |
| sync | - | 等待任务结束，如果任务失败，抛出异常 | - |
| isSuccess | - | 判断任务是否成功 | - |
| cause | - | 获取失败信息，非阻塞，如果没有失败，返回null | - |
| addLinstener | - | 添加回调，异步接收结果 | - |
| setSuccess | - | - | 设置成功结果 |
| setFailure | - | - | 设置失败结果 |
**JDK Future**
```java
@Slf4j
public class Fuature1 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService pool = Executors.newFixedThreadPool(5);
Future<? extends Integer> future = pool.submit((Callable<? extends Integer>) () -> {
log.info("waiting ...");
TimeUnit.SECONDS.sleep(3);
return 60;
});
log.info("---");
Integer result = future.get();
log.info("result:{}",result);
}
}
Netty Future
@Slf4j
public class Future2 {
public static void main(String[] args) {
NioEventLoopGroup loopGroup = new NioEventLoopGroup();
EventLoop eventLoop = loopGroup.next();
Future<? extends Integer> future = eventLoop.submit((Callable<? extends Integer>) () -> {
log.info("task runing ...");
TimeUnit.SECONDS.sleep(3);
return 80;
});
future.addListener(new FutureListener<Integer>(){
@Override
public void operationComplete(Future<Integer> integerFuture) throws Exception {
Integer result = integerFuture.getNow();
log.info("result:{}",result);
}
});
log.info("waiting ...");
}
}
Promise
Promise相当于一个容器，可以用于存放各个线程中的结果，然后让其他线程去获取该结果
@Slf4j
public class NettyPromise {
public static void main(String[] args) throws ExecutionException, InterruptedException {
NioEventLoopGroup group=new NioEventLoopGroup();
Promise<Integer> promise=new DefaultPromise<>(group.next());
new Thread(()->{
try {
log.info("calc ...");
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
promise.setFailure(e);
}
promise.setSuccess(100);
}).start();
log.info("result:{}",promise.get());
}
Handler&Pipline
ChannelHandler用来处理Channel上的各种事件，分为入站、出站两种。所有ChannelHandler被连成一串，就是Pipline
- 入站处理器通常是ChannelInboundHandlerAdapter的子类，主要用来读取客户端数据，写回结果
- 出站处理器通常是ChannelOutboundHandlerAdapter的子类，主要对写回结果进行加工
打个比喻，每个Channel是一个产品的加工车间，Pipline是车间中的流水线，ChannelHandler就是流水线上的各道工序，而后面的ByteBuf就是原材料，经过很多工序的加工：先经过一道道入站工序，再经过一道道出站工序最终变为产品
package com.vmware.netty.utils.s5;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import java.nio.charset.StandardCharsets;
@Slf4j
public class PiplineServer {
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().addLast("handler1",new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("1");
ctx.fireChannelRead(msg);
}
});
socketChannel.pipeline().addLast("handler2",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("2");
ctx.fireChannelRead(msg);
}
});
socketChannel.pipeline().addLast("handler3",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("3");
socketChannel.writeAndFlush(ctx.alloc().buffer().writeBytes("Server...".getBytes(StandardCharsets.UTF_8)));
super.channelRead(ctx,msg);
}
});
socketChannel.pipeline().addLast("handler4",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("4");
super.write(ctx, msg, promise);
}
});
socketChannel.pipeline().addLast("handler5",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("5");
super.write(ctx, msg, promise);
}
});
socketChannel.pipeline().addLast("handler6",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("6");
super.write(ctx, msg, promise);
}
});
}
}).bind(8888);
}
}
输出结果
2023-04-06 00:15:25.721 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer - 1
2023-04-06 00:15:25.722 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer - 2
2023-04-06 00:15:25.722 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer - 3
2023-04-06 00:15:25.722 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer - 6
2023-04-06 00:15:25.723 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer - 5
2023-04-06 00:15:25.723 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer - 4
可以看到，ChannelInboundHandlerAdapter是按照addLast的执行顺序执行的，而ChannelOutboundHandlerAdapter是按照addLast的逆序执行的。ChannelPipline的实现是一个ChannelHandlerContext(包装了ChannelHandler)组成的双向链表
在通过channel.pipline().addLast(name,handler)添加handler时，可以为handler取名字。这样可以调用pipline的addAfter、addBefore等方法更灵活的向pipline中添加handler
- pipline是一个结构带有head与tail指针的双向链表，其中的节点为handler
- 通过ctx.fireChannelRead(msg)等方法，将当前handler的处理结果传递给下一个handler
- 当有入站(Inbound)操作时，会从head开始向后调用handler，直到handler不是处理Inbound操作为止
- 当有出站(Outbound)操作时，会从tail开始向前调用handler，直到handler不是处理Outbound操作为止
OutboundHandler
- socketChannel.writeAndFlush()
当handler中调用该方法进行写操作时，会触发Outbound操作，此时是从tail向前寻找OutboundHandler
- ctx.writeAndFlush()
当handler中调用该方法进行写操作时，会触发Outbound操作，此时是从当前handler向前寻找OutboundHandler
修改服务器端代码
socketChannel.pipeline().addLast("handler4",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("4");
super.write(ctx, msg, promise);
}
});
socketChannel.pipeline().addLast("handler3",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("3");
ctx.writeAndFlush(ctx.alloc().buffer().writeBytes("Server...".getBytes(StandardCharsets.UTF_8)));
ctx.fireChannelRead(msg);
}
});
输出
2023-04-06 00:29:16.381 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer2 - 1
2023-04-06 00:29:16.381 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer2 - 2
2023-04-06 00:29:16.381 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer2 - 3
2023-04-06 00:29:16.382 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer2 - 4
EmbeddedChannel
EmbeddedChannel可以用于测试各种handler,通过其构造函数按顺序传入需要测试的handler,然后调用对应的Inbound和Outbound方法即可
@Slf4j
public class EmbededChannel {
public static void main(String[] args) {
ChannelInboundHandlerAdapter c1 = new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("1");
ctx.fireChannelRead(msg);
}
};
ChannelInboundHandlerAdapter c2 = new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("2");
ctx.fireChannelRead(msg);
}
};
ChannelInboundHandlerAdapter c3 = new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("3");
ctx.fireChannelRead(msg);
}
};
ChannelOutboundHandlerAdapter o1 = new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("4");
super.write(ctx,msg,promise);
}
};
ChannelOutboundHandlerAdapter o2 = new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("5");
super.write(ctx,msg,promise);
}
};
EmbeddedChannel channel = new EmbeddedChannel(c1, c2, c3, o1, o2);
//执行Inbound
channel.writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hello".getBytes(StandardCharsets.UTF_8)));
//执行Outbound
channel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hello".getBytes(StandardCharsets.UTF_8)));
}
}
ByteBuf
创建调试工具
public class BufUtil {
public 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());
}
}
该方法可以帮助我们更为详细地查看ByteBuf中的内容
创建ByteBuf
@Slf4j
public class ByteBufStudy {
public static void main(String[] args) {
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16);
BufUtil.log(buffer);
StringBuilder builder = new StringBuilder();
for (int index = 0; index < 20; index++) {
builder.append("a");
}
buffer.writeBytes(builder.toString().getBytes(StandardCharsets.UTF_8));
BufUtil.log(buffer);
}
}
执行结果
read index:0 write index:0 capacity:16
read index:0 write index:20 capacity:64
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000010| 61 61 61 61 |aaaa |
+--------+-------------------------------------------------+----------------+
创建方式
ByteBuf可以通过ByteBufAllocator选择allocator并调用对应的buffer方法来创建，默认使用直接内存作为ByteBuf，容量为256个字节，可以指定初始容量大小
当ByteBuf的容量无法容纳所有数据时，ByteBuf会自动进行扩容操作
当在handler中创建ByteBuf，建议使用ChannelHandlerContext ctx.alloc().buffer()来创建
直接内存与堆内存
声明直接内存类型的ByteBuf
//方式1
ByteBufAllocator.DEFAULT.buffer();
//方式2
ByteBufAllocator.DEFAULT.directBuffer();
声明堆内存类型的ByteBuf
ByteBufAllocator.DEFAULT.heapBuffer(16);
- 直接内存创建和销毁的代价昂贵，但读写性能高（少一次内存复制），适合配合池化功能一起用
- 直接内存对 GC 压力小，因为这部分内存不受 JVM 垃圾回收的管理，但也要注意及时主动释放
示例
@Slf4j
public class ByteBufStudy2 {
public static void main(String[] args) {
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
log.info("type:{}",buffer.getClass());
ByteBuf buffer1 = ByteBufAllocator.DEFAULT.heapBuffer();
log.info("type:{}",buffer1.getClass());
ByteBuf buffer2 = ByteBufAllocator.DEFAULT.directBuffer();
log.info("type:{}",buffer2.getClass());
}
}
输出
2023-04-06 01:35:13.138 [main] INFO com.vmware.netty.utils.buffer.ByteBufStudy2 - type:class io.netty.buffer.PooledUnsafeDirectByteBuf
2023-04-06 01:35:13.145 [main] INFO com.vmware.netty.utils.buffer.ByteBufStudy2 - type:class io.netty.buffer.PooledUnsafeHeapByteBuf
2023-04-06 01:35:13.145 [main] INFO com.vmware.netty.utils.buffer.ByteBufStudy2 - type:class io.netty.buffer.PooledUnsafeDirectByteBuf
池化与非池化
池化的最大意义在于可以重用 ByteBuf，优点有
- 没有池化，则每次都得创建新的 ByteBuf 实例，这个操作对直接内存代价昂贵，就算是堆内存，也会增加 GC 压力
- 有了池化，则可以重用池中 ByteBuf 实例，并且采用了与 jemalloc 类似的内存分配算法提升分配效率
- 高并发时，池化功能更节约内存，减少内存溢出的可能
池化功能是否开启，可以通过下面的系统环境变量来设置
-Dio.netty.allocator.type={unpooled|pooled}Copy
- 4.1 以后，非 Android 平台默认启用池化实现，Android 平台启用非池化实现
- 4.1 之前，池化功能还不成熟，默认是非池化实现
组成
ByteBuf主要由一下几个组成部分
最大容量与当前容量
- 在构造ByteBuf时，可以传入两个参数，分别代表初始容量和最大容量，若未传入第二个参数（最大容量），最大容量默认为Integer.MAX_VALUE
- 当ByteBuf容量无法容纳所有数据时，会进行扩容操作，若超出最大容量，会抛出
IndexOutOfBoundsException读写操作不同于ByteBuffer只用position进行控制，ByteBuf分别由读指针和写指针两个指针控制。进行读写操作时，无需进行模式的切换
- 读指针前的部分被称为废弃部分，是已经读过的内容
- 读指针与写指针之间的空间称为可读部分
- 写指针与当前容量之间的空间称为可写部分
最开始读写指针都在 0 位置
}
#### Handler&Pipline
`ChannelHandler`用来处理Channel上的各种事件，分为入站、出站两种。所有ChannelHandler被连成一串，就是Pipline
* 入站处理器通常是ChannelInboundHandlerAdapter的子类，主要用来读取客户端数据，写回结果
* 出站处理器通常是ChannelOutboundHandlerAdapter的子类，主要对写回结果进行加工
打个比喻，每个Channel是一个产品的加工车间，Pipline是车间中的流水线，ChannelHandler就是流水线上的各道工序，而后面的ByteBuf就是原材料，经过很多工序的加工：先经过一道道入站工序，再经过一道道出站工序最终变为产品
```java
package com.vmware.netty.utils.s5;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import java.nio.charset.StandardCharsets;
@Slf4j
public class PiplineServer {
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().addLast("handler1",new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("1");
ctx.fireChannelRead(msg);
}
});
socketChannel.pipeline().addLast("handler2",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("2");
ctx.fireChannelRead(msg);
}
});
socketChannel.pipeline().addLast("handler3",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("3");
socketChannel.writeAndFlush(ctx.alloc().buffer().writeBytes("Server...".getBytes(StandardCharsets.UTF_8)));
super.channelRead(ctx,msg);
}
});
socketChannel.pipeline().addLast("handler4",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("4");
super.write(ctx, msg, promise);
}
});
socketChannel.pipeline().addLast("handler5",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("5");
super.write(ctx, msg, promise);
}
});
socketChannel.pipeline().addLast("handler6",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("6");
super.write(ctx, msg, promise);
}
});
}
}).bind(8888);
}
}
输出结果
2023-04-06 00:15:25.721 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer - 1
2023-04-06 00:15:25.722 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer - 2
2023-04-06 00:15:25.722 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer - 3
2023-04-06 00:15:25.722 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer - 6
2023-04-06 00:15:25.723 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer - 5
2023-04-06 00:15:25.723 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer - 4
可以看到，ChannelInboundHandlerAdapter是按照addLast的执行顺序执行的，而ChannelOutboundHandlerAdapter是按照addLast的逆序执行的。ChannelPipline的实现是一个ChannelHandlerContext(包装了ChannelHandler)组成的双向链表
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-xJjoHjq2-1680802895652)(img/pipline.png)]
在通过channel.pipline().addLast(name,handler)添加handler时，可以为handler取名字。这样可以调用pipline的addAfter、addBefore等方法更灵活的向pipline中添加handler
- pipline是一个结构带有head与tail指针的双向链表，其中的节点为handler
- 通过ctx.fireChannelRead(msg)等方法，将当前handler的处理结果传递给下一个handler
- 当有入站(Inbound)操作时，会从head开始向后调用handler，直到handler不是处理Inbound操作为止
- 当有出站(Outbound)操作时，会从tail开始向前调用handler，直到handler不是处理Outbound操作为止
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZYaUkPwV-1680802895653)(img/pipline_call.png)]
OutboundHandler
- socketChannel.writeAndFlush()
当handler中调用该方法进行写操作时，会触发Outbound操作，此时是从tail向前寻找OutboundHandler
- ctx.writeAndFlush()
当handler中调用该方法进行写操作时，会触发Outbound操作，此时是从当前handler向前寻找OutboundHandler
修改服务器端代码
socketChannel.pipeline().addLast("handler4",new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("4");
super.write(ctx, msg, promise);
}
});
socketChannel.pipeline().addLast("handler3",new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("3");
ctx.writeAndFlush(ctx.alloc().buffer().writeBytes("Server...".getBytes(StandardCharsets.UTF_8)));
ctx.fireChannelRead(msg);
}
});
输出
2023-04-06 00:29:16.381 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer2 - 1
2023-04-06 00:29:16.381 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer2 - 2
2023-04-06 00:29:16.381 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer2 - 3
2023-04-06 00:29:16.382 [nioEventLoopGroup-2-2] INFO com.vmware.netty.utils.s5.PiplineServer2 - 4
EmbeddedChannel
EmbeddedChannel可以用于测试各种handler,通过其构造函数按顺序传入需要测试的handler,然后调用对应的Inbound和Outbound方法即可
@Slf4j
public class EmbededChannel {
public static void main(String[] args) {
ChannelInboundHandlerAdapter c1 = new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("1");
ctx.fireChannelRead(msg);
}
};
ChannelInboundHandlerAdapter c2 = new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("2");
ctx.fireChannelRead(msg);
}
};
ChannelInboundHandlerAdapter c3 = new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("3");
ctx.fireChannelRead(msg);
}
};
ChannelOutboundHandlerAdapter o1 = new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("4");
super.write(ctx,msg,promise);
}
};
ChannelOutboundHandlerAdapter o2 = new ChannelOutboundHandlerAdapter(){
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
log.info("5");
super.write(ctx,msg,promise);
}
};
EmbeddedChannel channel = new EmbeddedChannel(c1, c2, c3, o1, o2);
//执行Inbound
channel.writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hello".getBytes(StandardCharsets.UTF_8)));
//执行Outbound
channel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hello".getBytes(StandardCharsets.UTF_8)));
}
}
ByteBuf
创建调试工具
public class BufUtil {
public 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());
}
}
该方法可以帮助我们更为详细地查看ByteBuf中的内容
创建ByteBuf
@Slf4j
public class ByteBufStudy {
public static void main(String[] args) {
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16);
BufUtil.log(buffer);
StringBuilder builder = new StringBuilder();
for (int index = 0; index < 20; index++) {
builder.append("a");
}
buffer.writeBytes(builder.toString().getBytes(StandardCharsets.UTF_8));
BufUtil.log(buffer);
}
}
执行结果
read index:0 write index:0 capacity:16
read index:0 write index:20 capacity:64
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000010| 61 61 61 61 |aaaa |
+--------+-------------------------------------------------+----------------+
创建方式
ByteBuf可以通过ByteBufAllocator选择allocator并调用对应的buffer方法来创建，默认使用直接内存作为ByteBuf，容量为256个字节，可以指定初始容量大小
当ByteBuf的容量无法容纳所有数据时，ByteBuf会自动进行扩容操作
当在handler中创建ByteBuf，建议使用ChannelHandlerContext ctx.alloc().buffer()来创建
直接内存与堆内存
声明直接内存类型的ByteBuf
//方式1
ByteBufAllocator.DEFAULT.buffer();
//方式2
ByteBufAllocator.DEFAULT.directBuffer();
声明堆内存类型的ByteBuf
ByteBufAllocator.DEFAULT.heapBuffer(16);
- 直接内存创建和销毁的代价昂贵，但读写性能高（少一次内存复制），适合配合池化功能一起用
- 直接内存对 GC 压力小，因为这部分内存不受 JVM 垃圾回收的管理，但也要注意及时主动释放
示例
@Slf4j
public class ByteBufStudy2 {
public static void main(String[] args) {
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
log.info("type:{}",buffer.getClass());
ByteBuf buffer1 = ByteBufAllocator.DEFAULT.heapBuffer();
log.info("type:{}",buffer1.getClass());
ByteBuf buffer2 = ByteBufAllocator.DEFAULT.directBuffer();
log.info("type:{}",buffer2.getClass());
}
}
输出
2023-04-06 01:35:13.138 [main] INFO com.vmware.netty.utils.buffer.ByteBufStudy2 - type:class io.netty.buffer.PooledUnsafeDirectByteBuf
2023-04-06 01:35:13.145 [main] INFO com.vmware.netty.utils.buffer.ByteBufStudy2 - type:class io.netty.buffer.PooledUnsafeHeapByteBuf
2023-04-06 01:35:13.145 [main] INFO com.vmware.netty.utils.buffer.ByteBufStudy2 - type:class io.netty.buffer.PooledUnsafeDirectByteBuf
池化与非池化
池化的最大意义在于可以重用 ByteBuf，优点有
- 没有池化，则每次都得创建新的 ByteBuf 实例，这个操作对直接内存代价昂贵，就算是堆内存，也会增加 GC 压力
- 有了池化，则可以重用池中 ByteBuf 实例，并且采用了与 jemalloc 类似的内存分配算法提升分配效率
- 高并发时，池化功能更节约内存，减少内存溢出的可能
池化功能是否开启，可以通过下面的系统环境变量来设置
-Dio.netty.allocator.type={unpooled|pooled}Copy
- 4.1 以后，非 Android 平台默认启用池化实现，Android 平台启用非池化实现
- 4.1 之前，池化功能还不成熟，默认是非池化实现
组成
ByteBuf主要由一下几个组成部分
最大容量与当前容量
- 在构造ByteBuf时，可以传入两个参数，分别代表初始容量和最大容量，若未传入第二个参数（最大容量），最大容量默认为Integer.MAX_VALUE
- 当ByteBuf容量无法容纳所有数据时，会进行扩容操作，若超出最大容量，会抛出
IndexOutOfBoundsException读写操作不同于ByteBuffer只用position进行控制，ByteBuf分别由读指针和写指针两个指针控制。进行读写操作时，无需进行模式的切换
- 读指针前的部分被称为废弃部分，是已经读过的内容
- 读指针与写指针之间的空间称为可读部分
- 写指针与当前容量之间的空间称为可写部分
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-rTw42YEu-1680802895653)(img/bytebuf.png)]
最开始读写指针都在 0 位置
写入
常用方法
| 方法签名 | 含义 | 备注 |
|---|---|---|
| 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) | 写入字符串 | CharSequence为字符串类的父类，第二个参数为对应的字符集 |
注意
- 这些方法的未指明返回值的，其返回值都是 ByteBuf，意味着可以链式调用来写入不同的数据
- 网络传输中，默认习惯是 Big Endian（大端写入），使用 writeInt(int value)
还有一类方法是 set 开头的一系列方法，也可以写入数据，但不会改变写指针位置
扩容
当ByteBuf中的容量无法容纳写入的数据时，会自动进行扩容操作
扩容规则
- 如何写入后数据大小未超过 512 字节，则选择下一个 16 的整数倍进行扩容
- 例如写入后大小为 12 字节，则扩容后 capacity 是 16 字节
- 如果写入后数据大小超过 512 字节，则选择下一个 2^n
- 例如写入后大小为 513 字节，则扩容后 capacity 是 210=1024 字节（2^9=512 已经不够了）
- 扩容不能超过 maxCapacity，否则会抛出
java.lang.IndexOutOfBoundsException异常
Exception in thread "main" java.lang.IndexOutOfBoundsException: writerIndex(20) + minWritableBytes(8) exceeds maxCapacity(20): PooledUnsafeDirectByteBuf(ridx: 0, widx: 20, cap: 20/20)
...
读取
读取主要是通过一系列read方法进行读取，读取时会根据读取数据的字节数移动读指针
@Slf4j
public class ByteBufTest {
public static void main(String[] args) {
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
buffer.writeBytes(new byte[]{1, 2, 3, 4, 5, 6});
log.info("{}",buffer.readByte());
log.info("{}",buffer.readByte());
log.info("{}",buffer.readByte());
log.info("{}",buffer.readByte());
BufUtil.log(buffer);
}
}
读过的内容，就属于废弃部分了，再读只能读那些尚未读取的部分
2023-04-07 00:50:32.527 [main] INFO com.vmware.netty.utils.buf.ByteBufTest - 1
2023-04-07 00:50:32.527 [main] INFO com.vmware.netty.utils.buf.ByteBufTest - 2
2023-04-07 00:50:32.527 [main] INFO com.vmware.netty.utils.buf.ByteBufTest - 3
2023-04-07 00:50:32.527 [main] INFO com.vmware.netty.utils.buf.ByteBufTest - 4
read index:4 write index:6 capacity:256
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 05 06 |.. |
+--------+-------------------------------------------------+----------------+
如果需要重复读取，需要调用buffer.markReaderIndex()对读指针进行标记，并通过buffer.resetReaderIndex()将读指针恢复到mark标记的位置
@Slf4j
public class ByteBufTest {
public static void main(String[] args) {
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
buffer.writeBytes(new byte[]{1, 2, 3, 4, 5, 6});
log.info("{}",buffer.readByte());
buffer.markReaderIndex();//标记
log.info("{}",buffer.readByte());
log.info("{}",buffer.readByte());
log.info("{}",buffer.readByte());
buffer.resetReaderIndex();//回到标记位
log.info("{}",buffer.readByte());
}
}
输出
2023-04-07 00:53:40.216 [main] DEBUG io.netty.util.ResourceLeakDetectorFactory - Loaded default ResourceLeakDetector: io.netty.util.ResourceLeakDetector@4d49af10
2023-04-07 00:53:40.222 [main] INFO com.vmware.netty.utils.buf.ByteBufTest - 1
2023-04-07 00:53:40.222 [main] INFO com.vmware.netty.utils.buf.ByteBufTest - 2
2023-04-07 00:53:40.222 [main] INFO com.vmware.netty.utils.buf.ByteBufTest - 3
2023-04-07 00:53:40.222 [main] INFO com.vmware.netty.utils.buf.ByteBufTest - 4
2023-04-07 00:53:40.222 [main] INFO com.vmware.netty.utils.buf.ByteBufTest - 2
还有种办法是采用 get 开头的一系列方法，这些方法不会改变 read index
释放
由于 Netty 中有堆外内存的 ByteBuf 实现，堆外内存最好是手动来释放，而不是等 GC 垃圾回收。
- UnpooledHeapByteBuf 使用的是 JVM 内存，只需等 GC 回收内存即可
- UnpooledDirectByteBuf 使用的就是直接内存了，需要特殊的方法来回收内存
- PooledByteBuf 和它的子类使用了池化机制，需要更复杂的规则来回收内存
回收内存的源码实现，请关注下面方法的不同实现
protected abstract void deallocate()
Netty 这里采用了引用计数法来控制回收内存，每个 ByteBuf 都实现了 ReferenceCounted 接口
- 每个 ByteBuf 对象的初始计数为 1
- 调用 release 方法计数减 1，如果计数为 0，ByteBuf 内存被回收
- 调用 retain 方法计数加 1，表示调用者没用完之前，其它 handler 即使调用了 release 也不会造成回收
- 当计数为 0 时，底层内存会被回收，这时即使 ByteBuf 对象还在，其各个方法均无法正常使用
谁来负责 release 呢？
不是我们想象的（一般情况下）
ByteBuf buf = ...
try {
...
} finally {
buf.release();
}
请思考，因为 pipeline 的存在，一般需要将 ByteBuf 传递给下一个 ChannelHandler，如果在 finally 中 release 了，就失去了传递性（当然，如果在这个 ChannelHandler 内这个 ByteBuf 已完成了它的使命，那么便无须再传递）
基本规则是，谁是最后使用者，谁负责 release，详细分析如下
- 起点，对于 NIO 实现来讲，在 io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe#read 方法中首次创建 ByteBuf 放入 pipeline（line 163 pipeline.fireChannelRead(byteBuf)）
- 入站 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()) {}
TailContext 释放未处理消息逻辑
// io.netty.channel.DefaultChannelPipeline#onUnhandledInboundMessage(java.lang.Object)
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);
}
}
具体代码
// io.netty.util.ReferenceCountUtil#release(java.lang.Object)
public static boolean release(Object msg) {
if (msg instanceof ReferenceCounted) {
return ((ReferenceCounted) msg).release();
}
return false;
}
slice
【零拷贝】的体现之一，对原始 ByteBuf 进行切片成多个 ByteBuf，切片后的 ByteBuf 并没有发生内存复制，还是使用原始 ByteBuf 的内存，切片后的 ByteBuf 维护独立的 read，write 指针
无参 slice 是从原始 ByteBuf 的 read index 到 write index 之间的内容进行切片，切片后的 max capacity 被固定为这个区间的大小，因此不能追加 write
public class SliceTest {
public static void main(String[] args) {
ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
buf.writeBytes(new byte[]{1, 2, 3, 4, 5, 6, 7, 8});
buf.readByte();
ByteBuf byteBuf = buf.slice();//02 03 04 05 06 07 08
BufUtil.log(byteBuf);
byteBuf.writeByte(11);//error
BufUtil.log(byteBuf);
}
}
输出
read index:0 write index:7 capacity:7
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04 05 06 07 08 |....... |
+--------+-------------------------------------------------+----------------+
Exception in thread "main" java.lang.IndexOutOfBoundsException: writerIndex(7) + minWritableBytes(1) exceeds maxCapacity(7): UnpooledSlicedByteBuf(ridx: 0, widx: 7, cap: 7/7, unwrapped: PooledUnsafeDirectByteBuf(ridx: 1, widx: 8, cap: 256))
at io.netty.buffer.AbstractByteBuf.ensureWritable0(AbstractByteBuf.java:295)
at io.netty.buffer.AbstractByteBuf.writeByte(AbstractByteBuf.java:985)
at com.vmware.netty.utils.buf.SliceTest.main(SliceTest.java:15)
Process finished with exit code 1
如果原始 ByteBuf 再次读操作
public class SliceTest {
public static void main(String[] args) {
ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
buf.writeBytes(new byte[]{1, 2, 3, 4, 5, 6, 7, 8});
buf.readByte();
ByteBuf byteBuf = buf.slice();//02 03 04 05 06 07 08
buf.readByte();//再次读
BufUtil.log(byteBuf);
}
}
输出
read index:0 write index:7 capacity:7
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04 05 06 07 08 |....... |
+--------+-------------------------------------------------+----------------+
这时的byteBuf不受影响，因为它有独立的读写指针
但是如果byteBuf的内容发生了更改,原ByteBuf也会受到影响，因为底层都是同一块内存
public class SliceTest {
public static void main(String[] args) {
ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
buf.writeBytes(new byte[]{1, 2, 3, 4, 5, 6, 7, 8});
buf.readByte();
ByteBuf byteBuf = buf.slice();//02 03 04 05 06 07 08
byteBuf.setByte(0,16);
BufUtil.log(byteBuf);
}
}
输出
read index:0 write index:7 capacity:7
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 10 03 04 05 06 07 08 |....... |
+--------+-------------------------------------------------+----------------+
duplicate
零拷贝的体现之一，就好比截取了原始 ByteBuf 所有内容，并且没有 max capacity 的限制，也是与原始 ByteBuf 使用同一块底层内存，只是读写指针是独立的
copy
会将底层内存数据进行深拷贝，因此无论读写，都与原始 ByteBuf 无关
CompositeByteBuf
零拷贝的体现之一，可以将多个 ByteBuf 合并为一个逻辑上的 ByteBuf，避免拷贝
public class CompositeByteBufTest {
public static void main(String[] args) {
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
buffer.writeBytes(new byte[]{1, 2, 3, 4, 5});
ByteBuf buffer1 = ByteBufAllocator.DEFAULT.buffer();
buffer1.writeBytes(new byte[]{6, 7, 8, 9, 10});
CompositeByteBuf compositeByteBuf = ByteBufAllocator.DEFAULT.compositeBuffer();
//true 表示增加新的 ByteBuf 自动递增 write index, 否则 write index 会始终为 0
compositeByteBuf.addComponents(true,buffer,buffer1);
BufUtil.log(compositeByteBuf);
}
}
输出
read index:0 write index:10 capacity:10
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 07 08 09 0a |.......... |
+--------+-------------------------------------------------+----------------+
CompositeByteBuf 是一个组合的 ByteBuf，它内部维护了一个 Component 数组，每个 Component 管理一个 ByteBuf，记录了这个 ByteBuf 相对于整体偏移量等信息，代表着整体中某一段的数据。
- 优点，对外是一个虚拟视图，组合这些 ByteBuf 不会产生内存复制
- 缺点，复杂了很多，多次操作会带来性能的损耗
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));
输出
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 07 08 09 0a |.......... |
+--------+-------------------------------------------------+----------------+
也可以用来包装普通字节数组，底层也不会有拷贝操作
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));
输出
class io.netty.buffer.CompositeByteBuf
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 |...... |
+--------+-------------------------------------------------+----------------+
💡 ByteBuf 优势
- 池化 - 可以重用池中 ByteBuf 实例，更节约内存，减少内存溢出的可能
- 读写指针分离，不需要像 ByteBuffer 一样切换读写模式
- 可以自动扩容
- 支持链式调用，使用更流畅
- 很多地方体现零拷贝，例如 slice、duplicate、CompositeByteBuf
![JUC并发编程第一章之进程/并发/异步的概念[理解基本概念]](https://img-blog.csdnimg.cn/img_convert/1d754aee5ead868e246609eb2bb3d32d.webp?x-oss-process=image/format,png)
