包含自定义任务、定时任务等方法，当异步任务较多时需要添加将任务到线程池，通常有添加到Handler和Context两种方式

自定义任务：

当客户端的Channel与EventLoopGroup绑定后，在通道没有关闭之前，对应通道都会一直使用当初绑定的线程进行业务操作
image.png
Channel和EventLoop线程绑定关系

自定义普通任务：

在服务端读取客户端通道数据时往往需要做业务操作，当操作较为耗时时需要使用异步方式进行处理（异步任务执行的线程依旧是使用服务端ServerBootstrap定义的workGroup，只是操作是异步的），在添加多个异步任务时需要等待
对服务端自定义Handler的channelRead方法进行修改即可

代码：

  1. package simple;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.buffer.Unpooled;
  5. import io.netty.channel.ChannelHandlerContext;
  6. import io.netty.channel.ChannelInboundHandlerAdapter;
  8. import java.nio.charset.StandardCharsets;
  9. import java.util.concurrent.TimeUnit;
  12. /** 自定义管道 Handler
  13.  *  1、自定义一个 Handler，需要继承 Netty 规定好的某个 HandlerAddapter
  14.  *  2、这时我们自定义一个 Handler，才能称为一个 Handler
  15.  */
  16. public class NettyServerHandler extends ChannelInboundHandlerAdapter {
  19.     /*当有读取事件时该方法将被触发
  20.      *  参数一: ChannelHandlerContext 上下文对象，含有管道 PipeLine、通道
  21.      *  参数二: Object 客户端发送的数据，默认是Object需要转换
  22.      */
  23.     @Override
  24.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  26.         /* 用户自定义普通任务 : 使用异步处理 > 找到客户端 Channel 对应的 NioEventLoop 的 TaskQueue 中
  27.          *  用于处理耗时非常长的业务处理方式
  28.          */
  30.         ctx.channel().eventLoop().execute(()->{
  31.             try {
  32.                 System.out.println("ExecuteThread1 is: "+Thread.currentThread().getName());
  33.                 TimeUnit.SECONDS.sleep(3);
  34.                 ctx.writeAndFlush(Unpooled.copiedBuffer("3秒的异步任务已完成!",StandardCharsets.UTF_8));
  35.             } catch (InterruptedException e) {
  36.                 e.printStackTrace();
  37.             }
  38.         });
  40.         ctx.channel().eventLoop().execute(()->{
  41.             try {
  42.                 System.out.println("ExecuteThread2 is: "+Thread.currentThread().getName());
  43.                 TimeUnit.SECONDS.sleep(10);
  44.                 ctx.writeAndFlush(Unpooled.copiedBuffer("10秒的异步任务已完成!",StandardCharsets.UTF_8));
  45.             } catch (InterruptedException e) {
  46.                 e.printStackTrace();
  47.             }
  48.         });
  50.         System.out.println("异步任务开始处理!"+LocalDateTime.now()+" 当前线程是: "+Thread.currentThread().getName());
  52.     }
  55.     /* 数据读取完毕后触发
  56.      *   参数一: ChannelHandlerContext 上下文对象，含有管道 PipeLine、通道
  57.      */
  58.     @Override
  59.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  61.         //对发送数据进行编码后写入到缓存并刷新
  62.         ctx.writeAndFlush(Unpooled.copiedBuffer("服务端连接成功,欢迎!",StandardCharsets.UTF_8));
  64.     }
  67.     /* 处理异常，一般为关闭通道 */
  68.     @Override
  69.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  70.         ctx.close();        //和 ctx.channel().close() 是一个意思
  71.     }
  75. }

测试：

image.png
服务端使用异步方法轮流处理任务

自定义定时任务：

指定延时时间运行任务，定时任务被添加到 ScheduledTaskQueue 中，优先级比普通任务的 TaskQueue 低（异步任务执行的线程依旧是使用服务端ServerBootstrap定义的workGroup，只是操作是异步的）
定时时间以添加到队列中为开始，当之前存在普通任务时，存在普通任务执行结束后直接执行定时任务的情况，因此需要设置好定时时间
对服务端自定义Handler的channelRead方法进行修改即可

代码：

  1. package simple;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.buffer.Unpooled;
  5. import io.netty.channel.ChannelHandlerContext;
  6. import io.netty.channel.ChannelInboundHandlerAdapter;
  8. import java.nio.charset.StandardCharsets;
  9. import java.time.LocalDateTime;
  10. import java.util.concurrent.TimeUnit;
  13. /** 自定义管道 Handler
  14.  *  1、自定义一个 Handler，需要继承 Netty 规定好的某个 HandlerAddapter
  15.  *  2、这时我们自定义一个 Handler，才能称为一个 Handler
  16.  */
  17. public class NettyServerHandler extends ChannelInboundHandlerAdapter {
  19.     /*当有读取事件时该方法将被触发
  20.      *  参数一: ChannelHandlerContext 上下文对象，含有管道 PipeLine、通道
  21.      *  参数二: Object 客户端发送的数据，默认是Object需要转换
  22.      */
  23.     @Override
  24.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  27.         ctx.channel().eventLoop().execute(()->{
  28.             try {
  29.                 System.out.println("ExecuteThread2 is: "+Thread.currentThread().getName());
  30.                 TimeUnit.SECONDS.sleep(10);
  31.                 ctx.writeAndFlush(Unpooled.copiedBuffer("10秒的异步任务已完成!"+LocalDateTime.now(),StandardCharsets.UTF_8));
  32.             } catch (InterruptedException e) {
  33.                 e.printStackTrace();
  34.             }
  35.         });
  37.         System.out.println("异步任务开始处理!"+LocalDateTime.now()+" 当前线程是: "+Thread.currentThread().getName());
  39.         /* 用户自定义定时任务,任务被添加到 NioEventLoop 的 ScheduledTaskQueue 中 */
  40.         ctx.channel().eventLoop().schedule(()->{
  41.             System.out.println("ExecuteThread3 is: "+Thread.currentThread().getName());
  42.             ctx.writeAndFlush(Unpooled.copiedBuffer("自定义定时任务已完成!"+LocalDateTime.now(),StandardCharsets.UTF_8));
  43.         },5,TimeUnit.SECONDS);
  46.     }
  49.     /* 数据读取完毕后触发
  50.      *   参数一: ChannelHandlerContext 上下文对象，含有管道 PipeLine、通道
  51.      */
  52.     @Override
  53.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  55.         //对发送数据进行编码后写入到缓存并刷新
  56.         ctx.writeAndFlush(Unpooled.copiedBuffer("服务端连接成功,欢迎!"+LocalDateTime.now(),StandardCharsets.UTF_8));
  58.     }
  61.     /* 处理异常，一般为关闭通道 */
  62.     @Override
  63.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  64.         ctx.close();        //和 ctx.channel().close() 是一个意思
  65.     }
  69. }

测试：

image.png
定时任务队列排在普通任务后

线程池处理异步任务：

当客户端的Channel与EventLoopGroup和执行异步任务的DefaultEventLoopGroup绑定后，在通道没有关闭之前都会一直使用当初绑定的线程进行业务操作
image.png
Channel与EventLoop中线程绑定关系

Handler方式：

在handler创建线程组进行异步提交，异步方法使用的是单独创建的EventExecutorGroup内的线程而不是之前服务端中ServerBootstrap定义的workGroup，具体可参考源码中的 execute1
直接执行的方法使用的是workGroup线程，异步提交使用的是线程组线程
源码为 AbstractChannelHandlerContext 中的 write 方法

服务端Handler：

  1. public class NettyServerHandler extends ChannelInboundHandlerAdapter {
  4.     static final EventExecutorGroup group = new DefaultEventExecutorGroup(16);  //充当业务线程池
  7.     @Override
  8.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  11.         group.submit(()->{
  12.             try {
  13.                 System.out.println("ExecuteThread1 is: "+Thread.currentThread().getName());
  14.                 TimeUnit.SECONDS.sleep(3);
  16.                 /* 堆内存方式创建ByteBuf*/
  17.                 //ctx.writeAndFlush(Unpooled.copiedBuffer("3秒的异步任务已完成!"+ LocalDateTime.now(),StandardCharsets.UTF_8));
  19.                 /* 直接内存方式创建ButeBuf*/
  20.                 ByteBuf buffer = ctx.alloc().buffer();
  21.                 buffer.writeBytes(  ("3秒的异步任务已完成!"+LocalDateTime.now()).getBytes(StandardCharsets.UTF_8) );
  22.                 ctx.writeAndFlush(buffer);   
  24.             } catch (InterruptedException e) {
  25.                 e.printStackTrace();
  26.             }
  27.         });
  29.         group.submit(()->{
  30.             try {
  31.                 System.out.println("ExecuteThread2 is: "+Thread.currentThread().getName());
  32.                 TimeUnit.SECONDS.sleep(10);
  33.                 ctx.writeAndFlush(Unpooled.copiedBuffer("10秒的异步任务已完成!"+ LocalDateTime.now(),StandardCharsets.UTF_8));
  34.             } catch (InterruptedException e) {
  35.                 e.printStackTrace();
  36.             }
  37.         });
  39.         group.submit(()->{
  40.             try {
  41.                 System.out.println("ExecuteThread3 is: "+Thread.currentThread().getName());
  42.                 TimeUnit.SECONDS.sleep(12);
  43.                 ctx.writeAndFlush(Unpooled.copiedBuffer("12秒的异步任务已完成!"+ LocalDateTime.now(),StandardCharsets.UTF_8));
  44.             } catch (InterruptedException e) {
  45.                 e.printStackTrace();
  46.             }
  47.         });
  50.         System.out.println("异步任务开始处理!"+LocalDateTime.now()+" 当前线程是: "+Thread.currentThread().getName());
  51.         ctx.writeAndFlush(Unpooled.copiedBuffer("服务端开始异步任务开始处理~~~", CharsetUtil.UTF_8));
  53.     }
  56.     /* 数据读取完毕后触发
  57.      *   参数一: ChannelHandlerContext 上下文对象，含有管道 PipeLine、通道
  58.      */
  59.     @Override
  60.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  62.         //对发送数据进行编码后写入到缓存并刷新
  63.         ctx.writeAndFlush(Unpooled.copiedBuffer("服务端连接成功,欢迎!"+LocalDateTime.now(),StandardCharsets.UTF_8));
  65.     }
  68.     /* 处理异常，一般为关闭通道 */
  69.     @Override
  70.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  71.         ctx.close();
  72.     }
  76. }

测试：
异步任务使用内部线程组调用
image.png

Context方式：

将handler添加到pipeline的时候指定线程组即可，具体可参考源码中的 execute2
使用提交方法用的是workGroup线程执行任务，直接执行的是异步任务

服务端：

  1. package simple.execute2;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.buffer.Unpooled;
  5. import io.netty.channel.*;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioServerSocketChannel;
  9. import io.netty.util.concurrent.DefaultEventExecutorGroup;
  10. import io.netty.util.concurrent.EventExecutorGroup;
  12. import java.nio.charset.StandardCharsets;
  13. import java.time.LocalDateTime;
  14. import java.util.concurrent.TimeUnit;
  17. /* 异步任务添加到Handler版服务端代码 */
  18. public class NettyServer {
  21.     static final EventExecutorGroup group1 = new DefaultEventExecutorGroup(2);       //创建业务1线程池
  22.     static final EventExecutorGroup group2 = new DefaultEventExecutorGroup(2);       //创建业务2线程池
  24.     public static void main(String[] args) throws InterruptedException {
  26.         /* 创建 BossGroup 和 WorkerGroup 线程组
  27.          *   BossGroup 只处理连接请求，真正的和客户端业务处理会交给 WorkGrouop 完成
  28.          *     bossGroup 和 workerGroup 含有的子线程的个数默认为 CPU最大线程数 X 2，可以手动指定
  29.          */
  30.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);        //创建子线程的个数为1的 bossGroup
  31.         EventLoopGroup workerGroup = new NioEventLoopGroup(8);      //创建子线程的个数为8的 workerGroup
  32.         try {
  33.             //创建服务器端的启动对象，配置参数
  34.             ServerBootstrap bootstrap = new ServerBootstrap();
  36.             //使用链式参数配置启动参数
  37.             bootstrap.group(bossGroup,workerGroup)                          //设置两个线程组
  38.                     .channel(NioServerSocketChannel.class)                  //使用 NioSocketChannel 作为服务器通道实现
  39.                     .option(ChannelOption.SO_BACKLOG,120)             //设置线程队列等待连接的个数
  40.                     .childOption(ChannelOption.SO_KEEPALIVE,true)     //保持活动连接状态
  41.                     .childHandler(new ChannelInitializer<SocketChannel>() { //创建一个通道初始化对象
  43.                         //给PipeLine设置处理器
  44.                         @Override
  45.                         protected void initChannel(SocketChannel ch) throws Exception {
  46.                             ChannelPipeline pipeline = ch.pipeline();
  47.                             pipeline.addLast(group1,new ChannelInboundHandlerAdapter(){
  48.                                 @Override
  49.                                 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  52.                                     /* 使用的是workerGroup的线程 */
  53.                                     ctx.channel().eventLoop().execute(()->{
  54.                                         try {
  55.                                             System.out.println("ExecuteThread1 is: "+Thread.currentThread().getName());
  56.                                             TimeUnit.SECONDS.sleep(3);
  57.                                             ctx.writeAndFlush(Unpooled.copiedBuffer("耗时3秒的主要任务已完成!"+ LocalDateTime.now(),StandardCharsets.UTF_8));
  58.                                         } catch (InterruptedException e) {
  59.                                             e.printStackTrace();
  60.                                         }
  61.                                     });
  63.                                     /* 非execute执行的使用的是自定义的group线程组线程 */
  64.                                     System.out.println("异步任务开始处理!"+LocalDateTime.now()+" 当前线程是: "+Thread.currentThread().getName());
  65.                                     TimeUnit.SECONDS.sleep(10);
  66.                                     ctx.writeAndFlush(Unpooled.copiedBuffer("10秒的异步任务已完成!"+ LocalDateTime.now(),StandardCharsets.UTF_8));
  68.                                 }
  69.                             });
  71.                             pipeline.addLast(group2,new ChannelInboundHandlerAdapter(){
  72.                                 @Override
  73.                                 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  74.                                     //对发送数据进行编码后写入到缓存并刷新
  75.                                     ctx.writeAndFlush(Unpooled.copiedBuffer("服务端连接成功,欢迎!"+ LocalDateTime.now()+"当前线程是: "+Thread.currentThread().getName(), StandardCharsets.UTF_8));
  76.                                 }
  77.                             });
  78.                         }
  80.                     });
  82.             System.out.println("服务器 is ready .....");
  84.             //启动服务器绑定一个端口并且同步，生成一个 ChannelFuture 对象
  85.             ChannelFuture cf = bootstrap.bind(8889).sync();
  87.             //给CF注册监听器，监控关心的事件
  88.             cf.addListener(new ChannelFutureListener() {
  89.                 @Override
  90.                 public void operationComplete(ChannelFuture channelFuture) throws Exception {
  91.                     if(cf.isSuccess()){
  92.                         System.out.println("监听端口成功");
  93.                     }else{
  94.                         System.out.println("监听端口失败");
  95.                     }
  96.                 }
  97.             });
  99.             //对关闭通道进行监听（当有关闭通道的消息时才进行监听）
  100.             cf.channel().closeFuture().sync();
  102.         } finally {
  103.             bossGroup.shutdownGracefully();     //关闭资源
  104.             workerGroup.shutdownGracefully();   //关闭
  105.         }
  108.     }
  111. }

测试：
异步任务执行成功
image.png

优缺点对比：

第一种方式在handler中添加异步，可能更加的自由，比如如果需要访问数据库，那就异步，如果不需要，就不异步，异步会拖长接口啊应时间。因为需要将任务放进mpscTask 中。如果I0时间很短，Task 很多，可能一个循环下来，都没时间执行整个task,导致响应时间达不到指标
第二种方式是Netty标准方式(即加入到队列)，不同的事件（业务）操作可以指定不同的线程池，但是会将整个handler都交给业务线程池，不论耗时不耗时，都加入到队列里，不够灵活

源码实现：

由 AbstractChannelHandlerContext 的 invokeChannelRead 对具体调用的线程进行判断
image.png