概述
Netty 是一个异步的、基于事件驱动的网络应用框架，用于快速开发可维护、高性能的网络服务器和客户端
Netty 在 Java 网络应用框架中的地位就好比：Spring 框架在 JavaEE 开发中的地位
以下的框架都使用了 Netty，因为它们有网络通信需求！
为什么会有 Netty？Netty 的优势如下：
Hello World
目标
开发一个简单的服务器端和客户端
客户端向服务器端发送 hello, world
服务器仅接收，不返回
加入依赖
<dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>4.1.39.Final</version></dependency>
复制代码
服务器端
new ServerBootstrap() .group(new NioEventLoopGroup()) // 1 .channel(NioServerSocketChannel.class) // 2 .childHandler(new ChannelInitializer<NioSocketChannel>() { // 3 protected void initChannel(NioSocketChannel ch) { ch.pipeline().addLast(new StringDecoder()); // 5 ch.pipeline().addLast(new SimpleChannelInboundHandler<String>() { // 6 @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) { System.out.println(msg); } }); } }) .bind(8080); // 4
复制代码
代码解读
3 处：为啥方法叫 childHandler，接下来添加的处理器都是给 SocketChannel 用的，而不是给 ServerSocketChannel。ChannelInitializer 处理器（仅执行一次），它的作用是待客户端 SocketChannel 建立连接后，执行 initChannel 以便添加更多的处理器
4 处：ServerSocketChannel 绑定的监听端口
5 处：SocketChannel 的处理器，解码 ByteBuf => String
6 处：SocketChannel 的业务处理器，使用上一个处理器的处理结果
客户端
new Bootstrap() .group(new NioEventLoopGroup()) // 1 .channel(NioSocketChannel.class) // 2 .handler(new ChannelInitializer<Channel>() { // 3 @Override protected void initChannel(Channel ch) { ch.pipeline().addLast(new StringEncoder()); // 8 } }) .connect("127.0.0.1", 8080) // 4 .sync() // 5 .channel() // 6 .writeAndFlush(new Date() + ": hello world!"); // 7
复制代码
代码解读
3 处：添加 SocketChannel 的处理器，ChannelInitializer 处理器（仅执行一次），它的作用是待客户端 SocketChannel 建立连接后，执行 initChannel 以便添加更多的处理器
4 处：指定要连接的服务器和端口
5 处：Netty 中很多方法都是异步的，如 connect，这时需要使用 sync 方法等待 connect 建立连接完毕
6 处：获取 channel 对象，它即为通道抽象，可以进行数据读写操作
7 处：写入消息并清空缓冲区
8 处：消息会经过通道 handler 处理，这里是将 String => ByteBuf 发出
数据经过网络传输，到达服务器端，服务器端 5 和 6 处的 handler 先后被触发，走完一个流程
流程梳理
树立正确的观念
把 channel 理解为数据的通道
把 msg 理解为流动的数据，最开始输入是 ByteBuf，但经过 pipeline 的加工，会变成其它类型对象，最后输出又变成 ByteBuf
把 handler 理解为数据的处理工序工序有多道，合在一起就是 pipeline，pipeline 负责发布事件（读、读取完成...）传播给每个 handler， handler 对自己感兴趣的事件进行处理（重写了相应事件处理方法）handler 分 Inbound 和 Outbound 两类
把 eventLoop 理解为处理数据的工人工人可以管理多个 channel 的 io 操作，并且一旦工人负责了某个 channel，就要负责到底（绑定）工人既可以执行 io 操作，也可以进行任务处理，每位工人有任务队列，队列里可以堆放多个 channel 的待处理任务，任务分为普通任务、定时任务工人按照 pipeline 顺序，依次按照 handler 的规划（代码）处理数据，可以为每道工序指定不同的工人
相关组件
EventLoop
EventLoop 对象
EventLoop 本质是一个单线程执行器（同时维护了一个 Selector），里面有 run 方法处理 Channel 上源源不断的 io 事件。
它的继承关系比较复杂
EventLoop 组
EventLoopGroup 是一组 EventLoop，Channel 一般会调用 EventLoopGroup 的 register 方法来绑定其中一个 EventLoop，后续这个 Channel 上的 io 事件都由此 EventLoop 来处理（保证了 io 事件处理时的线程安全）其中 NioEventLoopGroup 可以处理 IO 事件，普通任务和定时任务；DefaultEventLoopGroup 不能处理 IO 事件，可以处理普通任务和定时任务；
以一个简单的实现为例：
// 内部创建了两个 EventLoop, 每个 EventLoop 维护一个线程DefaultEventLoopGroup group = new DefaultEventLoopGroup(2);System.out.println(group.next());System.out.println(group.next());System.out.println(group.next());
复制代码
输出
io.netty.channel.DefaultEventLoop@60f82f98io.netty.channel.DefaultEventLoop@35f983a6io.netty.channel.DefaultEventLoop@60f82f98
复制代码
也可以使用 for 循环
DefaultEventLoopGroup group = new DefaultEventLoopGroup(2);for (EventExecutor eventLoop : group) { System.out.println(eventLoop);}
复制代码
输出
io.netty.channel.DefaultEventLoop@60f82f98io.netty.channel.DefaultEventLoop@35f983a6
复制代码
NioEventLoop 处理普通任务
NioEventLoop 除了可以处理 io 事件，同样可以向它提交普通任务
NioEventLoopGroup nioWorkers = new NioEventLoopGroup(2);
log.debug("server start...");Thread.sleep(2000);nioWorkers.execute(()->{ log.debug("normal task...");});
复制代码
输出
22:30:36 [DEBUG] [main] c.i.o.EventLoopTest2 - server start...22:30:38 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - normal task...
复制代码
可以用来执行耗时较长的任务
NioEventLoop 处理定时任务
NioEventLoopGroup nioWorkers = new NioEventLoopGroup(2);
log.debug("server start...");Thread.sleep(2000);nioWorkers.scheduleAtFixedRate(() -> { log.debug("running...");}, 0, 1, TimeUnit.SECONDS);
复制代码
输出
22:35:15 [DEBUG] [main] c.i.o.EventLoopTest2 - server start...22:35:17 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - running...22:35:18 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - running...22:35:19 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - running...22:35:20 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - running......
复制代码
可以用来执行定时任务
NioEventLoop 处理 io 事件
服务器端两个 nio worker 工人
new ServerBootstrap() //第一个是boss 只负责accept事件，第二个是 工人，负责socketChannel上的读写 .group(new NioEventLoopGroup(1), new NioEventLoopGroup(2)) .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 byteBuf = msg instanceof ByteBuf ? ((ByteBuf) msg) : null; if (byteBuf != null) { byte[] buf = new byte[16]; ByteBuf len = byteBuf.readBytes(buf, 0, byteBuf.readableBytes()); log.debug(new String(buf)); } } }); } }).bind(8080).sync();
复制代码
客户端，启动三次，分别修改发送字符串为 zhangsan（第一次），lisi（第二次），wangwu（第三次）
public static void main(String[] args) throws InterruptedException { Channel channel = new Bootstrap() .group(new NioEventLoopGroup(1)) .handler(new ChannelInitializer<NioSocketChannel>() { @Override protected void initChannel(NioSocketChannel ch) throws Exception { System.out.println("init..."); ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG)); } }) .channel(NioSocketChannel.class).connect("localhost", 8080) .sync() .channel();
channel.writeAndFlush(ByteBufAllocator.DEFAULT.buffer().writeBytes("wangwu".getBytes())); Thread.sleep(2000); channel.writeAndFlush(ByteBufAllocator.DEFAULT.buffer().writeBytes("wangwu".getBytes()));
复制代码
最后输出
22:03:34 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - zhangsan 22:03:36 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - zhangsan 22:05:36 [DEBUG] [nioEventLoopGroup-3-2] c.i.o.EventLoopTest - lisi 22:05:38 [DEBUG] [nioEventLoopGroup-3-2] c.i.o.EventLoopTest - lisi 22:06:09 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - wangwu 22:06:11 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - wangwu
复制代码
可以看到两个工人轮流处理 channel，但工人与 channel 之间进行了绑定
再增加两个非 nio 工人，用于处理 耗时长的任务
DefaultEventLoopGroup normalWorkers = new DefaultEventLoopGroup(2);new ServerBootstrap() //第一个是boss 只负责accept事件，第二个是 工人，负责socketChannel上的读写 .group(new NioEventLoopGroup(1), new NioEventLoopGroup(2)) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<NioSocketChannel>() { @Override protected void initChannel(NioSocketChannel ch) { ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG)); ch.pipeline().addLast(normalWorkers,"myhandler", //设置使用normalWorkers来处理数据 new ChannelInboundHandlerAdapter() { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { ByteBuf byteBuf = msg instanceof ByteBuf ? ((ByteBuf) msg) : null; if (byteBuf != null) { byte[] buf = new byte[16]; ByteBuf len = byteBuf.readBytes(buf, 0, byteBuf.readableBytes()); log.debug(new String(buf)); } } }); } }).bind(8080).sync();
复制代码
客户端代码不变，启动三次，分别修改发送字符串为 zhangsan（第一次），lisi（第二次），wangwu（第三次）
输出
22:19:48 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x251562d5, L:/127.0.0.1:8080 - R:/127.0.0.1:52588] REGISTERED22:19:48 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x251562d5, L:/127.0.0.1:8080 - R:/127.0.0.1:52588] ACTIVE22:19:48 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x251562d5, L:/127.0.0.1:8080 - R:/127.0.0.1:52588] READ: 8B +-------------------------------------------------+ | 0 1 2 3 4 5 6 7 8 9 a b c d e f |+--------+-------------------------------------------------+----------------+|00000000| 7a 68 61 6e 67 73 61 6e |zhangsan |+--------+-------------------------------------------------+----------------+22:19:48 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x251562d5, L:/127.0.0.1:8080 - R:/127.0.0.1:52588] READ COMPLETE22:19:48 [DEBUG] [defaultEventLoopGroup-2-1] c.i.o.EventLoopTest - zhangsan 22:19:50 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x251562d5, L:/127.0.0.1:8080 - R:/127.0.0.1:52588] READ: 8B +-------------------------------------------------+ | 0 1 2 3 4 5 6 7 8 9 a b c d e f |+--------+-------------------------------------------------+----------------+|00000000| 7a 68 61 6e 67 73 61 6e |zhangsan |+--------+-------------------------------------------------+----------------+22:19:50 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x251562d5, L:/127.0.0.1:8080 - R:/127.0.0.1:52588] READ COMPLETE22:19:50 [DEBUG] [defaultEventLoopGroup-2-1] c.i.o.EventLoopTest - zhangsan 22:20:24 [DEBUG] [nioEventLoopGroup-4-2] i.n.h.l.LoggingHandler - [id: 0x94b2a840, L:/127.0.0.1:8080 - R:/127.0.0.1:52612] REGISTERED22:20:24 [DEBUG] [nioEventLoopGroup-4-2] i.n.h.l.LoggingHandler - [id: 0x94b2a840, L:/127.0.0.1:8080 - R:/127.0.0.1:52612] ACTIVE22:20:25 [DEBUG] [nioEventLoopGroup-4-2] i.n.h.l.LoggingHandler - [id: 0x94b2a840, L:/127.0.0.1:8080 - R:/127.0.0.1:52612] READ: 4B +-------------------------------------------------+ | 0 1 2 3 4 5 6 7 8 9 a b c d e f |+--------+-------------------------------------------------+----------------+|00000000| 6c 69 73 69 |lisi |+--------+-------------------------------------------------+----------------+22:20:25 [DEBUG] [nioEventLoopGroup-4-2] i.n.h.l.LoggingHandler - [id: 0x94b2a840, L:/127.0.0.1:8080 - R:/127.0.0.1:52612] READ COMPLETE22:20:25 [DEBUG] [defaultEventLoopGroup-2-2] c.i.o.EventLoopTest - lisi 22:20:27 [DEBUG] [nioEventLoopGroup-4-2] i.n.h.l.LoggingHandler - [id: 0x94b2a840, L:/127.0.0.1:8080 - R:/127.0.0.1:52612] READ: 4B +-------------------------------------------------+ | 0 1 2 3 4 5 6 7 8 9 a b c d e f |+--------+-------------------------------------------------+----------------+|00000000| 6c 69 73 69 |lisi |+--------+-------------------------------------------------+----------------+22:20:27 [DEBUG] [nioEventLoopGroup-4-2] i.n.h.l.LoggingHandler - [id: 0x94b2a840, L:/127.0.0.1:8080 - R:/127.0.0.1:52612] READ COMPLETE22:20:27 [DEBUG] [defaultEventLoopGroup-2-2] c.i.o.EventLoopTest - lisi 22:20:38 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x79a26af9, L:/127.0.0.1:8080 - R:/127.0.0.1:52625] REGISTERED22:20:38 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x79a26af9, L:/127.0.0.1:8080 - R:/127.0.0.1:52625] ACTIVE22:20:38 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x79a26af9, L:/127.0.0.1:8080 - R:/127.0.0.1:52625] READ: 6B +-------------------------------------------------+ | 0 1 2 3 4 5 6 7 8 9 a b c d e f |+--------+-------------------------------------------------+----------------+|00000000| 77 61 6e 67 77 75 |wangwu |+--------+-------------------------------------------------+----------------+22:20:38 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x79a26af9, L:/127.0.0.1:8080 - R:/127.0.0.1:52625] READ COMPLETE22:20:38 [DEBUG] [defaultEventLoopGroup-2-1] c.i.o.EventLoopTest - wangwu 22:20:40 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x79a26af9, L:/127.0.0.1:8080 - R:/127.0.0.1:52625] READ: 6B +-------------------------------------------------+ | 0 1 2 3 4 5 6 7 8 9 a b c d e f |+--------+-------------------------------------------------+----------------+|00000000| 77 61 6e 67 77 75 |wangwu |+--------+-------------------------------------------------+----------------+22:20:40 [DEBUG] [nioEventLoopGroup-4-1] i.n.h.l.LoggingHandler - [id: 0x79a26af9, L:/127.0.0.1:8080 - R:/127.0.0.1:52625] READ COMPLETE22:20:40 [DEBUG] [defaultEventLoopGroup-2-1] c.i.o.EventLoopTest - wangwu
复制代码
可以看到，nio 工人和 非 nio 工人也分别绑定了 channel（LoggingHandler 由 nio 工人执行，而我们自己的 handler 由非 nio 工人执行）
handler 执行中如何换人？
关键代码 io.netty.channel.AbstractChannelHandlerContext#invokeChannelRead()
static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) { final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next); //返回下一个handler的eventloop EventExecutor executor = next.executor(); // 下一个 handler 的事件循环是否与当前的事件循环是同一个线程 // 是，直接调用 if (executor.inEventLoop()) { next.invokeChannelRead(m); } // 不是，将要执行的代码作为任务提交给下一个事件循环处理（换人） else { executor.execute(new Runnable() { @Override public void run() { next.invokeChannelRead(m); } }); }}
复制代码
Channel
channel 的主要作用
close() 可以用来关闭 channel
closeFuture() 用来处理 channel 的关闭 sync 方法作用是同步等待 channel 关闭而 addListener 方法是异步等待 channel 关闭
pipeline() 方法添加处理器
write() 方法将数据写入，但不一定会立刻刷出，因为还有缓冲区
writeAndFlush() 方法将数据写入并刷出
ChannelFuture
这是刚才的客户端代码
new Bootstrap() .group(new NioEventLoopGroup()) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<Channel>() { @Override protected void initChannel(Channel ch) { ch.pipeline().addLast(new StringEncoder()); } }) .connect("127.0.0.1", 8080) .sync() .channel() .writeAndFlush(new Date() + ": hello world!");
复制代码
现在把它拆开来看
ChannelFuture channelFuture = new Bootstrap() .group(new NioEventLoopGroup()) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<Channel>() { @Override protected void initChannel(Channel ch) { ch.pipeline().addLast(new StringEncoder()); } }) .connect("127.0.0.1", 8080); // 1
channelFuture.sync().channel().writeAndFlush(new Date() + ": hello world!");
复制代码
注意 connect 方法是异步的，意味着不等连接建立，方法执行就返回了。因此 channelFuture 对象中不能【立刻】获得到正确的 Channel 对象
使用 sync 方法同步处理结果
实验如下：
ChannelFuture channelFuture = new Bootstrap() .group(new NioEventLoopGroup()) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<Channel>() { @Override protected void initChannel(Channel ch) { ch.pipeline().addLast(new StringEncoder()); } }) .connect("127.0.0.1", 8080);
System.out.println(channelFuture.channel()); // 1channelFuture.sync(); // 2System.out.println(channelFuture.channel()); // 3
复制代码
执行到 1 时，连接未建立，打印 [id: 0x2e1884dd]
执行到 2 时，sync 方法是同步等待连接建立完成
执行到 3 时，连接肯定建立了，打印 [id: 0x2e1884dd, L:/127.0.0.1:57191 - R:/127.0.0.1:8080]
使用 addListener(回调对象)方法异步处理结果
除了用 sync 方法可以让异步操作同步以外，还可以使用回调的方式：
ChannelFuture channelFuture = new Bootstrap() .group(new NioEventLoopGroup()) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<Channel>() { @Override protected void initChannel(Channel ch) { ch.pipeline().addLast(new StringEncoder()); } }) .connect("127.0.0.1", 8080);System.out.println(channelFuture.channel()); // 1channelFuture.addListener((ChannelFutureListener) future -> { System.out.println(future.channel()); // 2});
复制代码
执行到 1 时，连接未建立，打印 [id: 0x749124ba]
ChannelFutureListener 会在连接建立时被调用（其中 operationComplete 方法），因此执行到 2 时，连接肯定建立了，打印 [id: 0x749124ba, L:/127.0.0.1:57351 - R:/127.0.0.1:8080]
CloseFuture
closeFuture() 用来处理 channel 的关闭
@Slf4jpublic class CloseFutureClient { public static void main(String[] args) throws InterruptedException { NioEventLoopGroup group new NioEventLoopGroup(); ChannelFuture channelFuture = new Bootstrap() .group(group) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<NioSocketChannel>() { @Override // 在连接建立后被调用 protected void initChannel(NioSocketChannel ch) throws Exception { ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG)); ch.pipeline().addLast(new StringEncoder()); } }) .connect(new InetSocketAddress("localhost", 8080)); Channel channel = channelFuture.sync().channel(); log.debug("{}", channel); new Thread(()->{ Scanner scanner = new Scanner(System.in); while (true) { String line = scanner.nextLine(); if ("q".equals(line)) { channel.close(); // close 也是异步操作 //因此如果在这里写"处理关闭之后的操作"，并不一定是在关闭之后执行的，可能关闭之前就执行了// log.debug("处理关闭之后的操作"); break; } channel.writeAndFlush(line); } }, "input").start();
// 获取 CloseFuture 对象， 1) 同步处理关闭， 2) 异步处理关闭 ChannelFuture closeFuture = channel.closeFuture(); /*log.debug("waiting close..."); 1) 同步处理关闭 closeFuture.sync(); log.debug("处理关闭之后的操作");*/ //2) 异步处理关闭 closeFuture.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { log.debug("处理关闭之后的操作"); group.shutdownGracefully(); } }); }}
复制代码
优雅关闭 shutdownGracefully 方法。该方法会首先切换 EventLoopGroup 到关闭状态从而拒绝新的任务的加入，然后在任务队列的任务都处理完成后，停止线程的运行。从而确保整体应用是在正常有序的状态下退出的
异步提升的是什么
要点
Future & Promise
在异步处理时，经常用到这两个接口
首先要说明 netty 中的 Future 与 jdk 中的 Future 同名，但是是两个接口，netty 的 Future 继承自 jdk 的 Future，而 Promise 又对 netty Future 进行了扩展
JDK Future 只能同步等待任务结束（或成功、或失败）才能得到结果
netty Future 可以同步等待任务结束得到结果，也可以异步方式得到结果，但都是要等任务结束
netty Promise 不仅有 netty Future 的功能，而且脱离了任务独立存在，只作为两个线程间传递结果的容器
接下来看几个例子
同步处理任务
DefaultEventLoop eventExecutors = new DefaultEventLoop();//可以主动创建promise，这是一个结果容器DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
eventExecutors.execute(()->{ try { Thread.sleep(1000); log.debug("set success, {}",10); promise.setSuccess(10);//设置成功结果 } catch (InterruptedException e) { e.printStackTrace(); log.debug("set failure, {}", e.toString()); promise.setFailure(e); }});
log.debug("start...");log.debug("{}",promise.getNow()); // 还未产生结果时 会返回 nulllog.debug("{}",promise.get());// sync() 也会出现异常，只是 get 会再用 ExecutionException 包一层异常
复制代码
异步处理任务成功
DefaultEventLoop eventExecutors = new DefaultEventLoop();DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
// 设置回调，异步接收结果promise.addListener(future -> { // 这里的 future 就是上面的 promise log.debug("{}",future.getNow());});
// 等待 1000 后设置成功结果eventExecutors.execute(()->{ try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } log.debug("set success, {}",10); promise.setSuccess(10);});
log.debug("start...");
复制代码
同步处理任务失败 - await
DefaultEventLoop eventExecutors = new DefaultEventLoop();DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
eventExecutors.execute(() -> { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } RuntimeException e = new RuntimeException("error..."); log.debug("set failure, {}", e.toString()); promise.setFailure(e);});
log.debug("start...");log.debug("{}", promise.getNow());promise.await(); // 与 sync 和 get 区别在于，不会抛异常log.debug("result {}", (promise.isSuccess() ? promise.getNow() : promise.cause()).toString());
复制代码
输出
12:18:53 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - start...12:18:53 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - null12:18:54 [DEBUG] [defaultEventLoop-1-1] c.i.o.DefaultPromiseTest2 - set failure, java.lang.RuntimeException: error...12:18:54 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - result java.lang.RuntimeException: error...
复制代码
异步处理任务失败
DefaultEventLoop eventExecutors = new DefaultEventLoop();DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
promise.addListener(future -> { log.debug("result {}", (promise.isSuccess() ? promise.getNow() : promise.cause()).toString());});
eventExecutors.execute(() -> { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } RuntimeException e = new RuntimeException("error..."); log.debug("set failure, {}", e.toString()); promise.setFailure(e);});
log.debug("start...");
复制代码
输出
12:04:57 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - start...12:04:58 [DEBUG] [defaultEventLoop-1-1] c.i.o.DefaultPromiseTest2 - set failure, java.lang.RuntimeException: error...12:04:58 [DEBUG] [defaultEventLoop-1-1] c.i.o.DefaultPromiseTest2 - result java.lang.RuntimeException: error...
复制代码
await 死锁检查
DefaultEventLoop eventExecutors = new DefaultEventLoop();DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
eventExecutors.submit(()->{ System.out.println("1"); try { promise.await(); // 注意不能仅捕获 InterruptedException 异常 // 否则 死锁检查抛出的 BlockingOperationException 会继续向上传播 // 而提交的任务会被包装为 PromiseTask，它的 run 方法中会 catch 所有异常然后设置为 Promise 的失败结果而不会抛出 } catch (Exception e) { e.printStackTrace(); } System.out.println("2");});eventExecutors.submit(()->{ System.out.println("3"); try { promise.await(); } catch (Exception e) { e.printStackTrace(); } System.out.println("4");});
复制代码
输出
1234io.netty.util.concurrent.BlockingOperationException: DefaultPromise@47499c2a(incomplete) at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:384) at io.netty.util.concurrent.DefaultPromise.await(DefaultPromise.java:212) at com.itcast.oio.DefaultPromiseTest.lambda$main$0(DefaultPromiseTest.java:27) at io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38) at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:73) at io.netty.channel.DefaultEventLoop.run(DefaultEventLoop.java:54) at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918) at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) at java.lang.Thread.run(Thread.java:745)io.netty.util.concurrent.BlockingOperationException: DefaultPromise@47499c2a(incomplete) at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:384) at io.netty.util.concurrent.DefaultPromise.await(DefaultPromise.java:212) at com.itcast.oio.DefaultPromiseTest.lambda$main$1(DefaultPromiseTest.java:36) at io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38) at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:73) at io.netty.channel.DefaultEventLoop.run(DefaultEventLoop.java:54) at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918) at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) at java.lang.Thread.run(Thread.java:745)
复制代码
Handler & Pipeline
ChannelHandler 用来处理 Channel 上的各种事件，分为入站、出站两种。所有 ChannelHandler 被连成一串，就是 Pipeline
打个比喻，每个 Channel 是一个产品的加工车间，Pipeline 是车间中的流水线，ChannelHandler 就是流水线上的各道工序，而后面要讲的 ByteBuf 是原材料，经过很多工序的加工：先经过一道道入站工序，再经过一道道出站工序最终变成产品
先搞清楚顺序，服务端
new ServerBootstrap() .group(new NioEventLoopGroup()) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<NioSocketChannel>() { protected void initChannel(NioSocketChannel ch) { ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { System.out.println(1); ctx.fireChannelRead(msg); // 1 } }); ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { System.out.println(2); ctx.fireChannelRead(msg); // 2 } }); ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { System.out.println(3); ctx.channel().write(msg); // 3 } }); ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){ @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { System.out.println(4); ctx.write(msg, promise); // 4 } }); ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){ @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { System.out.println(5); ctx.write(msg, promise); // 5 } }); ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){ @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { System.out.println(6); ctx.write(msg, promise); // 6 } }); } }) .bind(8080);
复制代码
客户端
new Bootstrap() .group(new NioEventLoopGroup()) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<Channel>() { @Override protected void initChannel(Channel ch) { ch.pipeline().addLast(new StringEncoder()); } }) .connect("127.0.0.1", 8080) .addListener((ChannelFutureListener) future -> { future.channel().writeAndFlush("hello,world"); });
复制代码
服务器端打印：
可以看到，ChannelInboundHandlerAdapter 是按照 addLast 的顺序执行的，而 ChannelOutboundHandlerAdapter 是按照 addLast 的逆序执行的。ChannelPipeline 的实现是一个 ChannelHandlerContext（包装了 ChannelHandler） 组成的双向链表
入站处理器中，ctx.fireChannelRead(msg) 是 调用下一个入站处理器如果注释掉 1 处代码，则仅会打印 1 如果注释掉 2 处代码，则仅会打印 1 2
3 处的 ctx.channel().write(msg) 会 从尾部开始触发 后续出站处理器的执行如果注释掉 3 处代码，则仅会打印 1 2 3
类似的，出站处理器中，ctx.write(msg, promise) 的调用也会 触发上一个出站处理器如果注释掉 6 处代码，则仅会打印 1 2 3 6
ctx.channel().write(msg) vs ctx.write(msg)都是触发出站处理器的执行 ctx.channel().write(msg) 从尾部开始查找出站处理器 ctx.write(msg) 是从当前节点找上一个出站处理器 3 处的 ctx.channel().write(msg) 如果改为 ctx.write(msg) 仅会打印 1 2 3，因为节点 3 之前没有其它出站处理器了 6 处的 ctx.write(msg, promise) 如果改为 ctx.channel().write(msg) 会打印 1 2 3 6 6 6... 因为 ctx.channel().write() 是从尾部开始查找，结果又是节点 6 自己
ByteBuf
是对字节数据的封装，是 NIO 中 ByteBuffer 的增强版
ByteBuf 优势
池化：可以重用池中 ByteBuf 实例，更节约内存，减少内存溢出的可能
读写指针分离，不需要像 ByteBuffer 一样切换读写模式
可以自动扩容
支持链式调用，使用更流畅
很多地方体现零拷贝，例如 slice、duplicate、CompositeByteBuf
1）创建
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(10);log(buffer);
复制代码
上面代码创建了一个默认的 ByteBuf（池化基于直接内存的 ByteBuf），初始容量是 10。如果没有设置值，则默认是 256。并且是可以动态扩容的
输出
read index:0 write index:0 capacity:10
复制代码
其中 log 方法参考如下
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());}
复制代码
2）直接内存 vs 堆内存
这里的直接内存与 NIO 的一致，可以使用下面的代码来创建池化基于堆的 ByteBuf
ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(10);
复制代码
也可以使用下面的代码来创建池化基于直接内存的 ByteBuf
ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(10);//ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(10);//堆外直接内存
复制代码
3）池化 vs 非池化
池化的最大意义在于可以重用 ByteBuf，优点有
池化功能是否开启，可以通过下面的系统环境变量来设置
-Dio.netty.allocator.type={unpooled|pooled}
复制代码
4）组成
ByteBuf 由四部分组成
最开始读写指针都在 0 位置
5）写入
方法列表，省略一些不重要的方法
注意
先写入 4 个字节
buffer.writeBytes(new byte[]{1, 2, 3, 4});log(buffer);
复制代码
结果是
read index:0 write index:4 capacity:10 +-------------------------------------------------+ | 0 1 2 3 4 5 6 7 8 9 a b c d e f |+--------+-------------------------------------------------+----------------+|00000000| 01 02 03 04 |.... |+--------+-------------------------------------------------+----------------+
复制代码
再写入一个 int 整数，也是 4 个字节
buffer.writeInt(5);log(buffer);
复制代码
结果是
read index:0 write index:8 capacity:10 +-------------------------------------------------+ | 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 |........ |+--------+-------------------------------------------------+----------------+
复制代码
还有一类方法是 set 开头的一系列方法，也可以写入数据，但不会改变写指针位置
未完待续
文章转载自：SevenCoder
原文链接：https://www.cnblogs.com/seven97-top/p/18697450
体验地址：http://www.jnpfsoft.com/?from=001YH
评论