一、TaskQueue 任务队列

Netty主要基于主从Reactor多线程模型（如图）做了一定的改进，其中主从Reactor多线程模型有多个Reactor

1.1 为什么需要TaskQueue?

在PipeLine中会进行业务处理，当处理比较复杂的业务时需要大量的时间，这时候就会造成长时间的阻塞，所以需要将这些复杂的业务加入到taskQueue中进行异步处理，而不会影响到其他任务的正常进行，即不会造成长时间的阻塞

1.2 任务队列中的Task有三种典型使用场景

• 用户程序自定义的普通任务
• 用户自定义定时任务
• 非当前Reactor线程调用Channel的各种方法

例如： 在推送系统的业务流程里面，根据用户的标识，找到对应的Channel引用，然后调用Write类方法向该用户推送消息，就会进入到这种场景。最终的Write会提交到任务队列中后被异步消费

1.3 常规阻塞式实例代码演示

1.3.1 服务端NettyServer代码实现

package com.netty.taskQueue;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;import java.net.ServerSocket;public class NettyServer {public static void main(String[] args) throws InterruptedException {//创建线程池对象NioEventLoopGroup bossGroup = new NioEventLoopGroup();NioEventLoopGroup workGroup = new NioEventLoopGroup();//创建配置启动类ServerBootstrap serverBootstrap = new ServerBootstrap();try {//添加相关配置serverBootstrap.group(bossGroup, workGroup)//添加线程池组.channel(NioServerSocketChannel.class)//.option(ChannelOption.SO_BACKLOG, 128)//.childOption(ChannelOption.SO_KEEPALIVE, true)//.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new NettyServerHandler());}});//绑定端口ChannelFuture cf = serverBootstrap.bind(6668).sync();//监听通道关闭cf.channel().closeFuture().sync();}finally {bossGroup.shutdownGracefully();workGroup.shutdownGracefully();}}
}

1.3.2 服务端NettyServerHandler代码

package com.netty.taskQueue;import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;public class NettyServerHandler extends ChannelInboundHandlerAdapter {//读取客户端的数据@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println("这里会阻塞10秒......");Thread.sleep(10*1000);ctx.writeAndFlush(Unpooled.copiedBuffer("10秒后.....第二条消息回复......",CharsetUtil.UTF_8));System.out.println("来自客户端：" + ctx.channel().remoteAddress() + "  ：" + ((ByteBuf) msg).toString(CharsetUtil.UTF_8));}//回复消息给客户端@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.writeAndFlush(Unpooled.copiedBuffer("第一条消息回复",CharsetUtil.UTF_8));}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.channel().close();cause.printStackTrace();}
}

1.3.3 客户端NettyClient代码实现

package com.netty.taskQueue;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;public class NettyClient {public static void main(String[] args) throws InterruptedException {NioEventLoopGroup group = new NioEventLoopGroup();Bootstrap bootstrap = new Bootstrap();try {bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new NettyClientHandler());}});ChannelFuture cf = bootstrap.connect("localhost", 6668).sync();cf.channel().closeFuture().sync();}finally {group.shutdownGracefully();}}
}

1.3.4 客户端NettyClientHandler代码实现

package com.netty.simple;import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;public class NettyClientHandler extends ChannelInboundHandlerAdapter {//当通道就绪就会触发该方法@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println("client" + ctx);ctx.writeAndFlush(Unpooled.copiedBuffer("hello,Server:小沙弥", CharsetUtil.UTF_8));}//当通道有读取事件是，会触发@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf = (ByteBuf) msg;System.out.println("服务器回复的消息：" + buf.toString(CharsetUtil.UTF_8));System.out.println("服务器地址是："+ctx.channel().remoteAddress());}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}

结果💥

可以看到第一条回复 和第二条回复阻塞10秒后一块返回了，但是我们不想让第一条消息也阻塞，那么就需要用到下面的解决方案了，将回复2加入到任务队列中，进行异步处理

1.4 将任务加入任务队列进行对比

只需将上面的代码改为即可

结果💥

可以看到第一条消息正常回复，没有阻塞，10秒后第二条消息回复，阻塞了10秒，并没有影响到第一条消息

再加入一个任务

//加入第二个处理任务ctx.channel().eventLoop().execute(new Runnable() {@Overridepublic void run() {try {//这里会在第一个任务阻塞10秒后，再次阻塞10秒Thread.sleep(10*1000);ctx.channel().writeAndFlush(Unpooled.copiedBuffer("10秒后...回复第三条消息",CharsetUtil.UTF_8));} catch (InterruptedException e) {e.printStackTrace();}}});

结果💥

1.5 自定义定时任务

将上方代码修改为即可：

结果💥

1.6 非当前Reactor线程调用Channel的各种方法

例如：在推送系统的业务线程中，可以根据用户的标识，找到对应的Channel引用，然后调用在Write类方法向该用户推送消息，就会进入到这种场景，最终的Write会提交到任务队列中后被异步消费

在上方代码修改为即可：

结果💥

总结

• Netty 抽象出两组线程池，BossGroup专门负责接收客户端连接，WorkGroup专门负责网络读写操作
• NioEventLoop表示一个不断循环执行处理任务的线程，每个NIOEventLoop都有一个Selector用于监听绑定在其上的socket网络通道
• NioEventLoop内部采用串行化设计，从消息的读取-》解码-》处理-》编码-》发送，始终由IO线程NioEventLoop负责

NioEventLoopGroup下包含多个NioEventLoop
每个NioEventLoop中包含有一个Selector,一个taskQueue
每个NioEventLoop的selector上注册监听多个NioChannel
每个NioChannel只会绑定在唯一的NioEventLoop上
每个NioChannel都绑定有一个自己的ChannelPipeline

💥推荐阅读💥

上一篇：五、Netty核心技术之线程模型讲解

下一篇：五、Netty核心技术之线程模型讲解