Netty学习小记

向右看齐 2022-06-04 00:50 208阅读 0赞

Netty提供了方便快速的API用于搭建网络，  
一下是我学习Netty后的一些小笔记

Netty的线程模型  
Netty使用React线程模型  
Netty服务端Demo代码的初始化如下

public final class SecureChatServer {
        
    
        static final int PORT = Integer.parseInt(System.getProperty("port", "8992"));
    
        public static void main(String[] args) throws Exception {
            SelfSignedCertificate ssc = new SelfSignedCertificate();
            SslContext sslCtx = SslContext.newServerContext(ssc.certificate(), ssc.privateKey());
    
            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 SecureChatServerInitializer(sslCtx));
    
                b.bind(PORT).sync().channel().closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

Accept 线程：前一个bossGroup线程是Accept线程，用于接收建立连接的请求  
IO线程：workerGroup线程是IO线程  
逻辑线程：用于执行业务逻辑代码

NioEventLoop类包含了java.nio的selector操作，具体的IO操作代码在NioEventLoop中  
ServerBootstrap仅仅是配置参数类

SecureChatServerInitializer是ChannelInitializer实现类，用于socket接入进来之后的处理组件顺序，也就处理逻辑  
这里，我看看到ChannelPipeline对于加载进来的SocketChannel，做了一些组件处理

public class SecureChatServerInitializer extends ChannelInitializer<SocketChannel> {
        
    
        private final SslContext sslCtx;
    
        private final EventExecutorGroup logicThreadGroup ;
    
        public SecureChatServerInitializer(SslContext sslCtx, EventExecutorGroup logicThreadGroup) {
            this.sslCtx = sslCtx;
            this.logicThreadGroup = logicThreadGroup;
        }
    
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
    
            // Add SSL handler first to encrypt and decrypt everything.
            // In this example, we use a bogus certificate in the server side
            // and accept any invalid certificates in the client side.
            // You will need something more complicated to identify both
            // and server in the real world.
            pipeline.addLast(sslCtx.newHandler(ch.alloc()));
    
            // On top of the SSL handler, add the text line codec.
            pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
            pipeline.addLast(new StringDecoder());
            pipeline.addLast(new StringEncoder());
    
            // and then business logic.
            pipeline.addLast(logicThreadGroup, new SecureChatServerHandler());
        }
    }

pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));

对于每个socketChannel，ChannelHandler都是不共享的(也就是每一个SocketChannel都是享有独自的handler，除非  
ChannelHandler使用了注解 @Sharable 标记，不然ChannelHandler是不共享的);  
该组件的执行线程是IO线程也就是我们上面所说的workGroup线程

与之相反，如果我们需要使用非IO线程处理的花，我们使用如下的形式：  
pipeline.addLast(logicThreadGroup, new SecureChatServerHandler());  
对于一个逻辑线程处理的组件，他的执行者是固定的，一旦初始化后指定了执行线程后，  
执行该逻辑组件的线程就与该组件绑定起来，而不会改变，执行者ChannelHandlerInvoker是EventExecutor线程的一个封装。

每一个加入进来的handle都会绑定到一个ChannelHandlerContext(AbstractChannelHandlerContext)中去，  
在AbstractChannelHandlerContext源码中可以看到其实现了双向链表，顺序与ChannelPipeline中ChannelHandler  
的顺序是一致的。

ChannelHandlerContext提供了write和read等操作,但是read和write操作总是从IO线程最先发起的

@Override
        public void invokeRead(final ChannelHandlerContext ctx) {
            if (executor.inEventLoop()) {
                invokeReadNow(ctx);
            } else {
                AbstractChannelHandlerContext dctx = (AbstractChannelHandlerContext) ctx;
                Runnable task = dctx.invokeReadTask;
                if (task == null) {
                    dctx.invokeReadTask = task = new Runnable() {
                        @Override
                        public void run() {
                            invokeReadNow(ctx);
                        }
                    };
                }
                executor.execute(task);
            }
        }

@Override
        public void invokeWrite(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
            if (msg == null) {
                throw new NullPointerException("msg");
            }
            if (!validatePromise(ctx, promise, true)) {
                // promise cancelled
                ReferenceCountUtil.release(msg);
                return;
            }
    
            if (executor.inEventLoop()) {
                invokeWriteNow(ctx, msg, promise);
            } else {
                AbstractChannel channel = (AbstractChannel) ctx.channel();
                int size = channel.estimatorHandle().size(msg);
                if (size > 0) {
                    ChannelOutboundBuffer buffer = channel.unsafe().outboundBuffer();
                    // Check for null as it may be set to null if the channel is closed already
                    if (buffer != null) {
                        buffer.incrementPendingOutboundBytes(size);
                    }
                }
                safeExecuteOutbound(WriteTask.newInstance(ctx, msg, size, promise), promise, msg);
            }
        }

ChannelFuture，ChannelPromise这两个类提供了异步操作功能

static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

DefaultChannelGroup用于保存链接到服务器上的所有链接信息，相当于Netty提供的工具类

Netty提供了IdleStateHandler用来检测心跳，检测远程端是否存活

// An example that sends a ping message when there is no outbound traffic
     // for 30 seconds.  The connection is closed when there is no inbound traffic
     // for 60 seconds.
    
     public class MyChannelInitializer extends ChannelInitializer<Channel> {
        
          @Override
         public void initChannel(Channel channel) {
             channel.pipeline().addLast("idleStateHandler", new IdleStateHandler(60, 30, 0));
             channel.pipeline().addLast("myHandler", new MyHandler());
         }
     }

// Handler should handle the IdleStateEvent triggered by IdleStateHandler.
     public class MyHandler extends ChannelHandlerAdapter {
        
          @Override
         public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
             if (evt instanceof IdleStateEvent) {
                 IdleStateEvent e = (IdleStateEvent) evt;
                 if (e.state() == IdleState.READER_IDLE) {
                     ctx.close();
                 } else if (e.state() == IdleState.WRITER_IDLE) {
                     ctx.writeAndFlush(new PingMessage());
                 }
             }
         }
     }

IdleStateHandler将设计的超时的事件传递到MyHandler，我们可以在MyHandler中做超时判断，或者发送心跳包