NIO中的异步—ChannelFuture、CloseFuture以及异步提升在NIO中的应用

ChannelFuture

客户端调用connect后返回值为ChannelFuture对象，我们可以利用ChannelFuture中的channel()方法获取到Channel对象。

由于上述代为为客户端实现，若想启动客户端实现连接操作，必须编写服务端代码，实现如下：

package netty.simpleNetty.channel;import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;/*** 服务端配合ClientChannelFuture验证Channel关闭的异步以及回调函数实现*/
@Slf4j
public class ServerChannelFuture {public static void main(String[] args) throws InterruptedException {new ServerBootstrap().group(new NioEventLoopGroup(1), new NioEventLoopGroup(2)).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buffer = msg instanceof ByteBuf ? ((ByteBuf) msg) : null;if(buffer != null){byte[] bytes = new byte[16];ByteBuf len = buffer.readBytes(bytes, 0, buffer.readableBytes());log.info(new String(bytes));}}});}}).bind(8000).sync();}
}

随后运行上述客户端代码，观察到日志输出如下：

通过此输出日志发现借用channelFuture对象的channel方法获取的Channel对象并没有创建。

原因分析：connect方法是异步的，意味着不等待连接建立，方法执行就返回了。因此channelFuture对象中不能【立刻】获取到正确的Channel对象。

解决方法：

使用sync方法让异步操作同步等待连接建立。
使用回调方法，当connect连接建立后主动调用回调函数。

sync让异步操作同步

获取到channelFuture对象后，调用sync()方法，同步等待连接的结束。sync阻塞住当前线程，直到Nio线程连接建立完毕 。先启动服务端再启动客户端日志输出如下：

此时调用channelFuture.channle()方法获取到Channel，随后可以利用获取到的channel向服务端发送消息。

使用回调函数

调用获取的channelFuture对象的addListener()方法向其添加回调函数。

可以将上述代码用Lambda表达式简化如下：

服务端整体代码实现如下：

package netty.simpleNetty.channel;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;/*** 客户端关闭channel异步以及回调函数实现*/
@Slf4j
public class ClientChannelFuture {public static void main(String[] args) throws UnknownHostException, InterruptedException {ChannelFuture channelFuture = new Bootstrap().group(new NioEventLoopGroup(2)).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {}}).connect(new InetSocketAddress(InetAddress.getLocalHost(), 8000));log.info("Before sync : {}",channelFuture.channel());
//        channelFuture.sync();
//        log.info("After sync : {}", channelFuture.channel());channelFuture.addListener((ChannelFutureListener)future -> {log.info("After listen : {}", future.channel());});
//        channelFuture.addListener(new ChannelFutureListener() {
//            @Override
//            public void operationComplete(ChannelFuture future) throws Exception {
//                log.info("After listen : {}", future.channel());
//            }
//        });}
}

CloseFuture

上述操作探讨了调用connect方法建立连接后，根据chanelFuture获取到Channel对象的sync以及回调方法实现。在获取到Channel对象后，执行对应的writeAndFlush()方法发送消息后，需要及时关闭channel。

channel关闭操作的实现有sync同步等待以及添加回调的实现方法：代码如下：

package netty.simpleNetty.channel;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
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.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.Scanner;/*** Test CloseFuture的异步以及回调实现*/
@Slf4j
public class ClientCloseFuture {public static void main(String[] args) throws InterruptedException, UnknownHostException {Channel channel = new Bootstrap().group(new NioEventLoopGroup(2)).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new LoggingHandler(LogLevel.INFO));ch.pipeline().addLast(new StringEncoder());}}).connect(new InetSocketAddress(InetAddress.getLocalHost(), 8000)).sync().channel();log.info("channel : {}", channel);new Thread(()->{Scanner scanner = new Scanner(System.in);while (true){String line = scanner.nextLine();if("q".equals(line)){// channel的close也是一个异步操作,本线程不会执行close操作,而是交由一个其他线程执行close操作channel.close();break;}channel.writeAndFlush(line);}}).start();ChannelFuture closeFuture = channel.closeFuture();log.info("waiting close...");closeFuture.sync();log.info("closed...");channel.closeFuture().addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {log.info("closed...");}});//        channel.closeFuture().addListener((ChannelFutureListener) channelFuture1 ->{
//                    log.info("closed...");
//        });}
}

异步提升

针对上述connect以及close操作，真正执行对应方法是在一个新的线程中执行的，而不是在调用connect的线程中执行连接操作。

为什么不在一个线程中去执行建立连接、去执行关闭 channel【建立连接connet，关闭channel都是在EventLoopGroup中执行的。也即nio中执行的】，那样不是也可以吗？非要用这么复杂的异步方式：比如一个线程发起建立连接，另一个线程去真正建立连接。

因为 netty 异步方式用了多线程、多线程就效率高。（理解错误的）

netty中的异步指的是建立连接connet、读取数据read、关闭channel等不再一个线程中。而是由专门的线程（handler、EventLoop）负责执行专门的操作。

思考下面的场景，4 个医生给人看病，每个病人花费 20 分钟，而且医生看病的过程中是以病人为单位的，一个病人看完了，才能看下一个病人。假设病人源源不断地来，可以计算一下 4 个医生一天工作 8 小时，处理的病人总数是：4 * 8 * 3 = 96。