阅读 111

Netty源码分析

Netty入门

? Netty 是业界最流行的NO框架之一，它的健壮性、功能、性能、可定制性和可扩展性在同类框架中都是首屈一指的，它已经得到成百上千的商用项目验证，例如 Hadoop的RPC框架Avro就使用了Nety作为底层通信框架，其他还有业界主流的RPC框架，也使用Nety来构建高性能的异步通信能力。

? 参考：

优势：

  • API简单方便开发
  • 功能强大，内置多种编码解码器，支持主流协议
  • 定制能力强，通过ChannelHandler来定制
  • 性能不错
  • 社区强大，使用案列多

线程模型

  1. 设置服务端ServerBootStrap启动参数
  2. 通过ServerBootStrap的bind方法启动服务端，bind方法会在parentGroup中注册NioServerScoketChannel，监听客户端的连接请求
  3. Client发起连接CONNECT请求，bossGroup中的NioEventLoop不断轮循是否有新的客户端请求，如果有，ACCEPT事件触发
  4. ACCEPT事件触发后，parentGroup中NioEventLoop会通过NioServerSocketChannel获取到对应的代表客户端的NioSocketChannel，并将其注册到childGroup中
  5. workGroup中的NioEventLoop不断检测自己管理的NioSocketChannel是否有读写事件准备好，如果有的话，调用对应的ChannelHandler进行处理

生命周期

扩展点：

  • handlerAdded

    新建立的连接会按照初始化策略，把handler添加到该channel的pipeline里面，也就是channel.pipeline.addLast(new LifeCycleInBoundHandler)执行完成后的回调

  • channelRegistered

    当该连接分配到具体的worker线程后，该回调会被调用

  • channelActive

    channel的准备工作已经完成，所有的pipeline添加完成，并分配到具体的线上上，说明该channel准备就绪，可以使用了

  • channelRead

    客户端向服务端发来数据，每次都会回调此方法，表示有数据可读

  • channelReadComplete

    服务端每次读完一次完整的数据之后，回调该方法，表示数据读取完毕

  • channelInactive

    当连接断开时，该回调会被调用，说明这时候底层的TCP连接已经被断开了

  • channelUnRegistered

    对应channelRegistered，当连接关闭后，释放绑定的workder线程

  • handlerRemoved

    对应handlerAdded，将handler从该channel的pipeline移除后的回调方法

Firexxx方法执行链路：

在ChannelPipeline和ChannelHandlerContext中，都定义了相同的9个以fire开头的方法，如下所示

可以发现这两个接口定义的9个方法与ChannelInboundHandler定义的9个方法是一一对应的，只是在定义每个方法的时候，在前面加了1个fire。

从总体上来说，在调用的时候，是按照如下顺序进行的：

1、先是ChannelPipeline中的fireXXX方法被调用

2、ChannelPipeline中的fireXXX方法接着调用ChannelPipeline维护的ChannelHandlerContext链表中的第一个节点即HeadContext 的fireXXX方法

3、ChannelHandlerContext 中的fireXXX方法调用ChannelHandler中对应的XXX方法。由于可能存在多个ChannelHandler，因此每个ChannelHandler的xxx方法又要负责调用下一个ChannelHandlerContext的fireXXX方法，直到整个调用链完成

Netty服务端开发步骤

public class NettyServer {

    public static void main(String[] args) {
        new NettyServer().bing(7397);
    }

    private void bing(int port) {
        //配置服务端NIO线程组
        EventLoopGroup parentGroup = new NioEventLoopGroup(); //NioEventLoopGroup extends MultithreadEventLoopGroup Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
        EventLoopGroup childGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(parentGroup, childGroup)
                    .channel(NioServerSocketChannel.class)    //非阻塞模式
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childHandler(new MyChannelInitializer());
            ChannelFuture f = b.bind(port).sync();
            System.out.println("itstack-demo-netty server start done. {关注公众号：bugstack虫洞栈，获取源码}");
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            childGroup.shutdownGracefully();
            parentGroup.shutdownGracefully();
        }
    }
}

public class MyChannelInitializer extends ChannelInitializer {

    @Override
    protected void initChannel(SocketChannel channel) {
        System.out.println("链接报告开始");
        System.out.println("链接报告信息：有一客户端链接到本服务端");
        System.out.println("链接报告IP:" + channel.localAddress().getHostString());
        System.out.println("链接报告Port:" + channel.localAddress().getPort());
        System.out.println("链接报告完毕");
    }
}

? NioEventLoopGroup是个线程组包含了一组NIO线程，专门用来处理网络事件。实际上就是Reactor线程组，一般服务端需要创建两个NIO线程组，一个线程组用来处理网络连接请求，一个线程组用来处理读写请求，分别称为Boss线程和Worker线程。

? ServerBootstrap是Netty用于启动NIO服务端的辅助类，目的是降低开发的复杂度。接着配置Channel为NioServerSocketChannel,option函数是用来配置channel的TCP参数。最后通过方法Handler或childHandler添加处理器。处理器一般是用作：记录日志，编码解码，数据包处理，业务逻辑处理等等。

? 服务端创建配置完毕后通过函数bind来监听端口，调用sync函数会返回一个ChannelFeature类似于Feature。 f.channel().closeFuture().sync()执行会被阻塞，直到服务器链路关闭才会被唤醒。

Netty客户端开发步骤

public class NettyClient {

    public static void main(String[] args) {
        new NettyClient().connect("127.0.0.1", 7397);
    }

    private void connect(String inetHost, int inetPort) {
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(workerGroup);
            b.channel(NioSocketChannel.class);
            b.option(ChannelOption.AUTO_READ, true);
            b.handler(new MyChannelInitializer());
            ChannelFuture f = b.connect(inetHost, inetPort).sync();
            System.out.println("itstack-demo-netty client start done. {关注公众号：bugstack虫洞栈，获取源码}");
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }
}
public class MyChannelInitializer extends ChannelInitializer {
    @Override
    protected void initChannel(SocketChannel channel) throws Exception {
        System.out.println("链接报告开始");
        System.out.println("链接报告信息：本客户端链接到服务端。channelId：" + channel.id());
        System.out.println("链接报告完毕");
    }
}

和服务端的创建过程类似，将两个NIO线程组换成一个，ServerBootstrap换成Bootstrap，NioServerSocketChannel换成NioSocketChannel

TCP粘包和半包

? TCP是个“流”协议，所谓流，就是没有界限的一串数据。大家可以想想河里的流水，它们是连成一片的，其间并没有分界线。TCP底层并不了解上层业务数据的具体含义，它会根据TCP缓冲区的实际情况进行包的划分，所以在业务上认为，一个完整的包可能会被TCP拆分成多个包进行发送，也有可能把多个小的包封装成一个大的数据包发送，这就是所谓的TCP粘包和拆包问题。

? 粘包和半包指得是 不是一次正常的ByteBuf接收。

? 粘包：

? 接收端接收到的数据是由多个ByteBuf粘在一起的，通常可见于MTU较大发送的数据较小，这样会导致一次TCP发送会包含多个的ByteBuf。

? 拆包：

? 接收端的数据不是一次完整的数据，通常出现在MTU较小，完整数据不能一次性全部传输。

示意：

接收端收到的第一个包，正常。

接收端收到的第二个包，就是一个粘包。 将发送端的第二个包、第三个包，粘在一起了。

接收端收到的第三个包，第四个包，就是半包。将发送端的的第四个包，分开成了两个了。

为什么会产生粘包和半包？

? 在Netty的底层还是使用TCP进行数据传输的，应用层通过ByteBuf来进行数据存储到了底层操作系统时还是按照字节流的方式发送数据，到了接收端的时候再由字节数据封装成ByteBuf对象。对应Netty来说获得ByteBuf就是对接收缓存区的读取，但是上层应用读取底层缓存区的数据量是有限的，这就会导致较大的数据包不能一次性读完就会产生半包现象，如果是数据包较小，每次读取都读取到多个数据包这就会导致粘包现象。

如何解决这个问题？

一般来说解决半包和粘包问题一般通过ByteToMessageDecoder子类来实现。

在ServerBootStrap中配置处理器的时候就是通过ChannelInitializer来对客户端channel的pipeline的handler进行初始化的

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof ByteBuf) {
        CodecOutputList out = CodecOutputList.newInstance();
        try {
            ByteBuf data = (ByteBuf) msg;
            first = cumulation == null;
            if (first) {
                cumulation = data;
            } else {
                cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
            }
            // 半包协议在子类中进行判断，是否当前能读到包的内容。如果读不到则将内容缓存，直到一下次
            callDecode(ctx, cumulation, out);
        } catch (DecoderException e) {
            throw e;
        } catch (Exception e) {
            throw new DecoderException(e);
        } finally {
            if (cumulation != null && !cumulation.isReadable()) {
                numReads = 0;
                cumulation.release();
                cumulation = null;
            } else if (++ numReads >= discardAfterReads) {
                // We did enough reads already try to discard some bytes so we not risk to see a OOME.
                // See https://github.com/netty/netty/issues/4275
                numReads = 0;
                discardSomeReadBytes();
            }

            int size = out.size();
            decodeWasNull = !out.insertSinceRecycled();
            fireChannelRead(ctx, out, size);
            out.recycle();
        }
    } else {
        ctx.fireChannelRead(msg);
    }
}

定长协议

假设我们规定每3个字节，表示一个有效报文，如果我们分4次总共发送以下9个字节：

   +---+----+------+----+
   | A | BC | DEFG | HI |
   +---+----+------+----+

那么根据协议，我们可以判断出来，这里包含了3个有效的请求报文

  +-----+-----+-----+
  | ABC | DEF | GHI |
  +-----+-----+-----+

? 每个消息都是固定长度的，这样当超过这个长度以后都是另一个消息，对于长度不足的可以补充数据。一般很少用这种方式如果消息长度的弹性很大就需要将长度设置为消息的最大值，非常的浪费资源而且不灵活。

FixedLengthFrameDecoder:

? 在Netty中使用FixedLengthFrameDecoder来实现消息定长的操作。

? FixedLengthFrameDecoder 固定长度解码处理器，它能够按照指定的长度对消息进行自动解码。无论一次接收到多少数据报，它都会按照构造器中设置的固定长度进行解码，如果是半包消息，FixedLengthFrameDecoder 会缓存半包消息并等待下个包到达之后进行拼包合并，直到读取一个完整的消息包。

FixedLengthFrameDecoder 的使用：

Server端：

public class FixedLengthFrameDecoderServer {
    public static void main(String[] args) 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() { // (4)
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new FixedLengthFrameDecoder(3));
                            // 自定义这个ChannelInboundHandler打印拆包后的结果
                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                    if (msg instanceof ByteBuf) {
                                        ByteBuf packet = (ByteBuf) msg;
                                        System.out.println(
                                                new Date().toLocaleString() + ":" + packet.toString(Charset.defaultCharset()));
                                    }
                                }
                            });
                        }
                    });
            // Bind and start to accept incoming connections.
            ChannelFuture f = b.bind(8080).sync(); // (7)
            System.out.println("FixedLengthFrameDecoderServer Started on 8080...");
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

Client端：

public class FixedLengthFrameDecoderClient {
   public static void main(String[] args) throws Exception {
      EventLoopGroup workerGroup = new NioEventLoopGroup();
      try {
         Bootstrap b = new Bootstrap(); // (1)
         b.group(workerGroup); // (2)
         b.channel(NioSocketChannel.class); // (3)
         b.option(ChannelOption.SO_KEEPALIVE, true); // (4)
         b.handler(new ChannelInitializer() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
               ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                   //在于server建立连接后，即发送请求报文
                  public void channelActive(ChannelHandlerContext ctx) {
                     ByteBuf A = Unpooled.buffer().writeBytes("A".getBytes());
                     ByteBuf BC = Unpooled.buffer().writeBytes("BC".getBytes());
                     ByteBuf DEFG = Unpooled.buffer().writeBytes("DEFG".getBytes());
                     ByteBuf HI = Unpooled.buffer().writeBytes("HI".getBytes());
                     ctx.writeAndFlush(A);
                     ctx.writeAndFlush(BC);
                     ctx.writeAndFlush(DEFG);
                     ctx.writeAndFlush(HI);
                    }
               });
            }
         });
         // Start the client.
         ChannelFuture f = b.connect("127.0.0.1",8080).sync(); // (5)
         // Wait until the connection is closed.
         f.channel().closeFuture().sync();
      } finally {
         workerGroup.shutdownGracefully();
      }
   }
}

特殊字符分隔符协议

在包尾部增加回车或者空格符等特殊字符进行分割 。

例如，按行解析，遇到字符\n、\r\n的时候，就认为是一个完整的数据包。对于以下二进制字节流：

+--------------+
| ABC\nDEF\r\n |
+--------------+

那么根据协议，我们可以判断出来，这里包含了2个有效的请求报文

+-----+-----+
| ABC | DEF |
+-----+-----+
DelimiterBasedFrameXXX

通过DelimiterBasedFrameXXX可以实现自定义特殊字段的分割符协议：

DelimiterBasedFrameDecoder解码器

DelimiterBasedFrameEncoder编码器

添加到处理器链上

channel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, false, Delimiters.lineDelimiter()))
LineBasedframeDecoder

按照换行符来分割：

如果需要按照行来进行数据分割，则Netty内置了解码器：LineBasedframeDecoder

? LineBasedframeDecoder的工作原理是它依次遍历 ByteBuf中的可读字节，判断看是否有“\n”或者“\r\n”，如果有，就以此位置为结束位置，从可读索引到结束位置区间的字节就组成了一行。它是以换行符为结束标志的解码器，支持携带结束符或者不携带结束符两种解码方式，同时支持配置单行的最大长度。如果连续读取到最大长度后仍然没有发现换行符，就会抛岀异常，同时忽略掉之前读到的异常码流

变长协议

大多数的协议（私有或者公有），协议头中会携带长度字段，用于标识消息体或者整包消息的长度，例如SMPP、HTTP协议等。由于基于长度解码需求 的通用性，Netty提供了LengthFieldBasedFrameDecoder/LengthFieldPrepender，自动屏蔽TCP底层的拆包和粘 包问题，只需要传入正确的参数，即可轻松解决“读半包“问题。

? 发送方使用LengthFieldPrepender给实际内容Content进行编码添加报文头Length字段，接受方使用LengthFieldBasedFrameDecoder进行解码。协议格式如下所示：

+--------+----------+
| Length |  Content |
+--------+----------+

Length字段：

表示Conent部分的字节数，例如Length值为100，那么意味着Conent部分占用的字节数就是100。

Length字段本身是个整数，也要占用字节，一般会使用固定的字节数表示。例如我们指定使用2个字节(有符号)表示length，那么可以表示的最大值为32767(约等于32K)，也就是说，Content部分占用的字节数，最大不能超过32767。当然，Length字段存储的是Content字段的真实长度。

Content字段：

? 是我们要处理的真实二进制数据。 在发送Content内容之前，首先需要获取其真实长度，添加在内容二进制流之前，然后再发送。Length占用的字节数+Content占用的字节数，就是我们总共要发送的字节。

事实上，我们可以把Length部分看做报文头，报文头包含了解析报文体(Content字段)的相关元数据，例如Length报文头表示的元数据就是Content部分占用的字节数。当然，LengthFieldBasedFrameDecoder并没有限制我们只能添加Length报文头，我们可以在Length字段前或后，加上一些其他的报文头，此时协议格式如下所示：

  +---------+--------+----------+----------+
  |........ | Length |  ....... |  Content |
  +---------+--------+----------+----------+
LengthFieldPrepender

构造方法：

public LengthFieldPrepender(
            ByteOrder byteOrder, int lengthFieldLength,
            int lengthAdjustment, boolean lengthIncludesLengthFieldLength)
  • byteOrder：表示Length字段本身占用的字节数使用的是大端还是小端编码
  • lengthFieldLength：表示Length字段本身占用的字节数,只可以指定 1, 2, 3, 4, 或 8
  • lengthAdjustment：length的大小是否也计入总大小中
  • lengthIncludesLengthFieldLength：表示Length字段本身占用的字节数是否包含在Length字段表示的值中。
LengthFieldBasedFrameDecoder

构造方法：

public LengthFieldBasedFrameDecoder(
        ByteOrder byteOrder, int maxFrameLength, int lengthFieldOffset, int lengthFieldLength,
        int lengthAdjustment, int initialBytesToStrip, boolean failFast)

  • byteOrder：表示协议中Length字段的字节是大端还是小端

  • maxFrameLength： 表示协议中Content字段的最大长度，如果超出，则抛出TooLongFrameException异常。

  • lengthFieldOffset：表示Length字段的偏移量，即在读取一个二进制流时，跳过指定长度个字节之后的才是Length字段。如果Length字段之前没有其他报文头，指定为0即可。如果Length字段之前还有其他报文头，则需要跳过之前的报文头的字节数。

  • lengthFieldLength： 表示Length字段占用的字节数。指定为多少，需要看实际要求，不同的字节数，限制了Content字段的最大长度。

    • 如果lengthFieldLength是1个字节，那么限制为128bytes；
    • 如果lengthFieldLength是2个字节，那么限制为32767(约等于32K)；
    • 如果lengthFieldLength是3个字节，那么限制为8388608(约等于8M)；
    • 如果lengthFieldLength是4个字节，那么限制为2147483648(约等于2G)。lengthFieldLength与maxFrameLength并不冲突。例如我们现在希望限制报文Content字段的最大长度为32M。显然，我们看到了上面的四种情况，没有任何一个值，能刚好限制Content字段最大值刚好为32M。那么我们只能指定lengthFieldLength为4个字节，其最大限制2G是大于32M的，因此肯定能支持。但是如果Content字段长度真的是2G，server端接收到这么大的数据，如果都放在内存中，很容易造成内存溢出。为了避免这种情况，我们就可以指定maxFrameLength字段，来精确的指定Content部分最大字节数，显然，其值应该小于lengthFieldLength指定的字节数最大可以表示的值。

  • lengthAdjustment：length的大小是否也计入总大小中

  • initialBytesToStrip：解码后跳过的初始字节数，表示获取完一个完整的数据报文之后，忽略前面指定个数的字节。例如报文头只有Length字段，占用2个字节，在解码后，我们可以指定跳过2个字节。这样封装到ByteBuf中的内容，就只包含Content字段的字节内容不包含Length字段占用的字节。

  • failFast：如果为true，则表示读取到Length字段时，如果其值超过maxFrameLength，就立马抛出一个 TooLongFrameException，而为false表示只有当真正读取完长度域的值表示的字节之后，才会抛出 TooLongFrameException，默认情况下设置为true，建议不要修改，否则可能会造成内存溢出。

序列化器

NettyServer启动流程分析

//配置服务端NIO线程组
EventLoopGroup parentGroup = new NioEventLoopGroup();
EventLoopGroup childGroup = new NioEventLoopGroup();
try {
    ServerBootstrap b = new ServerBootstrap();
    b.group(parentGroup, childGroup)
        .channel(NioServerSocketChannel.class)    //非阻塞模式
        .option(ChannelOption.SO_BACKLOG, 128)
        .childHandler(new MyChannelInitializer());
    ChannelFuture f = b.bind(port).sync();
    System.out.println("itstack-demo-netty server start done. {关注公众号：bugstack虫洞栈，获取源码}");
    f.channel().closeFuture().sync();
} catch (InterruptedException e) {
    e.printStackTrace();
} finally {
    childGroup.shutdownGracefully();
    parentGroup.shutdownGracefully();
}

可以看到b.bind()方法是核心方法，其他方法只是给ServerBootstrap这个辅助类设置必要参数

AbstractBootStrap.bind()->doBind()

private ChannelFuture doBind(final SocketAddress localAddress) {
    final ChannelFuture regFuture = initAndRegister(); // 初始化一个Channel，作为Server的Channel
    final Channel channel = regFuture.channel();//从异步线程中阻塞等待初始化完毕
    if (regFuture.cause() != null) {
        return regFuture;
    }
    if (regFuture.isDone()) {
        // At this point we know that the registration was complete and successful.
        ChannelPromise promise = channel.newPromise();
        doBind0(regFuture, channel, localAddress, promise);
        return promise;
    } else {
        // Registration future is almost always fulfilled already, but just in case it‘s not.
        final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
        regFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                Throwable cause = future.cause();
                if (cause != null) {
                    // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
                    // IllegalStateException once we try to access the EventLoop of the Channel.
                    promise.setFailure(cause);
                } else {
                    // Registration was successful, so set the correct executor to use.
                    // See https://github.com/netty/netty/issues/2586
                    promise.registered();

                    doBind0(regFuture, channel, localAddress, promise);
                }
            }
        });
        return promise;
    }
}

进入到initAndRegister方法中去

final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
        channel = channelFactory.newChannel();// 通过工厂类的方式反射出前面设置的`NioServerSocketChannel`类
        init(channel);// 实例化的channel进行初始化
    } catch (Throwable t) {
        if (channel != null) {
            // channel can be null if newChannel crashed (eg SocketException("too many open files"))
            channel.unsafe().closeForcibly();
            // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
            return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
        }
        // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
        return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
    }
	// 向前面设置的parentGroup中绑定当前的channel
    ChannelFuture regFuture = config().group().register(channel);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }
    return regFuture;
}

核心方法还是init，因为这个方法是抽象方法，所以找到ServerBootStrap中

void init(Channel channel) throws Exception {
    final Map, Object> options = options0();
    synchronized (options) {
        setChannelOptions(channel, options, logger); // 设置channel TCP参数
    }
    final Map, Object> attrs = attrs0();
    synchronized (attrs) {
        for (Entry, Object> e: attrs.entrySet()) {
            @SuppressWarnings("unchecked")
            AttributeKey key = (AttributeKey) e.getKey();
            channel.attr(key).set(e.getValue());
        }
    }
    ChannelPipeline p = channel.pipeline();
    final EventLoopGroup currentChildGroup = childGroup;
    final ChannelHandler currentChildHandler = childHandler;
    final Entry, Object>[] currentChildOptions;
    final Entry, Object>[] currentChildAttrs;
    synchronized (childOptions) {
        currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0));
    }
    synchronized (childAttrs) {
        currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0));
    }
	// 给ServerBoostStrap中的pipeline最末尾加上一个ChannelInitializer
    p.addLast(new ChannelInitializer() {
        @Override
        public void initChannel(final Channel ch) throws Exception {
            final ChannelPipeline pipeline = ch.pipeline();
            ChannelHandler handler = config.handler();
            if (handler != null) {
                pipeline.addLast(handler);
            }

            ch.eventLoop().execute(new Runnable() {
                @Override
                public void run() {
                    pipeline.addLast(new ServerBootstrapAcceptor(
                        ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                }
            });
        }
    });
}


研究上述代码，是将ServerBootStrap中的Pipeline添加一个ChannelInitializer实现了initChannel方法的实现类


什么是ChannelInitializer？


 


?	是一种特殊的ChannelInboundHandler，它提供了在通道注册到eventLoop后初始化通道的简单方法，ChannelInitializer的主要目的是为程序员提供了一个简单的工具，用于在某个Channel注册到EventLoop后，对这个Channel执行一些初始化操作。ChannelInitializer虽然会在一开始会被注册到Channel相关的pipeline里，但是在初始化完成之后（Channel被register到某个EventLoop中），ChannelInitializer会将自己从pipeline中移除，不会影响后续的操作。


?	查阅DefaultChannelPipeline.addLast方法：


@Override
public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    final AbstractChannelHandlerContext newCtx;
    synchronized (this) {
        checkMultiplicity(handler);

        newCtx = newContext(group, filterName(name, handler), handler);

        addLast0(newCtx);

        // If the registered is false it means that the channel was not registered on an eventloop yet.
        // In this case we add the context to the pipeline and add a task that will call
        // ChannelHandler.handlerAdded(...) once the channel is registered.
        if (!registered) {
            newCtx.setAddPending();
            callHandlerCallbackLater(newCtx, true);
            return this;
        }

        EventExecutor executor = newCtx.executor();
        if (!executor.inEventLoop()) {
            newCtx.setAddPending();
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    callHandlerAdded0(newCtx);
                }
            });
            return this;
        }
    }
    callHandlerAdded0(newCtx);
    return this;
}
/**
** 添加一个回调接口
**/
private void callHandlerCallbackLater(AbstractChannelHandlerContext ctx, boolean added) {
    assert !registered;
    PendingHandlerCallback task = added ? new PendingHandlerAddedTask(ctx) : new PendingHandlerRemovedTask(ctx);
    PendingHandlerCallback pending = pendingHandlerCallbackHead;
    // 将这个handler放入链表的头部
    if (pending == null) {
        pendingHandlerCallbackHead = task;
    } else {
        // Find the tail of the linked-list.
        while (pending.next != null) {
            pending = pending.next;
        }
        pending.next = task;
    }
}



最终这个PendingHandlerCallback会调用到之前注册的ChannelInitializer的handlerAdded方法：


调用的链路如下：执行完子类的initChannel方法后，会将此handler从该channel中删除


所以最终将ServerBootstrapAcceptor放入了channel的pipline中


@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    if (ctx.channel().isRegistered()) {
        // This should always be true with our current DefaultChannelPipeline implementation.
        // The good thing about calling initChannel(...) in handlerAdded(...) is that there will be no ordering
        // surprises if a ChannelInitializer will add another ChannelInitializer. This is as all handlers
        // will be added in the expected order.
        initChannel(ctx);
    }
}   

private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
    if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) { // Guard against re-entrance.
        try {
            initChannel((C) ctx.channel());
        } catch (Throwable cause) {
            // Explicitly call exceptionCaught(...) as we removed the handler before calling initChannel(...).
            // We do so to prevent multiple calls to initChannel(...).
            exceptionCaught(ctx, cause);
        } finally {
            remove(ctx);
        }
        return true;
    }
    return false;
}

private void remove(ChannelHandlerContext ctx) {
    try {
        ChannelPipeline pipeline = ctx.pipeline();
        if (pipeline.context(this) != null) {
            pipeline.remove(this);
        }
    } finally {
        initMap.remove(ctx);
    }
}



分析一下ServerBootstrapAcceptor，基础于ChannelInboundHandlerAdapter，实现了channelRead和exceptionCaught方法


在分析ServerBootstrapAcceptor之前先看下去：


io.netty.bootstrap.AbstractBootstrap#initAndRegister


?	->io.netty.channel.AbstractChannel.AbstractUnsafe#register


?		->io.netty.channel.nio.AbstractNioChannel#doRegister

可以看到在这里将channel和boss select注册到一起了，但是没有指定监听事件（0）


@Override
protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
            return;
        } catch (CancelledKeyException e) {
            if (!selected) {
                // Force the Selector to select now as the "canceled" SelectionKey may still be
                // cached and not removed because no Select.select(..) operation was called yet.
                eventLoop().selectNow();
                selected = true;
            } else {
                // We forced a select operation on the selector before but the SelectionKey is still cached
                // for whatever reason. JDK bug ?
                throw e;
            }
        }
    }
}



?	回到之前的io.netty.bootstrap.AbstractBootstrap#doBind他会往下走，调用到doBind0方法，此方法就是将当前的channel设置为触发连接初始化事件。


?	再回到ServerBootstrapAcceptor#channelRead方法,当boss线程接收到了来自客户端的连接请求后，会将客户端channle绑定到worker线程的select上。这样就完成了服务端的初始化工作。


public void channelRead(ChannelHandlerContext ctx, Object msg) {
    final Channel child = (Channel) msg;

    child.pipeline().addLast(childHandler);

    setChannelOptions(child, childOptions, logger);

    for (Entry, Object> e: childAttrs) {
        child.attr((AttributeKey) e.getKey()).set(e.getValue());
    }

    try {
        childGroup.register(child).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    forceClose(child, future.cause());
                }
            }
        });
    } catch (Throwable t) {
        forceClose(child, t);
    }
}

原文：https://www.cnblogs.com/afsun/p/15236943.html

                
                            


            

                

                    
文章分类

                    
                        
代码人生

                    
                


                

                    
文章标签

                    

                                                
                            
netty

                        
                                                
                            
源码

                        
                                            

                

                

                    版权声明：本站是系统测试站点，无实际运营。本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至
                    XXXXXXo@163.com 举报，一经查实，本站将立刻删除。
                

            


            
        

        

        

            
相关推荐