netty的IO模型

Posted 2021-04-25 IWAN学编程

上一篇文章中, 我介绍了学习netty的前置知识, 即Java NIO. <  > 本文我们将介绍NIO的三种模型以及netty的改良措施. 

1.单Reactor单线程模式

单Reactor单线程的特点之一就是编程简单, 不涉及到多线程, 线程间的通信以及锁竞争的问题. 所有的工作全部由一个线程完成. 当然缺点也是显而易见的, 如今的电脑CPU都是多核的, 单线程的模式显然无法充分利用cpu性能. 并且单线程非常可能会进入死循环导致整个系统不可用.

请求解析: 

1. 当客户端有请求时, Reactor通过select监听客户端的请求事件, 然后由dispatch进行事件分发;

2. 如果是建立请求连接, 则由Acceptor建立连接, 然后创建一个Handler进行业务处理;

3. 如果不是建立请求连接, 则由Reactor分配对应的Handler进行处理.

2.单Reactor多线程模式

单Reactor多线程模型相比单Reactor单线程可以较大程度利用CPU的性能, 但是多线程的编程模型较为复杂. 在大并发的场景下, 单线程的事件监听和响应会出现性能瓶颈.

请求解析:

1. 当客户端有请求时, Reactor通过select监听客户端的请求事件, 然后由dispatch进行事件分发;

2. 如果是建立请求连接, 则由Acceptor建立连接, 然后创建一个Handler进行业务处理;

3. 如果不是建立请求连接, 则由Reactor分配对应的Handler进行处理.

4. 在此模型中, handler不负责事件处理, 而是把业务事件分配给对应线程处理. 

5. worker线程完成处理后,把结果返回给handler. 

3.主 从Reactor多线程模式

基于以上两种模型, 主从Reactor多线程模型在并发量上有极大改进. 这种模型在实际中有很多应用, 比如反向代理服务器nginx, 缓存服务器memcached, 还有本文所说的netty. 不过缺点也很明显, 就是学习难度大, 编程复杂. 

请求解析: 

1. Reactor主线程通过select监听连接事件, 然后通过Acceptor处理连接事件.

2. Acceptor处理连接事件之后, 主Reactor把连接分配给从Reactor.

3. 从Reactor创建handler进行各种事件处理.

4. handler通过read读取数据, 分配给后面的worker线程处理.

5. woker线程池完成业务处理后, 返回处理结果.

通过上述对三种reactor模型的介绍, 我们对Java NIO的三种编程模型已经较为了解. Netty是基于NIO主从Reactor模型做的进一步改进. 下面是Netty的IO模式图.

4. Netty的IO模型

请求解析:

1.Netty有两组线程池, BossGroup和WorkerGroup, Boss负责客户端的 连接, Wo rker负责网络读写. 它们的类型都是NioEventLoopGroup. 它包括一个selector和TaskQueue. 

2. BossGroup 处理accept事件, 与客户端连接, 然后生成NiosocketChannel, 并将其注册到worker NIOEventLoop上的selector. 并处理任务队列的任务.

3. WorkerGroup 处理read/write事件, 在对应的NioSocketChannel, 并处理任务队列的任务.

Netty代码示例(discard)

package com.iwan.netty.discard;
import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelPipeline;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;import io.netty.handler.logging.LogLevel;import io.netty.handler.logging.LoggingHandler;
/** * 服务端 */public final class DiscardServer {
 static final boolean SSL = System.getProperty("ssl") != null; static final int PORT = Integer.parseInt(System.getProperty("port", "8009"));
 public static void main(String[] args) throws Exception {

 EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) { ChannelPipeline p = ch.pipeline(); p.addLast(new DiscardServerHandler()); } });

 ChannelFuture f = b.bind(PORT).sync();
 f.channel().closeFuture().sync(); } finally { workerGroup.shutdownGracefully(); bossGroup.shutdownGracefully(); } }}

package com.iwan.netty.discard;
import io.netty.channel.ChannelHandlerContext;import io.netty.channel.SimpleChannelInboundHandler;
/** * 自定义的handler */public class DiscardServerHandler extends SimpleChannelInboundHandler<Object> {
 @Override public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { /** * 什么也不做 */ }
 @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { cause.printStackTrace(); ctx.close(); }}

package com.iwan.netty.discard;
import io.netty.bootstrap.Bootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelPipeline;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioSocketChannel;import io.netty.handler.ssl.SslContext;import io.netty.handler.ssl.SslContextBuilder;import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
/** * 客户端 */public final class DiscardClient {
 static final String HOST = System.getProperty("host", "127.0.0.1"); static final int PORT = Integer.parseInt(System.getProperty("port", "8009"));    static final int SIZE = Integer.parseInt(System.getProperty("size", "256"));    public static void main(String[] args) throws Exception { EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(group) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline p = ch.pipeline();
 p.addLast(new DiscardClientHandler()); } });
 ChannelFuture f = b.connect(HOST, PORT).sync();
 f.channel().closeFuture().sync(); } finally { group.shutdownGracefully(); } }}

package com.iwan.netty.discard;
import io.netty.buffer.ByteBuf;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelFutureListener;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.SimpleChannelInboundHandler;
/** * 自定义handler */public class DiscardClientHandler extends SimpleChannelInboundHandler<Object> {
 private ByteBuf content; private ChannelHandlerContext ctx;
 @Override public void channelActive(ChannelHandlerContext ctx) { this.ctx = ctx; content = ctx.alloc().directBuffer(DiscardClient.SIZE).writeZero(DiscardClient.SIZE); generateTraffic(); }
 @Override public void channelInactive(ChannelHandlerContext ctx) { content.release(); }
 @Override public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { }
 @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { cause.printStackTrace(); ctx.close(); }
 long counter;
 private void generateTraffic() { ctx.writeAndFlush(content.retainedDuplicate()).addListener(trafficGenerator); }
 private final ChannelFutureListener trafficGenerator = future -> { if (future.isSuccess()) { generateTraffic(); } else { future.cause().printStackTrace(); future.channel().close(); } };}

本文的内容较多, 并理解复杂. 认真阅读肯定会有收获, 明天的内容为使用netty写一个http服务器.