netty基础学习

2020-06-20

字数统计: 4.5k字 | 阅读时长≈ 20分

回顾NIO编程

NIO编程需要掌握Selector，需要掌握channel，buffer

多路复用器Selector有很多Channel注册信息和每个Channel对应感兴趣的事件operation，用SelectionKey代表

selector调用select()方法，阻塞获取有IO信息并且是感兴趣的 SelectionKey

对SelectionKey进行处理，获取key的channel，通过read读取数据，write写出数据，此时就会用到buffer，这个buffer有他自己的数据结构，类似数组，它的读写指针都是position，可以设置不可读写区域的索引(capacity-limit),也可以标记，用来回读或回写

Server
    //这是一个简单的服务器
    public class ServerDemo
{
    public static void main(String[] args) throws IOException {

        ServerSocketChannel serverSocketChannel=ServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress(8000));
        serverSocketChannel.configureBlocking(false);//配置该channel的模式

        Selector selector = Selector.open();

        serverSocketChannel.register(selector,serverSocketChannel.validOps());//注册到selector上

        boolean flag=true;
        while (flag){
            selector.select();//阻塞，至少有一个channel被多路复用选择器选上
            Set<SelectionKey> selectionKeys = selector.selectedKeys();//获取有IO信息的channel
            Iterator<SelectionKey> iterator = selectionKeys.iterator();//获取集合的迭代器
            ByteBuffer buffer = ByteBuffer.allocate(4096);
            while (iterator.hasNext()){
                SelectionKey key = iterator.next();
                iterator.remove();

                if (key.isAcceptable()){
                    ServerSocketChannel server = (ServerSocketChannel)key.channel();
                    SocketChannel client = server.accept();
                    client.configureBlocking(false);
                    client.register(selector,SelectionKey.OP_READ);
                    System.out.println("接待");
                }
                if (key.isReadable()){
                    System.out.println("客户端发送数据过来了");
                    buffer.clear();//避免以前误操作。指针位置不对。先重置

                    SocketChannel client =(SocketChannel) key.channel();
                    //读操作，buffer有多大，就从IO流中读取多少。IO流中没读完会继续存着。等着下次读
                    int read = client.read(buffer);//注意此时指针 position 位置已经变了
                    if (read==-1) {
                        key.cancel();
                        System.out.println("客户端正常断开");

                    }else {
                        byte[] bytes = new byte[buffer.position()];

                        buffer.flip();  //用limit记录当前读指针位置,定好可读区域的，然后把读指针position 置0

                        buffer.get(bytes);//注意指针 position 位置已经变了

                        System.out.println("-----------");
                        System.out.println(new String(bytes, 0, buffer.position()));

                        buffer.clear();//把所有指针重置
                    }
                }

            }
        }


    }
}




Client
    一个简单的客户端
    public class ClientDemo {
    public static void main(String[] args) throws IOException {
        SocketChannel client = SocketChannel.open(new InetSocketAddress(8000));
        client.configureBlocking(false);
        Selector selector = Selector.open();
        client.register(selector,client.validOps());
        ByteBuffer readBuffer = ByteBuffer.allocate(4096);
        ByteBuffer writeBuffer = ByteBuffer.allocate(4096);


        AtomicBoolean flag= new AtomicBoolean(true);
        new Thread(()->{
            while (flag.get()) {
                try {
                    selector.select(selectionKey -> {
                        if (selectionKey.isReadable()) {
                            try {
                                System.out.println("服务端发送数据过来了");
                                readBuffer.clear();//
                                //读操作，buffer有多大，就从IO流中读取多少。IO流中没读完会继续存着。等着下次读
                                int read = client.read(readBuffer);  //注意此时指针 position 位置已经变了
                                if (read == -1) {
                                    selectionKey.cancel();
                                    System.out.println("与服务器正常断开");
                                } else {
                                    byte[] bytes = new byte[readBuffer.position()];
                                    readBuffer.flip();  //用limit记录当前读指针位置,定好可读区域的，然后把读指针position 置0
                                    readBuffer.get(bytes);//注意指针 position 位置已经变了
                                    System.out.println("-----------");
                                    System.out.println(new String(bytes, 0, readBuffer.position()));
                                }
                            } catch (IOException e) {
                                e.printStackTrace();
                            } finally {
                                readBuffer.clear();//把所有指针重置
                            }
                        }
                        });
                }catch (Exception ex){
                    flag.set(false);
                };
            }
        }).start();

      //把下面看作为一个读操作整体
        writeBuffer.clear(); //防止以前操作buffer指针位置不当，clear清除的不是真正的byte数据，而是移动指针到指定初始位置
        writeBuffer.put("hello world".getBytes());//注意：position变了
        //注意。如果以前这个buffer有数据，并且数据量比本次大，那么一定要flip，用limit记录当前 放最后一个字节数据进去的position
        //然后position置0，mark=-1，最后使用channel的write写出去
        writeBuffer.flip();
        client.write(writeBuffer);
        writeBuffer.clear(); //考虑以后的操作
    }
}

总结：

NIO原生API编程，需要了解网络编程的一系列知识，需要时刻盯着buffer的position和limit，因为你可能一不小心就会把不需要写出去或者读进来的字节给写了出去或者读了进来，坑爹吧，然后还要手动管理channel 连接，发数据或者读数据，关闭的流程，可能还需要去自己编码，解码一个其他的协议，等等，还有很多需要考虑的地方

NIO对于新手编程来说非常不友好，所以netty这个NIO框架诞生了，我们只需要搞懂API，其他的安全性管理，buffer指针，编码解码啥的，不需要管，交给netty

Netty

入门首先就是得会用

开源得netty再github上给你使用的例子 https://github.com/netty/netty/tree/4.1/example

官方使用指南 https://netty.io/wiki/user-guide-for-4.x.html

//编写一个收到任何数据不做任何响应的服务器，这是最简单的一个服务


根据这个程序。提取关键类
public class DiscardServerHandler extends ChannelInboundHandlerAdapter { // (1)

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) { // (2)
        // Discard the received data silently.         
        ((ByteBuf) msg).release(); // (3)
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }
}

public class DiscardServer {
    
    private int port;
    
    public DiscardServer(int port) {
        this.port = port;
    }
    
    public void run() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1) 
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap(); // (2) 
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class) // (3)   
.childHandler(new ChannelInitializer<SocketChannel>() { // (4)
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ch.pipeline().addLast(new DiscardServerHandler());
                 }
             })
             .option(ChannelOption.SO_BACKLOG, 128)          // (5)
             .childOption(ChannelOption.SO_KEEPALIVE, true); // (6)
    
            // Bind and start to accept incoming connections.
            ChannelFuture f = b.bind(port).sync(); // (7)      
    
            // Wait until the server socket is closed.
            // In this example, this does not happen, but you can do that to gracefully
            // shut down your server.
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
    
    public static void main(String[] args) throws Exception {
        int port = 8080;
        if (args.length > 0) {
            port = Integer.parseInt(args[0]);
        }

        new DiscardServer(port).run();
    }
}

编程

其实我们对netty的编程其实就是对ChannelHandler的编程，ChannelHandler的方法都由netty的EventLoop的线程进行回调

例如

NioEventLoop的线程的run方法主要的活



public void execute(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }

        boolean inEventLoop = inEventLoop();
        if (inEventLoop) {
            addTask(task);
        } else {
            startThread();
            addTask(task);
            if (isShutdown() && removeTask(task)) {
                reject();
            }
        }

        if (!addTaskWakesUp && wakesUpForTask(task)) {
            wakeup(inEventLoop);
        }
    }

protected void run() {
        for (;;) {
            try {
                switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                    case SelectStrategy.CONTINUE: 重试
                        continue;
                    case SelectStrategy.SELECT:   
                        select(wakenUp.getAndSet(false));
					阻塞获取事件，或者任务
            
       //'wakenUp.compareAndSet(false, true)'总是在调用之前计算'selector.wakeup()' 以减少唤醒开销。(Selector.wakeup() 是一项昂贵（需要尽量少调用）的操作。）
                        //
    //然而，在这种方法中有一个竞争条件。当“wakenUp”过早设置为true时，将触发竞争条件。
                        // 如果“wakenUp”设置为true太早：
                        // 1) Selector is waken up between 'wakenUp.set(false)' and
                        //    'selector.select(...)'. (BAD)
                        // 2) Selector is waken up between 'selector.select(...)' and
                        //    'if (wakenUp.get()) { ... }'. (OK)
                        //
                       // 在第一种情况下，“wakenUp”设置为true，下面是选择器.select（…）'会立刻醒来。直到在下一轮中“wakenUp”再次设置为false, 'wakenUp.compareAndSet(false, true)' 将失败，因此唤醒选择器的任何尝试也将失败，导致以下情况 'wakenUp.compareAndSet(false, true)'调用 造成不必要地阻塞。
                        //
                        //要解决此问题，如果调用selector.select(...)之后wakenUp立即为true，我们将再次唤醒选择器

                        if (wakenUp.get()) {
                            selector.wakeup();
                        }
                    default:
                        // fallthrough
                }

                cancelledKeys = 0;
                needsToSelectAgain = false;
                final int ioRatio = this.ioRatio;
                if (ioRatio == 100) {
                    try {
                        processSelectedKeys();
                    } finally {
                     
                        runAllTasks();//也会获取延时队列到时间的任务
                    }
                } else {
                    final long ioStartTime = System.nanoTime();
                    try {
                        processSelectedKeys();
                    } finally {
                        // Ensure we always run tasks.
                        final long ioTime = System.nanoTime() - ioStartTime;
                        runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                    }
                }
            } catch (Throwable t) {
                handleLoopException(t);
            }
            //即使循环处理引发异常，也始终处理关闭。
            try {
                if (isShuttingDown()) {
                    closeAll();
                    if (confirmShutdown()) {
                        return;
                    }
                }
            } catch (Throwable t) {
                handleLoopException(t);
            }
        }
    }


看下来其实最主要的就是干了两件事 
     processSelectedKeys();//处理 IO事件，回调ChannelHandler方法
     runAllTasks();  //这里一般就是用户线程提交的任务，或者 EventLoop线程执行 executorlate方法提交的任务

什么是IO线程，就是EventLoop有channel注册 ,EventLoop里面的线程都叫IO线程
    
    需要注意的是 ，如果IO线程回调 ChannelHandler方法 就千万不能执行死循环，阻塞太久或者执行太久， 
    一般自定义 的ChannelHandler不推荐放到IO线程执行。所以一般复杂的业务线程都需要自己开一个独立的线程处理
     这里也不推荐 把复杂的用户任务放入IO线程的任务队列，EventLoop的任务队列尽量是netty内部的任务
        
      
        每一个channel对应一个EventLoop, 一个EventLoop对应多个Channel
        意思就是每个Channel都有 evenLoop引用，该Channel发生的所有事件都由这个evenLoop进行执行
        
      EventLoopGroup的作用 是 在 当一个Channel注册时，分配一个EventLoop 让Channel注册上去

两个重要的ChannelHandler

netty的ChannelInboundHandler

//{@link ChannelHandler}为Channel状态更改添加的回调方法。这允许用户轻松地知道 到状态更改。
public interface ChannelInboundHandler extends ChannelHandler {

   //{@link ChannelHandlerContext}的{@link Channel}已注册到其{@link EventLoop}，然后回调此方法
    void channelRegistered(ChannelHandlerContext ctx) throws Exception;

  //{@link ChannelHandlerContext}的{@link Channel}已从其{@link EventLoop}中注销，然后回调此方法
    void channelUnregistered(ChannelHandlerContext ctx) throws Exception;

    
    //{@link ChannelHandlerContext}的{@link Channel}现在处于活动状态，然后回调此方法
    //这个活动状态 如果是ServerSocket，那么就bind成功，就是活动状态
    //如果是Socket，那就是connect成功，就是活动状态
    void channelActive(ChannelHandlerContext ctx) throws Exception;


//已注册的{@link ChannelHandlerContext}的{@link Channel}现在处于非活动状态，并已到达其生存期的末尾。
    void channelInactive(ChannelHandlerContext ctx) throws Exception;

  //当前{@link Channel}已从远程机 读取到消息时调用。
    void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception;


  //当前读取操作读取的最后一条消息已被{@link#channelRead（ChannelHandlerContext，Object）}调用。如果{@link ChannelOption#AUTO_READ}处于关闭状态，则在调用{@link ChannelHandlerContext#READ（）之前，不会再尝试从当前{@link Channel}读取入站数据。
    /*
    if (!config.isAutoRead() && !isReadPending()) {
                    removeReadOp();
                }
    */
    
    //一般都不会把 ChannelOption#AUTO_READ设置成false，这个方法在消息读完后调用
    //如果isAutoRead()时false，那么需要发送一个read(ChannelHandlerContext ctx)的outbound操作，才能进行读取一次IO数据
    void channelReadComplete(ChannelHandlerContext ctx) throws Exception;

   //在触发用户事件时调用。
    void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception;


//在{@link Channel}的可写状态更改后调用。您可以使用{@link Channel#isWritable（）}检查channel状态。
    void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception;

   //如果抛出{@link Throwable}，则调用
    @Override
    @SuppressWarnings("deprecated")
    void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;
}

netty的ChannelOutboundHandler

//{@link ChannelHandler}将收到IO出站操作的通知。
public interface ChannelOutboundHandler extends ChannelHandler {
    /**
     * 用于 执行绑定操作
     * @param ctx         为其进行绑定操作的{@link ChannelHandlerContext}
     * @param localAddress  应该绑定到的{@link SocketAddress}
     * @param promise       操作完成后通知的{@link ChannelPromise}
     * @throws Exception    thrown if an error accour
     */
    void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception;

    /**
     *用于 执行连接操作
     * @param ctx               为其执行连接操作的{@link ChannelHandlerContext}
     * @param remoteAddress     应该连接到的{@link SocketAddress}
     * @param localAddress      连接时用作源的{@link SocketAddress}
     * @param promise           操作完成后通知的{@link ChannelPromise}
     * @throws Exception        thrown if an error accour
     */
    void connect(
            ChannelHandlerContext ctx, SocketAddress remoteAddress,
            SocketAddress localAddress, ChannelPromise promise) throws Exception;

    /**
     * 用于 执行断开连接操作
     * @param ctx               为其执行断开连接操作的{@link ChannelHandlerContext}
     * @param promise          操作完成后通知的{@link ChannelPromise}
     * @throws Exception        thrown if an error accour
     */
    void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;

    //用于 执行关闭操作
    void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;

    /**
     * 用于 从当前注册的{@link EventLoop}执行注销操作。
     */
    void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;

    /**
     * 拦截{@link ChannelHandlerContext#read（）}。
     */
    void read(ChannelHandlerContext ctx) throws Exception;

    // 用于 执行写入操作。写入操作将通过{@link ChannelPipeline}写入消息。然后，一旦调用{@link Channel{flush（）}，就可以将它们刷新到实际的{@link Channel}
    void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception;

    // 用于 执行刷新操作。刷新操作将尝试刷新所有以前写入的挂起消息到 channel里面，然后发送出去。
    void flush(ChannelHandlerContext ctx) throws Exception;
}

两个必须的ChannelHandlerContext

HeadContext

他会做pipeline 出战操作的传播事件的最后的默认实现

final class HeadContext extends AbstractChannelHandlerContext
            implements ChannelOutboundHandler, ChannelInboundHandler {

        private final Unsafe unsafe;

        HeadContext(DefaultChannelPipeline pipeline) {
            super(pipeline, null, HEAD_NAME, false, true);
            unsafe = pipeline.channel().unsafe();
            setAddComplete();
        }

        @Override
        public ChannelHandler handler() {
            return this;
        }

        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
            // NOOP
        }

        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
            // NOOP
        }

        @Override
        public void bind(
                ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise)
                throws Exception {
            unsafe.bind(localAddress, promise);
        }

        @Override
        public void connect(
                ChannelHandlerContext ctx,
                SocketAddress remoteAddress, SocketAddress localAddress,
                ChannelPromise promise) throws Exception {
            unsafe.connect(remoteAddress, localAddress, promise);
        }

        @Override
        public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            unsafe.disconnect(promise);
        }

        @Override
        public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            unsafe.close(promise);
        }

        @Override
        public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            unsafe.deregister(promise);
        }

        @Override
        public void read(ChannelHandlerContext ctx) {
            unsafe.beginRead();
        }

        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            unsafe.write(msg, promise);
        }

        @Override
        public void flush(ChannelHandlerContext ctx) throws Exception {
            unsafe.flush();
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.fireExceptionCaught(cause);
        }

        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            invokeHandlerAddedIfNeeded();
            ctx.fireChannelRegistered();
        }

        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            ctx.fireChannelUnregistered();

            // Remove all handlers sequentially if channel is closed and unregistered.
            if (!channel.isOpen()) {
                destroy();
            }
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.fireChannelActive();

            readIfIsAutoRead();
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            ctx.fireChannelInactive();
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ctx.fireChannelRead(msg);
        }

        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.fireChannelReadComplete();

            readIfIsAutoRead();
        }

        private void readIfIsAutoRead() {
            if (channel.config().isAutoRead()) {
                channel.read();
            }
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            ctx.fireUserEventTriggered(evt);
        }

        @Override
        public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
            ctx.fireChannelWritabilityChanged();
        }
    }

TailContext

入站传播事件的默认实现

final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {

        TailContext(DefaultChannelPipeline pipeline) {
            super(pipeline, null, TAIL_NAME, true, false);
            setAddComplete();
        }

        @Override
        public ChannelHandler handler() {
            return this;
        }

        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception { }

        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception { }

        @Override
        public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { }

        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception { }

        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            // This may not be a configuration error and so don't log anything.
            // The event may be superfluous for the current pipeline configuration.
            ReferenceCountUtil.release(evt);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            onUnhandledInboundException(cause);
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            onUnhandledInboundMessage(msg);
        }

        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { }
    }

入站和出站传播


pipeline 可以调用传播方法
    如果是入站操作，那么 是从pipeline的head节点开始调用，因为head是一个ChannelHandlerContext,所以可以使用head的传播方法，他的传播是找 next节点是入站类型的ChannelHandlerContext，然后一直这么调用，可以再某一个ChannelHandlerContext停止往下找调用，如果不停止，直到tail节点，tail节点会做默认的处理
    
    如果是出站操作，那么 是从pipeline的tail节点开始找prev节点类型是出站类型的ChannelHandlerContext，然后一直调用，可以再某一个ChannelHandlerContext停止往上找调用，如果不停止，直到head节点，head节点会做默认的处理
    
    
    
ChannelHandlerContext传播，入站只传送到next节点的ChannelHandlerContext，出站只传送到prev节点的ChannelHandlerContext
    /*
    private AbstractChannelHandlerContext findContextInbound() {
        AbstractChannelHandlerContext ctx = this;
        do {
            ctx = ctx.next;
        } while (!ctx.inbound);
        return ctx;
    }

    private AbstractChannelHandlerContext findContextOutbound() {
        AbstractChannelHandlerContext ctx = this;
        do {
            ctx = ctx.prev;
        } while (!ctx.outbound);
        return ctx;
    } 
    */
    
    
 pipeline的传播方法最终还是ChannelHandlerContext的传播方法

信息读取和写出需要用到的ByteBuf


Pooled  申请的内存放到池子 进行重用
    
UnPooled   不放到池子里，用一次申请一次  
 
   通过调用静态方法获取 ByteBuf

简单使用：

server



public class NettyServerDemo {

    public static void main(String[] args) {

        //初始化两个大线程池，bossGroup用来注册NioServerSocketChannel对象
        //workerGroup用来注册 NioServerSocketChannel对象接待的NioSocketChannel对象
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();


        //一个测试的业务线程池，线程数也是一个，用来测试
        // 每一个ChannelHandlerContext可以有自己独特的执行线程
        EventLoopGroup bossBusinessGroup = new NioEventLoopGroup(1);

        ////创建一个服务端启动器对象，这个启动器将会帮你做一些默认的实现，使得使用变得更加简单
        // 当然也可以自己写一些复杂的代码启动
        ServerBootstrap serverBootstrap = new ServerBootstrap();


         serverBootstrap
                 //设置group和childGroup
                 .group(bossGroup, workerGroup)
                //设置Channel类型
                 .channel(NioServerSocketChannel.class)
                 //给NioServerSocketChannel对象的pipeline添加一个handler
                 //这个ChannelHandler里面initChannel方法会在 channel注册成功后
                 // 然后用NioServerSocketChannel对象注册的IO线程调用
                 .handler(new ChannelInitializer<>() {
                     @Override
                     protected void initChannel(Channel ch) throws Exception {
                         //ch就是NioServerSocketChannel对象
                         ChannelPipeline pipeline = ch.pipeline();
                         System.out.println("===================");
                         System.out.println("IO线程："+Thread.currentThread());

                         System.out.println("====================");

                         //在pipeline中添加一个ChannelHandler，并指定执行它的线程池
                         //如果不指定线程池，那么执行线程就是ch的IO线程
                         // public EventExecutor executor() {
                         //        if (executor == null) {
                         //           return channel().eventLoop();
                         //        } else {
                         //           return executor;
                         //        }
                         //}
                         pipeline.addLast(bossBusinessGroup, new MyChannelInboundHandlerAdapter());

                     }
                 }).childHandler(new ChannelInitializer<>() {
                     @Override
                     protected void initChannel(Channel ch) throws Exception {
                         System.out.println("所有的SocketChannel的pipeline都会添加这个 ChannelHandler");
                     }
                });

        try {
              //获取一个bind IO操作的ChannelFuture,然后获取channel
              //获取该channel的closeFuture，最后同步等待
              serverBootstrap.bind(8000).channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }
}
class MyChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter{


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {



        //如果此ChannelHandler被添加到 NioServerSocketChannel对象的pipeline上的话
        System.out.println("====================");
        System.out.println("msg就是一个NioSocketChannel对象："+msg.getClass());
        System.out.println("执行MyChannelInboundHandlerAdapter的业务线程"+Thread.currentThread());
        System.out.println("=======================");

        //一定要传播，当然可以不传播，但是必须手动把msg释放
        //ReferenceCountUtil.release(msg);
        super.channelRead(ctx, msg);

    }
}

client

1
2

总结：

一个channel应该被注册到一个EventLoop, 此后该Channel的IO操作都由此 EventLoop去执行

一个Channel应该有一个ChannelPipeline对象 ,这个 ChannelPipeline对象可以对IO操作结果进行多重的处理

在多重处理时引入了一个 ChannelHandlerContext的对象， ChannelHandlerContext看名字就知道是一个ChannelHandler对象的上下文对象，那么这个上下文对象包含了 ChannelHandler对象引用，也包含了所属ChannelPipeline对象的引用，还包含 ChannelHandlerContext对象在事件传播或者事件执行时的执行者EventExecutor引用，如果此引用为空，事件传播或者事件执行时则使用所属pipeline的channel的 IO事件执行器EventLoop，

这里我认为的事件叫做一个IO操作的结果

对应的IO操作完成，pipeline就会对此IO操作完成信息进行事件传播，pipeline的链表上的 ChannelHandlerContext会取到ChannelHandler，使用对应的执行器调用对应的回调方法

最难的是 netty的内存分配算法：还未搞懂

本文作者： 忘忧症
本文链接： https://NepenthesZGW.github.io/2020/06/20/framework/Netty/Netty基础学习/
版权声明： 本博客所有文章除特别声明外，均采用 MIT 许可协议。转载请注明出处！