Netty源码分析系列--8. Channel和ChannelPipeline

Channel的创建

前文中提到ServerBootStrap在启动配置中的方法channel（NioServerSocketChannel.class）会创建一个ReflectiveChannelFactory并赋值给成员变量channelFactory。当调用绑定端口号方法bind(8899)时，本质是通过反射调用NioServerSocketChannel的构造函数来创建Channel。

1. 创建ChannelFactory实例：

public B channel(Class<? extends C> channelClass) {
    if (channelClass == null) {
        throw new NullPointerException("channelClass");
    }
    return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
}

2. NioServerSocketChannel的构造函数

public class NioServerSocketChannel extends AbstractNioMessageChannel
                         implements io.netty.channel.socket.ServerSocketChannel {

  private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();

  private static ServerSocketChannel newSocket(SelectorProvider provider) {
      try {
          return provider.openServerSocketChannel();
      } catch (IOException e) {
          throw new ChannelException(
                  "Failed to open a server socket.", e);
      }
  }

  // 1. 构造函数，调用静态方法创建ServerSocketChannel
  public NioServerSocketChannel() {
      this(newSocket(DEFAULT_SELECTOR_PROVIDER));
  }

  // 2. 调用父类构造函数
  public NioServerSocketChannel(ServerSocketChannel channel) {
      super(null, channel, SelectionKey.OP_ACCEPT);
      config = new NioServerSocketChannelConfig(this, javaChannel().socket());
  }
}

• 在构造函数中首先调用静态方法newSocket()，创建一个Java Nio ServerSocketChannel实例，它继承SelectableChannel。

• 然后调用父类AbstractNioChannel的构造函数，它会持有上一步创建的ServerSocketChannel实例并赋值给变量ch，同时配置为非阻塞ch.configureBlocking(false)，如下：

  protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
      super(parent);
      // 1. NioServerSocketChannle(netty)持有ServerSocketChannel(jdk)
      this.ch = ch;
      this.readInterestOp = readInterestOp;
      try {
          // 2. 配置为非阻塞
          ch.configureBlocking(false);
      } catch (IOException e) {
          try {
              ch.close();
          } catch (IOException e2) {
              if (logger.isWarnEnabled()) {
                  logger.warn(
                          "Failed to close a partially initialized socket.", e2);
              }
          }
  
          throw new ChannelException("Failed to enter non-blocking mode.", e);
      }
  }
  

ChannelPipeline

1. ChannelPipeline的创建

继续查看父类AbstractChannel可以发现，在构造函数中，创建了ChannelPipeline，并赋值给pipeline对象。

protected AbstractChannel(Channel parent, ChannelId id) {
    this.parent = parent;
    this.id = id;
    unsafe = newUnsafe();
    // 1. 赋值给pipeline对象
    pipeline = newChannelPipeline();
}

protected DefaultChannelPipeline newChannelPipeline() {
    // 当前对象this，即Channel传入
    return new DefaultChannelPipeline(this);
}

• Channel通过变量pipeline持有ChannelPipeline。
• DefaultChannelPipeline的构造函数接受this为参数，因此ChannelPipeline也持有Channel。

2. ChannelPipeline的拦截过滤器模式

• ChannelPipeline中可以添加多个ChannelHandler，I/O事件在ChannelPipeline中依次传递。它也提供了添加、删除ChannelHandler的方法，如addLast()、removeLast()等。

ChannelPipeline是线程安全的，可以随时添加或删除ChannelHandler。

• ChannelPipeline中的ChannelHandler分为InboundHandler和OutBoundHandler：InboundHandler只处理I/O输入请求，OutBoundHandler只处理I/O输出请求。例如：

   ChannelPipeline p = ...;
   p.addLast("1", new InboundHandlerA());
   p.addLast("2", new InboundHandlerB());
   p.addLast("3", new OutboundHandlerA());
   p.addLast("4", new OutboundHandlerB());
   p.addLast("5", new InboundOutboundHandlerX());
  

image.png

1至5是ChannelHandler的整体添加顺序，其中1和2是InboundHandler，3和4是OutboundHandler，而5即可以处理输入又可以处理输出。因此I/O事件的传播如上图：
输入: 1 -> 2 -> 5
输出: 5 -> 4 -> 3

• ChannelPipeline中I/O事件的传播依靠调用ChannelHandlerContext的方法：

public class MyInboundHandler extends {@link ChannelInboundHandlerAdapter} {
  @Override
  public void channelActive({@link ChannelHandlerContext} ctx) {
      System.out.println("Connected!");
      ctx.fireChannelActive();
  }
}

在上例中，自定义Handler MyInboundHandler在自己的channelActive方法中调用了ctx.fireChannelActive()方法，将事件传播给下一个Handler。

3. ChannelPipeline中添加耗时任务的方式

当自定义Handler中需要处理耗时较长的任务时，有2种方式：

• 添加到ChannelPipeline时，指定事件执行组EventExecutorGroup：

  EventExecutorGroup group = new DefaultEventExecutorGroup(16);
  
  ChannelPipeline pipeline = ......;
  
  pipeline.addLast("decoder", new MyProtocolDecoder());
  pipeline.addLast("encoder", new MyProtocolEncoder());
  // 指定事件执行组
  pipeline.addLast(group, "handler", new MyBusinessLogicHandler());
  

• 或者在自定义Handler中使用线程池ExecutorService处理。