当前位置：首页 > news >正文

Netty应用(六) 之异步 Channel

news 来源：原创 2024/5/12 9:32:54

12.Netty异步的相关概念

12.1 异步编程的概念

12.2 方式1：主线程阻塞，等待异步线程完成调用，然后主线程发起请求IO

12.3 方式2：主线程注册异步线程，异步线程去回调发起请求IO

12.4 细节注释

12.5 异步的好处

13.Netty中异步设计（内部原理）

13.1 关于Future

13.1.1 JDK中的future

13.1.2 netty中的future

13.2 关于promise

13.3 本章问题

14.Channel

14.1 netty封装之后的常见API

14.1.1 writeAndFlush(Object msg)

14.1.2 write(Object msg)

14.1.3 close

14.1.4 优雅的关闭 shutdownGracefully()

12.Netty异步的相关概念

12.1 异步编程的概念

异步编程和多线程编程

多线程编程：

多线程编程中每一个线程都是公平，平等的关系，举例子：多个客户端client请求与服务端交互，假设说服务端给每一个客户端都开启一个线程去处理，那么这多个线程是公平，平等的

异步编程：

通常是在一个主要的线程(main线程)中，异步开启一个其他线程去处理一些阻塞耗时的操作，但是当该异步开启的其他线程处理完耗时阻塞的操作后，要把最终得出的结果返回给主要的线程(main线程)。异步编程是分主次的，而不是平等的关系。

不过二者都是多线程！

异步编程有两种方式：

1、主线程阻塞，等待连接线程完成调用，然后主线程发起请求IO。

这个方式不好，因为执行下面发送IO的操作是在主线程，他阻塞等待了连接的异步线程。实际上还是在等待。

2、主线程注册异步线程，异步线程去回调发起请求IO。

这种方式是更高效的，因为主线程没有阻塞，他等着回调就完了，主线程继续往下做他的事。

在Netty中有一个原则：

如果是关于网络IO的地方，都设计为异步处理。比如上面说的connect()，网络连接方法自然是网络IO，是异步的。

还有就是我们代码中的channel.writeAndFlush("hello netty");这里再写数据就是IO，也是异步线程执行的。

所以如果你的channel.writeAndFlush("hello netty");这一行后面需要他的返回结果，也需要阻塞等待。类似这样。

Channel channel = future.channel();

ChannelFuture writeAndFlushFuture = channel.writeAndFlush("hello netty");

writeAndFlushFuture.sync();

还有就是.close()关闭socket也是异步的。关闭socket自然要做网络IO。

所以就是一旦是异步，如果你下面要处理就需要考虑阻塞等待，或者监听回调。

12.2 方式1：主线程阻塞，等待异步线程完成调用，然后主线程发起请求IO

package com.messi.netty_core_02.netty03;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.DefaultEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class MyNettyServer {private static final Logger log = LoggerFactory.getLogger(MyNettyServer.class);public static void main(String[] args) {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.channel(NioServerSocketChannel.class);serverBootstrap.group(new NioEventLoopGroup(1),new NioEventLoopGroup(8));DefaultEventLoopGroup defaultEventLoopGroup = new DefaultEventLoopGroup();serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new StringDecoder());ch.pipeline().addLast(defaultEventLoopGroup,new ChannelInboundHandlerAdapter(){@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.info("服务端读取到的数据为：{}",msg);}});}});serverBootstrap.bind(8000);}}

package com.messi.netty_core_02.netty03;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.net.InetSocketAddress;public class MyNettyClient {private static final Logger log = LoggerFactory.getLogger(MyNettyClient.class);public static void main(String[] args) throws Exception{Bootstrap bootstrap = new Bootstrap();bootstrap.channel(NioSocketChannel.class);bootstrap.group(new NioEventLoopGroup());bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder());}});ChannelFuture future = bootstrap.connect(new InetSocketAddress(8000));future.sync();Channel channel = future.channel();channel.writeAndFlush("leomessi");log.info("--------netty-sync--------");System.out.println("处理业务操作......");log.info("--------netty-sync--------");}}

测试与分析

存在性能瓶颈，影响后续业务操作，所以引出方式2

12.3 方式2：主线程注册异步线程，异步线程去回调发起请求IO

代码

package com.messi.netty_core_02.netty03;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.DefaultEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class MyNettyServer {private static final Logger log = LoggerFactory.getLogger(MyNettyServer.class);public static void main(String[] args) {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.channel(NioServerSocketChannel.class);serverBootstrap.group(new NioEventLoopGroup(1),new NioEventLoopGroup(8));DefaultEventLoopGroup defaultEventLoopGroup = new DefaultEventLoopGroup();serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new StringDecoder());ch.pipeline().addLast(defaultEventLoopGroup,new ChannelInboundHandlerAdapter(){@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.info("服务端读取到的数据为：{}",msg);}});}});serverBootstrap.bind(8000);}}

package com.messi.netty_core_02.netty03;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.net.InetSocketAddress;public class MyNettyClient {private static final Logger log = LoggerFactory.getLogger(MyNettyClient.class);public static void main(String[] args) throws Exception{Bootstrap bootstrap = new Bootstrap();bootstrap.channel(NioSocketChannel.class);bootstrap.group(new NioEventLoopGroup());bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new StringEncoder());}});ChannelFuture future = bootstrap.connect(new InetSocketAddress(8000));future.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {Channel channel = future.channel();channel.writeAndFlush("leomessi");}});log.info("--------netty-sync--------");System.out.println("处理业务操作......");log.info("--------netty-sync--------");}}

测试与分析

测试结果：

异步新线程回调注册监听的匿名内部类重写的方法：

12.4 细节注释

细节1

为什么需要future.sync()使得main线程阻塞等待到异步线程处理完网络连接操作后才能让main线程继续执行？

因为网络连接操作是后续一切网络IO的基础，如果你网络连接这一io都没做好，那么你后续read，write这些网络io你怎么做？你没法保证，所以就像方式1那种main线程阻塞等待的方式，其实就是为了保证网络连接正确建立，避免后续其他网络io出问题。

但是由于main线程同步阻塞的方式太低效，所以才引出了后续注册监听，异步回调的方式来优化性能！也就是方式2。但是有一点仍然需要注意，我们要把网络io操作(read或write)放在异步线程处理完连接后所回调的回调方法中！为什么？还问为什么吗，见上即可，就是为了保证read，write这些网络io操作的执行是建立在正确无误完成网络连接之后的呀。对于一些无关网络io的业务操作，main线程由于不阻塞了，所以也可以执行。这样是不是把这个同步阻塞的范围从阻塞"整个main线程"缩小到只阻塞"必须网络连接完成后才能执行read，write等网络io"了。【代码详细见方式1和方式2】

12.5 异步的好处

1.提高系统的吞吐量

分析：这个肯定，可以在相同时间内处理更多的客户端操作或业务操作

2.效率上的提高

分析：原来需要t ms处理完的操作，异步后可能只需要t/2 ms。但绝对不是1+1=2，1+1<2，即两个线程达到的性能高度绝对不是理论上两个线程的性能水平，因为线程间通信，切换都需要耗费性能。

13.Netty中异步设计（内部原理）

先给一个结论：

jdk的future只支持阻塞同步式的future

netty的future是在jdk的future的基础上做的扩展，支持阻塞同步式的future，也支持异步监听处理式的future

netty的Promise是在netty的future的基础上做的扩展，支持阻塞同步式的future，也支持异步监听处理式的future，同样支持得到异步监听的结果是正确还是失败的(这一点前面二者都是不支持的)

我们之前用的bootstrap.connect(new InetSocketAddress(8000));返回的就是ChannelFuture这个异步化的支持方式。

而在netty中还实现了Promise这个异步支持，他的功能是最全的，所以他用的比较多。而且实际上我们上面的代码中，ChannelFuture future = bootstrap.connect(new InetSocketAddress(8000));

这个ChannelFuture 其实也是promise的类型，可以断点验证一下：可以看到Future实际上是一个promise类型的内部类。Promise是netty中最底层的继承类，所以他最强大，所以自然也就用的多一些。

13.1 关于Future

我们上面看来netty中的future就是继承自JDK中的future，并且对他做了扩展。下面我们分别来看看他们的区别和差异

13.1.1 JDK中的future

测试：

13.1.2 netty中的future

等待阻塞方式

测试：

监听的方式

13.2 关于promise

后续编码Promise用的不多，Handler用的多，但是Netty底层Promise用的多

上面我们看到了netty中的future提供了异步监听的方式来实现异步处理，我们在future中异步执行了callable中的call方法，最后返回结果的时候回调了listener中的回调方法，把future当成参数传给了回调方法，我们在回调方法中就能通过future.get()获取到call中的执行结果了。

于是问题来了：

1.但是如果我们编程的时候用的不是callable，而是使用的Runnable，runnable是没有返回值的，你这时候这个操作就无法得到返回值了。于是这个当前的架构就不行了。

而我们说promise是继承自future(netty)的，它的功能更加强大了，它实际上就能解决这个Runnable没有返回值的问题，因为promise可以进行设置结果是成功还是失败。

2.有些人会说既然你Runnable不可以有返回值，那么使用Callable问题不就解决了吗？但是你应该清楚，即便我们使用了Callable接口，我们在call中返回的值在实际开发中可能是一个code码值，或者是一个调用结果。你如何知道这个返回结果是正常成功的还是失败的呢？你总的要告诉承接你这个结果的地方是正常的还是失败的吧。也就是你要正常和异常都要给出一个结果返回，你的异步到底执行的如何了。这个需要返回让外部其他线程(main线程)知道。

而我的主线程和你异步线程交互的东西就是异步的返回值，那么这时候就得知道这个返回。所以你不仅仅要把处理的数据作为返回，还要把处理的结果也返回。你也可以包一层比如一个类Result里面有data还有code码，你封装起来返回。但是这就引入新的类了，我们希望返回的数据就是数据，结果就是结果。于是netty为我们封装了一个新的类型就是Promise。

promise不仅仅告诉你结果返回的是什么，还告诉你结果是成功的还是失败的

promise的异步阻塞操作

测试：

promise的异步监听操作

测试：

总结

Promise让无返回值的Runnable可以设置返回值并且设置返回值是正确还是失败的

13.3 本章问题

关于监听器的执行顺序问题

DefaultEventLoopGroup defaultEventLoopGroup = new DefaultEventLoopGroup(2);
EventLoop eventLoop = defaultEventLoopGroup.next();
// 1
Future<String> future = eventLoop.submit(new Callable<String>() {@Overridepublic String call() throws Exception {logger.debug("call begin ***************");TimeUnit.SECONDS.sleep(2);logger.debug("thread is {}", Thread.currentThread().getName());return "hello netty";}
});
// 2
future.addListener(new GenericFutureListener<Future<? super String>>() {@Overridepublic void operationComplete(Future<? super String> future) throws Exception{logger.debug("get call result is {}", future.get());}
})

之前我们有这么一段代码，我们说eventLoop提交了异步任务，异步线程去执行，然后下面的future添加监听器来监听，等到异步任务执行结束了，就回调这个监听器里面的operationComplete方法。既然是异步的多线程必然就有CPU调度问题，假如在1处的异步代码执行完了，2处的代码还没被调度，也就是异步执行完了，监听器还没注册进去，那你的监听器是不是就无法监听到异步结果了呢？？？？意思就是你异步执行完了的回调是要回调谁呢？源码中利用兜底操作解决这一CPU调度问题。

来看源码：(future的实际执行类型是promise，也就是我们直接去DefaultPromise去查看这个addListener的方法)

// io.netty.util.concurrent.DefaultPromise#addListener
public Promise<V> addListener(GenericFutureListener<? extends Future<? super V>>
listener) {checkNotNull(listener, "listener");// 这里是添加监听器的地方synchronized (this) {addListener0(listener);}// 这里是这个问题的重点，当异步线程执行完成后 isDone()返回true。否则返回false。
//如果返回true且你已经完成监听，那么回调operationComplete方法
//如果返回true但是你没监听，也就是你没有来得及监听，异步结果就执行完了，这里也会再次执行一遍监听操作，添加进去然后进行回调//所以这其实是个兜底操作。if (isDone()) {notifyListeners();}return this;
}

WebFlux

WebFlux Spring响应式框架。

WebFlux百分之80都是Netty的API，其余扩展都是在Handler上做的扩展。

14.Channel

1.首先netty中的channel是对原来jdk中NIO下面的ServerSocketChannel和SocketChannel的封装

2.这个封装最终体现出来的有两个好处：

（1）统一了channel的编程模型，客户无需再区分ServerSocketChannel还是SocketChannel，都是bootstrap，channel

（2）提供了更加强大和丰富的功能，各种编解码处理器，TCP Socket缓冲区大小的设计。滑动窗口大小的设计。你像你之前使用NIO编程，你怎么设置这些操作系统级别的参数？设置不了的。但是netty这里可以，后续都会一一讲到这些API

这些增强的功能也就是我们需要关注的那些API实现

14.1 netty封装之后的常见API

14.1.1 writeAndFlush(Object msg)

以前我们在NIO的时候，写出数据用的是NIO的write(ByteBuffer byteBuffer)，你要写一个字符串出去，还得把字符串转为ByteBuffer作为参数传递进去，才能写出去。现在Netty里面的writeAndFlush(Object msg)里面接收的是一个Object对象。你不用区分任何的类型，直接写就行了。

而且这个方法你一看名字就知道，写入并且刷新，意思就是写入到操作系统内核的Socket缓冲区后立马刷新通过网络发出。所以当你使用这个方法发数据时，服务端立马能接收到的。

但是注意：netty中网络IO都会异步开启一个新线程(NioEventLoop)来执行

14.1.2 write(Object msg)

这个方法也是接收的Object，不用转为ByteBuffer类型就能发送，不用问你在网络传输不能直接传字符串，肯定是要转的，netty其实也就是封装了NIO，所以他底层肯定给你转成了ByteBuffer，只是不用你做这个操作了，参数接收的更统一了。

但是这个方法就只是写出了，写到了操作系统socket缓冲区里面，除非缓冲区写满了，不然他不会写完就发出数据，你需要接一步，channel.flush()才能把数据及时刷出去。

14.1.3 close

以前我们刚开始写netty代码的时候是如下这样：

// 客户端连接端口到服务端，连接到了返回的ChannelFuture是个异步操作。，这里是新的线程处理的连接

ChannelFuture connect = bootstrap.connect(new InetSocketAddress(8000));

// 在这里阻塞，等到连接上了，成功了在返回，然后往下走

connect.sync();

// 连接上了，在这个连接获取通道，也就是当初NIO的SC。此时这里是main线程

Channel channel = connect.channel();

// 往服务端写数据，往出写

channel.writeAndFlush("hello netty");

channel发送完了数据之后就不管了，实际上这里之后程序还没有结束，因为启动了一些socket程序，是后台线程，导致这个程序不会结束。

但是实际上我们开发可能会要结束这个程序，就需要channel.close();加一句。

这句代码会帮你关闭一些后台资源。

但是这个时候有个需求，我们在执行完close之后要执行一些业务逻辑，这时候我们第一反应可能就是这样，有两种方案：1.同步阻塞等待close关闭完成 2.监听

方案1

public class MyNettyClient {private static final Logger log = LoggerFactory.getLogger(MyNettyClient.class);public static void main(String[] args) throws InterruptedException {Bootstrap bootstrap = new Bootstrap() ;NioEventLoopGroup group = new NioEventLoopGroup();bootstrap.group(group);bootstrap.channel(NioSocketChannel.class);bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new LoggingHandler());ch.pipeline().addLast(new StringEncoder());}});ChannelFuture future = bootstrap.connect(new InetSocketAddress(8000));future.sync();Channel channel = future.channel();channel.writeAndFlush("leomessi");System.out.println("MyNettyClient.main");ChannelFuture close = channel.close();close.sync();log.info("当channel关闭后，执行一些业务逻辑.....");}
}

方案2：

引入LoggingHandler

测试

其实同步阻塞和监听都无所谓，测试一个即可，这里以监听方案进行测试：

测试存在的问题和细节

你debug测试的时候，把断点处改为Thread，这样断点调试阻塞的是当前main线程，而不是所有线程。因为netty针对所有网络IO都是异步开启新线程的方式进行处理，你要是设置为ALL阻塞所有线程，你异步开启的新线程是不是也阻塞住了，那么你怎么write写出，怎么flush刷新给出去呢？？？？都阻塞了，啥都别想了。。

引入LoggingHandler打印输出日志，使用Grep这一功能

选中然后右键

看到如下日志：

14.1.4 优雅的关闭 shutdownGracefully()

我们可以看到一个现象：

我们看到日志显示事件和通道实际上被关闭了。但是左边的红点显示程序还没被关闭。这个不难理解，我们上面用的是bootstrap.group(new NioEventLoopGroup());，这个NioEventLoopGroup被创建出来，其实是个线程池，这个线程池还活着呢，所以还有后台线程在，你这个程序结束不掉。netty认为直接结束不安全，万一除了通道事件还有别的线程在做任务，这样直接全结束不安全。

补充

因为我们new NioEventLoopGroup()，按照netty以及本计算机核数来计算的话，创建出32个线程。虽然说NioEventLoopGroup的线程是被用作网络IO做异步线程开启。但是假设说如果使用得当，当我们处理普通任务时，也可以使用NioEventLoopGroup中的线程去做异步（只需要把任务提交给NioEventLoop即可）。

当channel.close()调用后，close方法只占用一个NioEventLoop线程，其他31个线程就算不都运行着，其中可能还有很多线程是有未执行完的任务的，所以如果channel.close()后就关闭整个client的所有线程，那么太不地道，太绝户了。所以使用的是eventLoopGroup.shutdownGracefully()，优雅的关闭，当所有的线程(32个线程)都结束处理之后，才会真正关闭client

所以引入优雅的关闭：eventLoopGroup.shutdownGracefully()

测试：

那么问题来了，如果我们不关闭通道，直接优雅关闭会咋样。试一下：

package com.messi.netty_core_02.netty04;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.net.InetSocketAddress;public class MyNettyClient {private static final Logger log = LoggerFactory.getLogger(MyNettyClient.class);public static void main(String[] args) throws InterruptedException {Bootstrap bootstrap = new Bootstrap() ;NioEventLoopGroup group = new NioEventLoopGroup();bootstrap.group(group);bootstrap.channel(NioSocketChannel.class);bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new LoggingHandler());ch.pipeline().addLast(new StringEncoder());}});ChannelFuture future = bootstrap.connect(new InetSocketAddress(8000));future.sync();Channel channel = future.channel();channel.writeAndFlush("leomessi");System.out.println("MyNettyClient.main");//        ChannelFuture close = channel.close();//        close.sync();
//
//        log.info("当channel关闭后，执行一些业务逻辑.....");
//        close.addListener(new GenericFutureListener<Future<? super Void>>() {
//            @Override
//            public void operationComplete(Future<? super Void> future) throws Exception {
//                log.info("当channel关闭后，执行一些业务逻辑.....");
//            }
//        });
//        //优雅的关闭group.shutdownGracefully();}}

测试结果：

这个优雅的关闭同样会close关闭channel，但是不会打印输出，可能底层源码走的逻辑不一样，但是肯定是把channel关闭了，因为状态已经显示为INACTIVE。为什么会这样？因为对于channel的操作，比如说连接或关闭channel通道，其实都是NioEventLoopGroup其中一个线程去做的，那么优雅的关闭是关闭该线程组中的所有，只不过是等待他们都做完自己的线程任务再优雅的关闭，最终都会关闭。

实际就是结束的客户端，既然是优雅关闭，他关闭的实际就是等待EventLoopGroup里面所有线程的任务都结束才完成关闭，实际上这就是优雅关闭的定义，后面好好研究一下优雅关闭(kill -15和kill -9的故事，以及jdk中对于Linux关闭事件的捕获实现)。

注释：kiil -9 +进程号A ：强制关闭进程号A所对应的进程，即使你进程A中还有很多线程没有执行完毕，也得强制关闭掉。