Netty 的组件和设计

Channel & EventLoop & ChannelFuture

概览:

• Channel — Socket

• EventLoop — 控制流、多线程处理、并发

• ChannelFuture — 异步通知

Channel 接口

public interface Channel
extends AttributeMap, ChannelOutboundInvoker, java.lang.Comparable<Channel>

文档:

与网络套接字或能够进行I / O操作（例如读取，写入，连接和绑定）的组件的联系

通道为用户提供：

• 通道的当前状态（例如，它是否已打开？是否已连接？），

• 通道的配置参数（例如接收缓冲区大小），

• 通道支持的I / O操作（例如，读取，写入，连接和绑定），以及

• ChannelPipeline，处理所有与通道关联的I / O事件和请求

所有I / O操作都是异步的

Netty中的所有I / O操作都是异步的。 这意味着任何I / O调用都将立即返回，而不能保证所请求的I / O操作已在调用结束时完成。 相反，将返回一个带有ChannelFuture实例的实例，该实例将在请求的I / O操作成功，失败或取消时通知您

通道是分层的

频道可以具有父项，具体取决于其创建方式。 例如，ServerSocketChannel接受的SocketChannel将返回ServerSocketChannel作为其parent（）的父级。

层次结构的语义取决于Channel所属的传输实现。 例如，您可以编写一个新的Channel实现，以创建共享一个套接字连接的子通道，就像BEEP和SSH一样。

下调访问特定于传输的操作

某些传输公开了特定于该传输的其他操作。 将Channel向下转换为子类型以调用此类操作。 例如，对于旧的I / O数据报传输，DatagramChannel提供了多播加入/离开操作。

释放资源

一旦完成Channel，调用ChannelOutboundInvoker.close（）或ChannelOutboundInvoker.close（ChannelPromise）释放所有资源非常重要。 这样可以确保以适当的方式（即文件句柄）释放所有资源。

基本的 I/O 操作（bind()、connect()、read()和 write()）依赖于底层网络传输所提供的原语。在基于 Java 的网络编程中，其基本的构造是 class Socket。Netty 的 Channel 接口所提供的 API，大大地降低了直接使用 Socket 类的复杂性

Channel 也是拥有许多预定义的、专门化实现的广泛类层次结构的根，下面是一个简短的部分清单:

• EmbeddedChannel

• LocalServerChannel

• NioDatagramChannel

• NioSctpChannel

• NioSocketChannel

EventLoop 接口

EventLoop 定义了 Netty 的核心抽象，用于处理连接的生命周期中所发生的事件

关系图示:

图4.png

• 一个 EventLoopGroup 包含一个或者多个 EventLoop；

• 一个 EventLoop 在它的生命周期内只和一个 Thread 绑定；

• 所有由 EventLoop 处理的 I/O 事件都将在它专有的 Thread 上被处理；

• 一个 Channel 在它的生命周期内只注册于一个 EventLoop；

• 一个 EventLoop 可能会被分配给一个或多个 Channel。

ChannelFuture接口

public interface ChannelFuture extends Future<java.lang.Void>

文档:

异步通道I / O操作的结果

Netty中的所有I / O操作都是异步的。 这意味着任何I / O调用都将立即返回，而不能保证所请求的I / O操作已在调用结束时完成。 相反，将返回一个ChannelFuture实例，该实例为您提供有关I / O操作的结果或状态的信息。

ChannelFuture未完成或已完成。 I / O操作开始时，将创建一个新的将来对象。 新的future最初并未完成-因为I / O操作尚未完成，所以既不会成功，失败也不会取消。 如果I / O操作成功完成，失败或通过取消完成，则将来标记为已完成，并带有更多特定信息，例如失败原因。 请注意，即使失败和取消也属于完成状态。

提供了各种方法，可让您检查I / O操作是否已完成，等待完成以及检索I / O操作的结果。 它还允许您添加ChannelFutureListeners，以便在I / O操作完成时得到通知。

使用addListener（GenericFutureListener）而不是await（）

建议在可能的情况下，首选addListener（GenericFutureListener）而不是await（），以便在完成I / O操作并执行任何后续任务时得到通知。

addListener（GenericFutureListener）是非阻塞的。 它只是将指定的ChannelFutureListener添加到ChannelFuture，并且与将来关联的I / O操作完成时，I / O线程将通知侦听器。 ChannelFutureListener完全不阻塞，因此可产生最佳的性能和资源利用率，但是如果您不习惯事件驱动的编程，则实现顺序逻辑可能会比较棘手。

相反，await（）是阻塞操作。 一旦被调用，调用者线程将阻塞直到操作完成。 使用await（）实现顺序逻辑比较容易，但是调用者线程会不必要地阻塞，直到完成I / O操作并且线程间通知的成本相对较高为止。 此外，在特定情况下还可能出现死锁，这将在下面进行描述。

不要在ChannelHandler中调用await（）

ChannelHandler中的事件处理程序方法通常由I / O线程调用。 如果await（）是由I / O线程调用的事件处理程序方法调用的，则它正在等待的I / O操作可能永远不会完成，因为await（）会阻塞它正在等待的I / O操作， 这是一个死锁。

// // BAD - NEVER DO THIS
//   @Override
//  public void channelRead(ChannelHandlerContext ctx, Object msg) {
//      ChannelFuture future = ctx.channel().close();
//      future.awaitUninterruptibly();
//      // Perform post-closure operation
//      // ...
//  }

//  // GOOD
//   @Override
//  public void channelRead(ChannelHandlerContext ctx, Object msg) {
//      ChannelFuture future = ctx.channel().close();
//      future.addListener(new ChannelFutureListener() {
//          public void operationComplete(ChannelFuture future) {
//              // Perform post-closure operation
//              // ...
//          }
//      });
//  }

尽管有上述缺点，但是在某些情况下，调用await（）更方便。 在这种情况下，请确保不要在I / O线程中调用await（）。 否则，将引发BlockingOperationException来防止死锁。

不要混淆I / O超时并等待超时

您用Future.await（long），Future.await（long，TimeUnit），Future.awaitUninterruptible（long）或Future.awaitUninterruptible（long，TimeUnit）指定的超时值根本与I / O超时无关。 如果I / O操作超时，则将来将被标记为“失败完成”，如上图所示。 例如，应通过特定于传输的选项配置连接超时：

// BAD - NEVER DO THIS
//  Bootstrap b = ...;
//  ChannelFuture f = b.connect(...);
//  f.awaitUninterruptibly(10, TimeUnit.SECONDS);
//  if (f.isCancelled()) {
//      // Connection attempt cancelled by user
//  } else if (!f.isSuccess()) {
//      // You might get a NullPointerException here because the future
//      // might not be completed yet.
//      f.cause().printStackTrace();
//  } else {
//      // Connection established successfully
//  }

//  // GOOD
//  Bootstrap b = ...;
//  // Configure the connect timeout option.
//  b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000);
//  ChannelFuture f = b.connect(...);
//  f.awaitUninterruptibly();

//  // Now we are sure the future is completed.
//  assert f.isDone();

//  if (f.isCancelled()) {
//      // Connection attempt cancelled by user
//  } else if (!f.isSuccess()) {
//      f.cause().printStackTrace();
//  } else {
//      // Connection established successfully
//  }

图示源自文档:

图1.png

ChannelHandler接口

public interface ChannelHandler

处理I / O事件或拦截I / O操作，并将其转发到其ChannelPipeline中的下一个处理程序。

子类型

ChannelHandler本身不提供许多方法，但是通常必须实现其子类型之一：

• ChannelInboundHandler处理入站I / O事件，以及

• ChannelOutboundHandler处理出站I / O操作。

另外，为了您的方便，提供了以下适配器类：

• ChannelInboundHandlerAdapter处理入站I / O事件，

• ChannelOutboundHandlerAdapter来处理出站I / O操作

• ChannelDuplexHandler处理入站和出站事件

上下文对象

ChannelHandler随ChannelHandlerContext对象一起提供。 ChannelHandler应该通过上下文对象与其所属的ChannelPipeline进行交互。 使用上下文对象，ChannelHandler可以在上游或下游传递事件，动态修改管道或存储特定于处理程序的信息（使用AttributeKeys）。

状态管理

ChannelHandler通常需要存储一些状态信息。 推荐的最简单方法是使用成员变量：

//  public interface Message {
//      // your methods here
//  }

//  public class DataServerHandler extends SimpleChannelInboundHandler<Message> {

//      private boolean loggedIn;

//       @Override
//      public void channelRead0(ChannelHandlerContext ctx, Message message) {
//          if (message instanceof LoginMessage) {
//              authenticate((LoginMessage) message);
//              loggedIn = true;
//          } else (message instanceof GetDataMessage) {
//              if (loggedIn) {
//                  ctx.writeAndFlush(fetchSecret((GetDataMessage) message));
//              } else {
//                  fail();
//              }
//          }
//      }
//      ...
//  }

因为处理程序实例具有专用于一个连接的状态变量，所以您必须为每个新通道创建一个新的处理程序实例，以避免竞争状态，未经身份验证的客户端可以获取机密信息：

// Create a new handler instance per channel.
 // See ChannelInitializer.initChannel(Channel).
//  public class DataServerInitializer extends ChannelInitializer<Channel> {
//       @Override
//      public void initChannel(Channel channel) {
//          channel.pipeline().addLast("handler", new DataServerHandler());
//      }
//  }

使用AttributeKeys

尽管建议使用成员变量来存储处理程序的状态，但是由于某些原因，您可能不想创建许多处理程序实例。 在这种情况下，可以使用ChannelHandlerContext提供的AttributeKeys：

//  public interface Message {
//      // your methods here
//  }

//   @Sharable
//  public class DataServerHandler extends SimpleChannelInboundHandler<Message> {
//      private final AttributeKey<Boolean> auth =
//            AttributeKey.valueOf("auth");

//       @Override
//      public void channelRead(ChannelHandlerContext ctx, Message message) {
//          Attribute<Boolean> attr = ctx.attr(auth);
//          if (message instanceof LoginMessage) {
//              authenticate((LoginMessage) o);
//              attr.set(true);
//          } else (message instanceof GetDataMessage) {
//              if (Boolean.TRUE.equals(attr.get())) {
//                  ctx.writeAndFlush(fetchSecret((GetDataMessage) o));
//              } else {
//                  fail();
//              }
//          }
//      }
//      ...
//  }

现在，处理程序的状态已附加到ChannelHandlerContext上，您可以将相同的处理程序实例添加到不同的管道中：

//  public class DataServerInitializer extends ChannelInitializer<Channel> {

//      private static final DataServerHandler SHARED = new DataServerHandler();

//       @Override
//      public void initChannel(Channel channel) {
//          channel.pipeline().addLast("handler", SHARED);
//      }
//  }

ChannelPipeline接口

public interface ChannelPipeline
extends ChannelInboundInvoker, ChannelOutboundInvoker, java.lang.Iterable<java.util.Map.Entry<java.lang.String,ChannelHandler>>

建立管道

每个通道都有其自己的管道，并且在创建新通道时会自动创建它

事件如何在管道中流动

下图描述了ChannelPipeline中的ChannelHandler通常如何处理I / O事件。 I / O事件由ChannelInboundHandler或ChannelOutboundHandler处理，并通过调用ChannelHandlerContext中定义的事件传播方法（例如ChannelHandlerContext.fireChannelRead（Object）和ChannelOutboundInvoker.write（Object））转发到其最近的处理程序。

图2.png

入站事件由入站处理程序按自下而上的方向进行处理，如图中左侧所示。 入站处理程序通常处理图底部的I / O线程生成的入站数据。 通常通过实际的输入操作（例SocketChannel.read（ByteBuffer））从远程对等方读取入站数据。 如果入站事件超出了顶部入站处理程序的范围，则将其静默丢弃，或者在需要引起注意时记录下来。

出站事件由出站处理程序按自上而下的方向进行处理，如图中右侧所示。 出站处理程序通常会生成或转换出站流量，例如写请求。 如果出站事件超出了底部出站处理程序，则由与通道关联的I / O线程处理。 I / O线程通常执行实际的输出操作，例如SocketChannel.write（ByteBuffer）

例如，假设我们创建了以下管道：

 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());

在上面的示例中，其名称以Inbound开头的类表示它是一个入站处理程序。 名称以Outbound开头的类表示它是一个出站处理程序。

在给定的示例配置中，事件进入时，处理程序评估顺序为1、2、3、4、5。 当事件出站时，顺序为5、4、3、2、1。最重要的是，ChannelPipeline跳过对某些处理程序的求值以缩短堆栈深度：

• 3和4没有实现ChannelInboundHandler，因此入站事件的实际评估顺序为：1、2和5。

• 1和2没有实现ChannelOutboundHandler，因此出站事件的实际评估顺序为：5、4和3。

• 如果5同时实现ChannelInboundHandler和ChannelOutboundHandler，则入站和出站事件的评估顺序可能分别为125和543。

将事件转发到下一个处理程序

如您在图中所示，您可能会注意到，处理程序必须调用ChannelHandlerContext中的事件传播方法，以将事件转发到其下一个处理程序。 这些方法包括：

入站事件传播方法：

ChannelHandlerContext.fireChannelRegistered()
ChannelHandlerContext.fireChannelActive()
ChannelHandlerContext.fireChannelRead(Object)
ChannelHandlerContext.fireChannelReadComplete()
ChannelHandlerContext.fireExceptionCaught(Throwable)
ChannelHandlerContext.fireUserEventTriggered(Object)
ChannelHandlerContext.fireChannelWritabilityChanged()
ChannelHandlerContext.fireChannelInactive()
ChannelHandlerContext.fireChannelUnregistered()

出站事件传播方法：

ChannelOutboundInvoker.bind(SocketAddress, ChannelPromise)
ChannelOutboundInvoker.connect(SocketAddress, SocketAddress, ChannelPromise)
ChannelOutboundInvoker.write(Object, ChannelPromise)
ChannelHandlerContext.flush()
ChannelHandlerContext.read()
ChannelOutboundInvoker.disconnect(ChannelPromise)
ChannelOutboundInvoker.close(ChannelPromise)
ChannelOutboundInvoker.deregister(ChannelPromise)

以下示例说明了事件传播通常是如何完成的：

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

 public class MyOutboundHandler extends ChannelOutboundHandlerAdapter {
      @Override
     public void close(ChannelHandlerContext ctx, ChannelPromise promise) {
         System.out.println("Closing ..");
         ctx.close(promise);
     }
 }

建立管道

假定用户在管道中具有一个或多个ChannelHandler，以接收I / O事件（例如，读取）并请求I / O操作（例如，写入和关闭）。 例如，典型的服务器在每个通道的管道中将具有以下处理程序，但是您的里程可能会根据协议和业务逻辑的复杂性和特征而有所不同：

• Protocol Decoder-将二进制数据（例如ByteBuf）转换为Java对象

• Protocol Encoder-将Java对象转换为二进制数据

• Business Logic Handler- 执行实际的业务逻辑（例如数据库访问）。

它可以表示为以下示例所示：

static final EventExecutorGroup group = new DefaultEventExecutorGroup(16);
 ...

 ChannelPipeline pipeline = ch.pipeline();

 pipeline.addLast("decoder", new MyProtocolDecoder());
 pipeline.addLast("encoder", new MyProtocolEncoder());

 // Tell the pipeline to run MyBusinessLogicHandler's event handler methods
 // in a different thread than an I/O thread so that the I/O thread is not blocked by
 // a time-consuming task.
 // If your business logic is fully asynchronous or finished very quickly, you don't
 // need to specify a group.
 pipeline.addLast(group, "handler", new MyBusinessLogicHandler());

线程安全

由于ChannelPipeline是线程安全的，因此可以随时添加或删除ChannelHandler。 例如，您可以在即将交换敏感信息时插入加密处理程序，并在交换后将其删除。

ChannelPipeline 提供了 ChannelHandler 链的容器，并定义了用于在该链上传播入站和出站事件流的 API。当 Channel 被创建时，它会被自动地分配到它专属的ChannelPipeline

ChannelHandler 安装到 ChannelPipeline 中的过程如下所示：

• 一个ChannelInitializer的实现被注册到了ServerBootstrap中

• 当 ChannelInitializer.initChannel()方法被调用时，ChannelInitializer
  将在 ChannelPipeline 中安装一组自定义的 ChannelHandler；

• ChannelInitializer 将它自己从 ChannelPipeline 中移除

ChannelPipeline 和 ChannelHandler 之间的共生关系

• ChannelHandler 是专为支持广泛的用途而设计的，可以将它看作是处理往来 Channel-Pipeline 事件（包括数据）的任何代码的通用容器

• 使得事件流经 ChannelPipeline 是 ChannelHandler 的工作，它们是在应用程序的初始化或者引导阶段被安装的

这些对象接收事件、执行它们所实现的处理逻辑，并将数据传递给链中的下一个 ChannelHandler。它们的执行顺序是由它们被添加的顺序所决定的。实际上，被我们称为 ChannelPipeline 的是这些 ChannelHandler 的编排顺序

当ChannelHandler被添加到ChannelPipeline时，它将会被分配一个ChannelHandler-
Context，其代表了 ChannelHandler 和 ChannelPipeline 之间的绑定。虽然这个对象可
以被用于获取底层的 Channel，但是它主要还是被用于写出站数据

在Netty中，有两种发送消息的方式。你可以直接写到Channel中，也可以 写到和Channel-
Handler相关联的ChannelHandlerContext对象中。前一种方式将会导致消息从Channel-
Pipeline 的尾端开始流动，而后者将导致消息从 ChannelPipeline 中的下一个 Channel-
Handler 开始流动。

ChannelInitializer

public abstract class ChannelInitializer<C extends Channel>
extends ChannelInboundHandlerAdapter

特殊的ChannelInboundHandler，提供了一种轻松的方法来将Channel注册到其EventLoop后对其进行初始化。 实现通常在AbstractBootstrap.handler（ChannelHandler），AbstractBootstrap.handler（ChannelHandler）和ServerBootstrap.childHandler（ChannelHandler）的上下文中使用，以设置通道的ChannelPipeline。

public class MyChannelInitializer extends ChannelInitializer {
     public void initChannel(Channel channel) {
         channel.pipeline().addLast("myHandler", new MyHandler());
     }
 }

 ServerBootstrap bootstrap = ...;
 ...
 bootstrap.childHandler(new MyChannelInitializer());
 ...

请注意，此类已标记为ChannelHandler.Sharable，因此必须安全地重用该实现。

ChannelHandlerContext

public interface ChannelHandlerContext
extends AttributeMap, ChannelInboundInvoker, ChannelOutboundInvoker

使ChannelHandler与其ChannelPipeline和其他处理程序进行交互。 处理程序除其他外，可以通知ChannelPipeline中的下一个ChannelHandler以及动态修改其所属的ChannelPipeline。

通知

您可以通过调用此处提供的各种方法之一来通知同一ChannelPipeline中最接近的处理程序。 请参考ChannelPipeline以了解事件的流向。

修改管道

您可以通过调用pipeline（）来获取处理程序所属的ChannelPipeline。 一个非平凡的应用程序可以在运行时动态地在管道中插入，删除或替换处理程序。、

检索以备后用

您可以保留ChannelHandlerContext供以后使用，例如在处理程序方法外部触发事件，即使是从其他线程触发也是如此。

 public class MyHandler extends ChannelDuplexHandler {

     private ChannelHandlerContext ctx;

     public void beforeAdd(ChannelHandlerContext ctx) {
         this.ctx = ctx;
     }

     public void login(String username, password) {
         ctx.write(new LoginMessage(username, password));
     }
     ...
 }

存储状态信息

attr（AttributeKey）允许您存储和访问与处理程序及其上下文相关的有状态信息。 请参考ChannelHandler来学习各种推荐的管理状态信息的方法。

一个处理程序可以具有多个上下文

请注意，可以将ChannelHandler实例添加到多个ChannelPipeline中。 这意味着一个ChannelHandler实例可以具有多个ChannelHandlerContext，因此，如果将一个实例多次添加到一个或多个ChannelPipelines中，则可以使用不同的ChannelHandlerContext调用该实例。

例如，以下处理程序将具有与添加到管道中的次数一样多的独立AttributeKey，无论它是多次添加到同一管道还是多次添加到不同的管道：

 public class FactorialHandler extends ChannelInboundHandlerAdapter {

   private final AttributeKey<Integer> counter = AttributeKey.valueOf("counter");

   // This handler will receive a sequence of increasing integers starting
   // from 1.
    @Override
   public void channelRead(ChannelHandlerContext ctx, Object msg) {
     Integer a = ctx.attr(counter).get();

     if (a == null) {
       a = 1;
     }

     attr.set(a * (Integer) msg);
   }
 }

 // Different context objects are given to "f1", "f2", "f3", and "f4" even if
 // they refer to the same handler instance.  Because the FactorialHandler
 // stores its state in a context object (using an AttributeKey), the factorial is
 // calculated correctly 4 times once the two pipelines (p1 and p2) are active.
 FactorialHandler fh = new FactorialHandler();

 ChannelPipeline p1 = Channels.pipeline();
 p1.addLast("f1", fh);
 p1.addLast("f2", fh);

 ChannelPipeline p2 = Channels.pipeline();
 p2.addLast("f3", fh);
 p2.addLast("f4", fh);

引导

Netty 的引导类为应用程序的网络层配置提供了容器，这涉及将一个进程绑定到某个指定的端口，或者将一个进程连接到另一个运行在某个指定主机的指定端口上的进程

有两种类型的引导：一种用于客户端（简单地称为 Bootstrap），而另一种（ServerBootstrap）用于服务器。无论你的应用程序使用哪种协议或者处理哪种类型的数据，唯一决定它使用哪种引导类的是它是作为一个客户端还是作为一个服务器

比较 Bootstrap 类:

区别:

• ServerBootstrap 将绑定到一个端口，因为服务器必须要监听连接，而 Bootstrap 则是由想要连接到远程节点的客户端应用程序所使用的

• 引导一个客户端只需要一个 EventLoopGroup，但是一个ServerBootstrap 则需要两个

解释第二个区别:

因为服务器需要两组不同的 Channel。第一组将只包含一个 ServerChannel，代表服务器自身的已绑定到某个本地端口的正在监听的套接字。而第二组将包含所有已创建的用来处理传入客户端连接（对于每个服务器已经接受的连接都有一个）的 Channel。图 3-4 说明了这个模型，并且展示了为何需要两个不同的 EventLoopGroup。

图示(具有两个 EventLoopGroup 的服务器):

图3.png

与 ServerChannel 相关联的 EventLoopGroup 将分配一个负责为传入连接请求创建Channel 的 EventLoop。一旦连接被接受，第二个 EventLoopGroup 就会给它的 Channel分配一个 EventLoop

参考

<< Netty实战 >>