Netty如何使用异步编程
关于异步编程, JDK提供了Future接口, 但是此接口存在以下问题 :
- A线程发起异步请求, A线程调用get()方法会被阻塞, 虽然提供了超时的get(timeout)但也无法满足要求.
- 针对异步请求调用返回成功或异常或被取消三种情况都认为是完成, 不能细化这三种情况.
于是Netty提供了自己的Future接口
我们来看下Netty是如何实现异步编程
我们先看下简单的测试用例代码
服务端部分代码如下
serverBootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) {
ChannelPipeline channelPipeline = ch.pipeline();
channelPipeline.addLast(new StringEncoder());
channelPipeline.addLast(new StringDecoder());
channelPipeline.addLast(new MyInHandler());
}
});MyInHandler部分代码如下
@Override
public void channelActive(ChannelHandlerContext ctx) {
String content = "From Netty Server...";
ChannelFuture channelFuture = ctx.writeAndFlush(content);
channelFuture.addListener((future) -> {
log.info("yyy");
});
log.info("xxx");
}当我们通过客户端连接到服务端之后, 服务端会先打印yyy 再打印xxx .这样并没有达到异步编程的效果.
我们上述的代码想达到的效果是: 在调用完writeAndFlush方法向客户端写数据后, 数据未必及时写出去， 但也不要阻塞当前线程, 线程依然可以继续向下'走', 等数据写出去之后, 再来回到之前添加的监听. 所以打印顺序应该是先xxx再yyy
那么为什么打印结果与我们期望的不同呢? 我们来分析下.
根据目前的代码结构, 执行channelActive方法内代码的线程是IO线程, 如果读过我之前的文章的小伙伴, 应该知道我说的这个IO线程是什么意思. IO线程执行完writeAndFlush方法后, 数据已经写入到TCP缓冲区, 虽然接下来设置了监听, 但是既然数据已经写入成功了, 那么就直接执行监听, 也就是直接打印了yyy , 接着就打印了xxx .
那么我们改动一下我们的服务端代码结构
EventLoopGroup businessGroup = new NioEventLoopGroup(8);
serverBootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) {
ChannelPipeline channelPipeline = ch.pipeline();
channelPipeline.addLast(new StringEncoder());
channelPipeline.addLast(new StringDecoder());
channelPipeline.addLast(businessGroup, new MyInHandler());
}
});我们只是改变了一个地方channelPipeline.addLast(businessGroup, new MyInHandler())
再来测试一次. 得到我们想要的结果: 先打印xxx 再打印yyy
改动之后, 执行channelActive方法的线程(姑且叫A线程)不再是IO线程, 而是businessGroup中的某个线程. 此时就会存在两个线程. A线程只会把被写的数据放在IO线程对应的taskQueue中就返回了, 添加了一个监听后就打印了xxx . 等IO线程写完数据后执行监听, 但实际上监听中的代码依然是A线程执行, 最后就打印了yyy .
上面的说法并没有错, 但依然不严谨. 代码改动前与改动后不同点就是多了一个A线程, 它负责执行channelActive方法. A线程只会把数据写入到IO线程对应的taskQueue, 具体的写操作必须且只能由IO线程来完成. A线程只要添加一个监听就可以, 等IO线程完成写之后来调用监听就可以, 但具体执行监听里面的内容依然是A线程来完成. 假如IO线程执行的比较快, 还没等A线程添加监听, IO线程就把数据写成功了, 那么A线程在执行添加监听的代码时, 判断数据已经写成功, 那么就直接执行监听里面的内容, 即打印yyy 最后再打印xxx .
归根结底就是不要阻塞A线程, 通过这样添加监听的方式可以做到不阻塞A线程.
腾讯云开发者
