Netty 使用异步线程池执行耗时任务

约定不等于承诺〃 2022-09-10 10:15 437阅读 0赞

当遇到在Handler需要执行耗时较高的操作时候，可以采用异步的方式来解决，多线程异步实现方式有两种：在Handler中添加线程池和在Context中添加线程池。任务队列并没有使用多线程，它是使用同一个线程执行IO操作和运行任务队列中的任务。

一 任务队列

  1.     这种方式运行任务队列线程和事件循环线程是同一个线程，并没有使用新的线程。
  3.    Netty的事件循环EventLoop是一个不断循环着执行读取就绪事件、处理事件、运行任务队列这三个操作的一个线程。事件循环关联了一个任务队列，用于存放耗时较长的业务处理操作。事件循环线程先读取就绪事件，如果任务队列为空则用阻塞一定时间的方式读取，如果任务队列非空，则使用非阻塞的方式读取，读取就绪事件。读取后进行处理事件，记录处理事件花费的时间，在任务队列中不断取出任务执行，默认花费在任务队列上取执行任务的时间和处理事件的时间相同，可以使用ioradio参数调整IO操作和非IO操作花费的时间比例，到时间后又去读取就绪事件。
  5. 在处理器中使用ChannelHandlerContext获取channel获取事件循环提交任务
  7. package com.tech.netty.netty.source.async;
  10. import io.netty.bootstrap.ServerBootstrap;
  11. import io.netty.channel.ChannelFuture;
  12. import io.netty.channel.ChannelFutureListener;
  13. import io.netty.channel.ChannelInitializer;
  14. import io.netty.channel.ChannelOption;
  15. import io.netty.channel.nio.NioEventLoopGroup;
  16. import io.netty.channel.socket.SocketChannel;
  17. import io.netty.channel.socket.nio.NioServerSocketChannel;
  19. import java.util.HashSet;
  20. import java.util.Set;
  22. /**
  23.  * @author lw
  24.  * @since 2021/8/11
  25.  **/
  26. public class NettyServer {
  28.     public static void main(String[] args) throws InterruptedException {
  30.         //1 创建两个线程组bossGroup workerGroup
  31.         //2 bossGroup处理客户端连接请求 workerGroup处理客户端数据读写请求
  32.         //3 他们都是事件循环组 默认含有CPU核数*2个事件循环NioEventLoop
  33.         NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
  34.         NioEventLoopGroup workerGroup = new NioEventLoopGroup();
  35.         try {
  36.             //创建服务器端启动对象，进行参数配置
  37.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  39.             serverBootstrap.group(bossGroup, workerGroup)  //设置两个线程组
  40.                     .channel(NioServerSocketChannel.class)  //设置服务器端通道为NioServerSocketChannel
  41.                     .option(ChannelOption.SO_BACKLOG, 128) //设置连接队列允许的连接个数
  42.                     .childOption(ChannelOption.SO_KEEPALIVE, true) //设置连接连接状态为活动连接
  43.                     .childHandler(new ChannelInitializer<SocketChannel>() { //给workerGroup NioEventLoop通过管道添加处理器
  44.                         @Override
  45.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
  46.                             socketChannel.pipeline().addLast(new NettyServerHandler());
  47.                         }
  48.                     });
  49.             //启动服务器并绑定一个端口
  50.             ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
  51.             System.out.println("ok");
  52.             //对通道的关闭事件进行监听
  53.             channelFuture.channel().closeFuture().sync();
  54.         } finally {
  55.             bossGroup.shutdownGracefully();
  56.             workerGroup.shutdownGracefully();
  57.         }
  58.     }
  59. }
  61. package com.tech.netty.netty.source.async;
  63. import io.netty.buffer.ByteBuf;
  64. import io.netty.buffer.Unpooled;
  65. import io.netty.channel.ChannelHandlerContext;
  66. import io.netty.channel.ChannelInboundHandlerAdapter;
  67. import io.netty.util.CharsetUtil;
  68. import lombok.extern.slf4j.Slf4j;
  70. /**
  71.  * @author lw
  72.  * @since 2021/8/11
  73.  **/
  74. //自定义一个Handler
  75. @Slf4j
  76. public class NettyServerHandler extends ChannelInboundHandlerAdapter {
  78.     @Override
  79.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
  80.         log.info("channel={} | handlerAdded",ctx.channel().id());
  81.     }
  83.     //读取客户端发送的消息
  84.     // ChannelHandlerContext 是一个上下文对象，可以获取通道和管道信息
  85.     // msg 是客户端发送的消息
  86.     @Override
  87.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  88.         log.info("服务器读取线程 "+Thread.currentThread().getName());
  89.         ByteBuf byteBuf = (ByteBuf) msg;
  90.         System.out.println("客户端发送消息是："+byteBuf.toString(CharsetUtil.UTF_8));
  92.         ctx.channel().eventLoop().execute(() -> {
  93.             try {
  94.                 log.info("第一个任务开始执行");
  95.                 Thread.sleep(5*1000);
  96.             } catch (InterruptedException e) {
  97.                 e.printStackTrace();
  98.             }
  99.             ctx.writeAndFlush(Unpooled.copiedBuffer("hello 客户端 1",CharsetUtil.UTF_8));
  100.             log.info("第一个提交任务的线程 "+Thread.currentThread().getName());
  101.         });
  103.         ctx.channel().eventLoop().execute(() -> {
  104.             try {
  105.                 log.info("第二个任务开始执行");
  106.                 Thread.sleep(5*1000);
  107.             } catch (InterruptedException e) {
  108.                 e.printStackTrace();
  109.             }
  110.             ctx.writeAndFlush(Unpooled.copiedBuffer("hello 客户端 1",CharsetUtil.UTF_8));
  111.             log.info("第二个提交任务的线程 "+Thread.currentThread().getName());
  112.         });
  114.         ctx.writeAndFlush(Unpooled.copiedBuffer("go on....",CharsetUtil.UTF_8));
  115.         log.info("go on...");
  116.     }
  118.     //数据读取完毕
  119.     @Override
  120.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  121.     }
  123.     //处理异常 一般是需要关闭通道
  124.     @Override
  125.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  126.         ctx.close();
  127.     }
  128. }
  130. package com.tech.netty.netty.source.async;
  132. import io.netty.bootstrap.Bootstrap;
  133. import io.netty.channel.ChannelFuture;
  134. import io.netty.channel.ChannelInitializer;
  135. import io.netty.channel.nio.NioEventLoopGroup;
  136. import io.netty.channel.socket.SocketChannel;
  137. import io.netty.channel.socket.nio.NioSocketChannel;
  139. /**
  140.  * @author lw
  141.  * @since 2021/8/11
  142.  **/
  143. public class NettyClient {
  144.     public static void main(String[] args) throws InterruptedException {
  145.         //客户端需要一个事件循环组
  146.         NioEventLoopGroup eventExecutors = new NioEventLoopGroup();
  147.         try {
  148.             // 创建客户端启动对象 并配置参数
  150.             Bootstrap bootstrap = new Bootstrap();
  151.             bootstrap.group(eventExecutors)  //设置线程组
  152.                     .channel(NioSocketChannel.class)  //设置客户端通道
  153.                     .handler(new ChannelInitializer<SocketChannel>() {  //设置处理器
  154.                         @Override
  155.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
  156.                             socketChannel.pipeline().addLast(new NettyClientHandler());
  157.                         }
  158.                     });
  159.             System.out.println("...客户端 is ready...");
  160.             //连接服务器
  161.             ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 7000).sync();
  162.             //对通道关闭事件进行监听
  163.             channelFuture.channel().closeFuture().sync();
  164.         } finally {
  165.             eventExecutors.shutdownGracefully();
  166.         }
  168.     }
  169. }
  171. package com.tech.netty.netty.source.async;
  173. import io.netty.buffer.ByteBuf;
  174. import io.netty.buffer.Unpooled;
  175. import io.netty.channel.ChannelHandlerContext;
  176. import io.netty.channel.ChannelInboundHandlerAdapter;
  177. import io.netty.util.CharsetUtil;
  179. /**
  180.  * @author lw
  181.  * @since 2021/8/11
  182.  **/
  183. public class NettyClientHandler extends ChannelInboundHandlerAdapter {
  185.     //当通道就绪会触发该方法 给服务器发送数据
  186.     @Override
  187.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  188.         ctx.writeAndFlush(Unpooled.copiedBuffer("hello 服务器", CharsetUtil.UTF_8));
  189.     }
  191.     //当通道有读取事件时触发
  192.     @Override
  193.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  194.         ByteBuf byteBuf = (ByteBuf) msg;
  195.         System.out.println("服务器回复的消息："+byteBuf.toString(CharsetUtil.UTF_8));
  196.     }
  198.     @Override
  199.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  200.        cause.printStackTrace();
  201.         ctx.close();
  202.     }
  203. }
  205. 通过结果可以看到
  207. handler线程和运行任务队列里任务的线程是同一个线程
  208. 任务队列里任务按序逐一取出执行

#

二 在Handler中定义线程池

通过在Handler中定义一个线程池，执行耗时较高的操作，IO线程执行IO操作，业务操作由业务线程来处理，提高IO效率，如果业务操作中需要进行IO操作，则在调用的时候Netty会判断当前线程是否为IO线程，如果不是则会则会将IO操作添加到对应事件循环的任务队列中，有事件循环的IO线程来运行。

  1. package com.tech.netty.netty.source.async;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.buffer.Unpooled;
  5. import io.netty.channel.ChannelHandlerContext;
  6. import io.netty.channel.ChannelInboundHandlerAdapter;
  7. import io.netty.util.CharsetUtil;
  8. import io.netty.util.concurrent.DefaultEventExecutor;
  9. import io.netty.util.concurrent.DefaultEventExecutorGroup;
  10. import io.netty.util.concurrent.EventExecutorGroup;
  11. import lombok.extern.slf4j.Slf4j;
  13. import java.util.concurrent.Callable;
  15. /**
  16.  * @author lw
  17.  * @since 2021/8/11
  18.  **/
  19. //自定义一个Handler
  20. @Slf4j
  21. public class NettyServerHandler extends ChannelInboundHandlerAdapter {
  23.     static final EventExecutorGroup group=new DefaultEventExecutorGroup(16);
  25.     @Override
  26.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
  27.         log.info("channel={} | handlerAdded",ctx.channel().id());
  28.     }
  30.     //读取客户端发送的消息
  31.     // ChannelHandlerContext 是一个上下文对象，可以获取通道和管道信息
  32.     // msg 是客户端发送的消息
  33.     @Override
  34.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  35. //        log.info("服务器读取线程 "+Thread.currentThread().getName());
  36. //        ByteBuf byteBuf = (ByteBuf) msg;
  37. //        log.info("客户端发送消息是："+byteBuf.toString(CharsetUtil.UTF_8));
  38. //
  39. //        ctx.channel().eventLoop().execute(() -> {
  40. //            try {
  41. //                log.info("第一个任务开始执行");
  42. //                Thread.sleep(5*1000);
  43. //            } catch (InterruptedException e) {
  44. //                e.printStackTrace();
  45. //            }
  46. //            ctx.writeAndFlush(Unpooled.copiedBuffer("hello 客户端 1",CharsetUtil.UTF_8));
  47. //            log.info("第一个提交任务的线程 "+Thread.currentThread().getName());
  48. //        });
  49. //
  50. //        ctx.channel().eventLoop().execute(() -> {
  51. //            try {
  52. //                log.info("第二个任务开始执行");
  53. //                Thread.sleep(5*1000);
  54. //            } catch (InterruptedException e) {
  55. //                e.printStackTrace();
  56. //            }
  57. //            ctx.writeAndFlush(Unpooled.copiedBuffer("hello 客户端 1",CharsetUtil.UTF_8));
  58. //            log.info("第二个提交任务的线程 "+Thread.currentThread().getName());
  59. //        });
  60. //
  61. //        ctx.writeAndFlush(Unpooled.copiedBuffer("go on....",CharsetUtil.UTF_8));
  62.         final Object msgCop=msg;
  63.         final ChannelHandlerContext ctxCop=ctx;
  64.         //在handler中添加业务线程池 执行耗时任务
  65.         group.submit(new Callable<Object>() {
  66.             @Override
  67.             public Object call() throws Exception {
  68.                 ByteBuf buff = (ByteBuf) msgCop;
  69.                 byte[] bytes = new byte[buff.readableBytes()];
  70.                 buff.readBytes(bytes);
  71.                 String s = new String(bytes, CharsetUtil.UTF_8);
  72.                 log.info("读取到消息：{}",s);
  73.                 Thread.sleep(10*1000);
  74.                 ctxCop.writeAndFlush(Unpooled.copiedBuffer("hello client",CharsetUtil.UTF_8));
  75.                 log.info("发送完成");
  76.                 return null;
  77.             }
  78.         });
  79.         group.submit(new Callable<Object>() {
  80.             @Override
  81.             public Object call() throws Exception {
  82.                 ByteBuf buff = (ByteBuf) msgCop;
  83.                 byte[] bytes = new byte[buff.readableBytes()];
  84.                 buff.readBytes(bytes);
  85.                 String s = new String(bytes, CharsetUtil.UTF_8);
  86.                 log.info("1读取到消息：{}",s);
  87.                 Thread.sleep(10*1000);
  88.                 ctxCop.writeAndFlush(Unpooled.copiedBuffer("hello client1",CharsetUtil.UTF_8));
  89.                 log.info("1发送完成");
  90.                 return null;
  91.             }
  92.         });
  93.         log.info("go on...");
  94.     }
  96.     //数据读取完毕
  97.     @Override
  98.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  99.         ctx.flush();
  100.     }
  102.     //处理异常 一般是需要关闭通道
  103.     @Override
  104.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  105.         cause.printStackTrace();
  106.         ctx.close();
  107.     }
  108. }

执行结果：

  1. IO线程打印了go on
  2. 两个业务线程同时开始读取消息，休眠10s 发送消息

实现异步执行。

#

三 在Context中指定线程池

在pipeline添加handler时，指定业务线程池，则handler中的非IO操作都由业务线程来运行，遇到IO操作将操作封装为Task提交到任务队列，由事件循环的线程来运行任务队列中的任务时执行。

  1. package com.tech.netty.netty.source.async;
  4. import io.netty.bootstrap.ServerBootstrap;
  5. import io.netty.channel.ChannelFuture;
  6. import io.netty.channel.ChannelFutureListener;
  7. import io.netty.channel.ChannelInitializer;
  8. import io.netty.channel.ChannelOption;
  9. import io.netty.channel.nio.NioEventLoopGroup;
  10. import io.netty.channel.socket.SocketChannel;
  11. import io.netty.channel.socket.nio.NioServerSocketChannel;
  12. import io.netty.util.concurrent.DefaultEventExecutorGroup;
  13. import io.netty.util.concurrent.EventExecutorGroup;
  15. import java.util.HashSet;
  16. import java.util.Set;
  18. /**
  19.  * @author lw
  20.  * @since 2021/8/11
  21.  **/
  22. public class NettyServer {
  24.     static final EventExecutorGroup group=new DefaultEventExecutorGroup(16);
  26.     public static void main(String[] args) throws InterruptedException {
  28.         //1 创建两个线程组bossGroup workerGroup
  29.         //2 bossGroup处理客户端连接请求 workerGroup处理客户端数据读写请求
  30.         //3 他们都是事件循环组 默认含有CPU核数*2个事件循环NioEventLoop
  31.         NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
  32.         NioEventLoopGroup workerGroup = new NioEventLoopGroup();
  33.         try {
  34.             //创建服务器端启动对象，进行参数配置
  35.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  37.             serverBootstrap.group(bossGroup, workerGroup)  //设置两个线程组
  38.                     .channel(NioServerSocketChannel.class)  //设置服务器端通道为NioServerSocketChannel
  39.                     .option(ChannelOption.SO_BACKLOG, 128) //设置连接队列允许的连接个数
  40.                     .childOption(ChannelOption.SO_KEEPALIVE, true) //设置连接连接状态为活动连接
  41.                     .childHandler(new ChannelInitializer<SocketChannel>() { //给workerGroup NioEventLoop通过管道添加处理器
  42.                         @Override
  43.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
  44.                             socketChannel.pipeline()
  45.                                     .addLast(group,new NettyServerHandler());
  46.                         }
  47.                     });
  48.             //启动服务器并绑定一个端口
  49.             ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
  50.             System.out.println("ok");
  51.             //对通道的关闭事件进行监听
  52.             channelFuture.channel().closeFuture().sync();
  53.         } finally {
  54.             bossGroup.shutdownGracefully();
  55.             workerGroup.shutdownGracefully();
  56.         }
  57.     }
  58. }
  60. package com.tech.netty.netty.source.async;
  62. import io.netty.buffer.ByteBuf;
  63. import io.netty.buffer.Unpooled;
  64. import io.netty.channel.ChannelHandlerContext;
  65. import io.netty.channel.ChannelInboundHandlerAdapter;
  66. import io.netty.util.CharsetUtil;
  67. import io.netty.util.concurrent.DefaultEventExecutorGroup;
  68. import io.netty.util.concurrent.EventExecutorGroup;
  69. import lombok.extern.slf4j.Slf4j;
  71. import java.util.concurrent.Callable;
  73. /**
  74.  * @author lw
  75.  * @since 2021/8/11
  76.  **/
  77. //自定义一个Handler
  78. @Slf4j
  79. public class NettyServerHandler extends ChannelInboundHandlerAdapter {
  81.     static final EventExecutorGroup group = new DefaultEventExecutorGroup(16);
  83.     @Override
  84.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
  85.         log.info("channel={} | handlerAdded", ctx.channel().id());
  86.     }
  88.     //读取客户端发送的消息
  89.     // ChannelHandlerContext 是一个上下文对象，可以获取通道和管道信息
  90.     // msg 是客户端发送的消息
  91.     @Override
  92.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  93. //        log.info("服务器读取线程 "+Thread.currentThread().getName());
  94. //        ByteBuf byteBuf = (ByteBuf) msg;
  95. //        log.info("客户端发送消息是："+byteBuf.toString(CharsetUtil.UTF_8));
  96. //
  97. //        ctx.channel().eventLoop().execute(() -> {
  98. //            try {
  99. //                log.info("第一个任务开始执行");
  100. //                Thread.sleep(5*1000);
  101. //            } catch (InterruptedException e) {
  102. //                e.printStackTrace();
  103. //            }
  104. //            ctx.writeAndFlush(Unpooled.copiedBuffer("hello 客户端 1",CharsetUtil.UTF_8));
  105. //            log.info("第一个提交任务的线程 "+Thread.currentThread().getName());
  106. //        });
  107. //
  108. //        ctx.channel().eventLoop().execute(() -> {
  109. //            try {
  110. //                log.info("第二个任务开始执行");
  111. //                Thread.sleep(5*1000);
  112. //            } catch (InterruptedException e) {
  113. //                e.printStackTrace();
  114. //            }
  115. //            ctx.writeAndFlush(Unpooled.copiedBuffer("hello 客户端 1",CharsetUtil.UTF_8));
  116. //            log.info("第二个提交任务的线程 "+Thread.currentThread().getName());
  117. //        });
  118. //
  119. //        ctx.writeAndFlush(Unpooled.copiedBuffer("go on....",CharsetUtil.UTF_8));
  120.         final Object msgCop = msg;
  121.         final ChannelHandlerContext ctxCop = ctx;
  122.         //在handler中添加业务线程池 执行耗时任务
  123. //        group.submit(new Callable<Object>() {
  124. //            @Override
  125. //            public Object call() throws Exception {
  126. //                ByteBuf buff = (ByteBuf) msgCop;
  127. //                byte[] bytes = new byte[buff.readableBytes()];
  128. //                buff.readBytes(bytes);
  129. //                String s = new String(bytes, CharsetUtil.UTF_8);
  130. //                log.info("读取到消息：{}",s);
  131. //                Thread.sleep(10*1000);
  132. //                ctxCop.writeAndFlush(Unpooled.copiedBuffer("hello client",CharsetUtil.UTF_8));
  133. //                log.info("发送完成");
  134. //                return null;
  135. //            }
  136. //        });
  137. //        group.submit(new Callable<Object>() {
  138. //            @Override
  139. //            public Object call() throws Exception {
  140. //                ByteBuf buff = (ByteBuf) msgCop;
  141. //                byte[] bytes = new byte[buff.readableBytes()];
  142. //                buff.readBytes(bytes);
  143. //                String s = new String(bytes, CharsetUtil.UTF_8);
  144. //                log.info("1读取到消息：{}",s);
  145. //                Thread.sleep(10*1000);
  146. //                ctxCop.writeAndFlush(Unpooled.copiedBuffer("hello client1",CharsetUtil.UTF_8));
  147. //                log.info("1发送完成");
  148. //                return null;
  149. //            }
  150. //        });
  152.         ByteBuf buff = (ByteBuf) msgCop;
  153.         byte[] bytes = new byte[buff.readableBytes()];
  154.         buff.readBytes(bytes);
  155.         String s = new String(bytes, CharsetUtil.UTF_8);
  156.         log.info("读取到消息：{}", s);
  157.         Thread.sleep(10 * 1000);
  158.         ctxCop.writeAndFlush(Unpooled.copiedBuffer("hello client", CharsetUtil.UTF_8));
  159.         log.info("发送完成");
  161.         log.info("go on...");
  162.     }
  164.     //数据读取完毕
  165.     @Override
  166.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  167.         ctx.flush();
  168.     }
  170.     //处理异常 一般是需要关闭通道
  171.     @Override
  172.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  173.         cause.printStackTrace();
  174.         ctx.close();
  175.     }
  176. }

这样NettyServerHandler中的操作将会使用业务线程运行，在调用IO操作时（读写数据）会提交任务队列由时间循环线程执行。

当给handler指定线程池后，会发现将任务提交到了队列

watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_20_color_FFFFFF_t_70_g_se_x_16

于此同时日志前缀输出不再是nioEventLoopGroup，而是 DefaultEventExecutorGroup，说明配置的线程池生效。

发表评论

表情：
评论列表 （有 0 条评论，437人围观）

还没有评论，来说两句吧...

相关阅读