Netty：Netty的介绍以及它的核心组件（一）—— Channel

阅读原文时间：2023年07月08日阅读：3

1. Netty 介绍

　　Netty 是一个无阻塞的输入/输出（NIO）框架，它使开发低级网络服务器和客户端变得相对简单。Netty为需要在套接字级别上工作的开发人员提供了令人难以置信的强大功能，例如，在开发客户端和服务器之间的自定义通信协议时。它支持SSL / TLS，具有阻塞和非阻塞统一API，以及灵活的线程模型。它又快又高效。

　　Netty 的异步、非阻塞 I/O 模型是为高度可扩展的体系结构设计的，与类似的其他阻塞模型相比，它可能允许更高的吞吐量。基本上，非阻塞服务器在单个线程上异步运行所有请求（没有函数会“阻塞”事件循环）。这与阻塞服务器模型形成对比，后者通常在单独的线程上运行每个请求。通过不必在负载增加时切换线程或创建新线程，非阻塞模型可以减少开销，并在流量增加时更快地进行扩展。

　　Netty 的核心组件：Channel、回调、Future、事件和ChannelHandler。

2. Channel 通道

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

　　Channel 为用户提供：

通道的当前状态（例如，它是否打开？是否已连接？）
通道的 ChannelConfig 配置参数（例如，接收缓冲区大小）
通道支持的 I/O 操作（例如，读取，写入，连接和绑定）
处理所有与该通道关联的I/O事件和请求的 ChannelPipeline

　　Netty 的 Channel 接口的一些类型（Java类图）：

　　

　　Channel 主要衍生有这几个类型的接口：
Http2StreamChannel：支持 HTTP/2 协议的 Channel。
ServerChannel：一个接收新连接然后创建子Channel的Channel组件， ServerSocketChannel 就是一个很好的例子， ServerSocketChannel 接受连接，然后创建出 SocketChannel。
SctpChannel：一个处理 SCTP/IP 协议（流控制传输协议 (Stream Control Transmission Protocol)）的 Channel。
DatagramChannel：一个处理 UDP/IP 协议的 Channel。
DuplexChannel：一个拥有两个端点并能在每个端点独立关闭的全双工Channel。

　　Channel 的一些特性：

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

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

2）Channel是分层的

　　Channel可以有一个parent（父级}，具体取决创建方式。例如，ServerSocketChannel接受的SocketChannel将返回ServerSocketChannel作为其在parent上的父级。层次结构的语义取决于Channel所属的transport实现。

3）向下访问特定于传输的操作

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

4）释放资源

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

　　那么 Channel 在什么时候创建？

　　Netty 服务端引导类 ServerBootstrap#bind() 方法中的 doBind() 方法调用 initAndRegister() ，方法具体内容再移步到父类 AbstractBootstrap#initAndRegister 方法处，创建是有 Channel 工厂来创建的，类型是由 Netty 的 BootStrap 实例化时指定的 Channel 类型，例如 NioSocketChannel.class。

1 /**
2 * Create a new {@link Channel} and bind it.
3 */
4 public ChannelFuture bind(SocketAddress localAddress) {
5 validate();
6 if (localAddress == null) {
7 throw new NullPointerException("localAddress");
8 }
9 return doBind(localAddress);
10 }

　　 doBind(localAddress); 具体方法：

1 private ChannelFuture doBind(final SocketAddress localAddress) {
2 final ChannelFuture regFuture = initAndRegister();
3 final Channel channel = regFuture.channel();
4 if (regFuture.cause() != null) {
5 return regFuture;
6 }
7
8 if (regFuture.isDone()) {
9 // At this point we know that the registration was complete and successful.
10 ChannelPromise promise = channel.newPromise();
11 doBind0(regFuture, channel, localAddress, promise);
12 return promise;
13 } else {
14 // Registration future is almost always fulfilled already, but just in case it's not.
15 final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
16 regFuture.addListener(new ChannelFutureListener() {
17 @Override
18 public void operationComplete(ChannelFuture future) throws Exception {
19 Throwable cause = future.cause();
20 if (cause != null) {
21 // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
22 // IllegalStateException once we try to access the EventLoop of the Channel.
23 promise.setFailure(cause);
24 } else {
25 // Registration was successful, so set the correct executor to use.
26 // See https://github.com/netty/netty/issues/2586
27 promise.registered();
28
29 doBind0(regFuture, channel, localAddress, promise);
30 }
31 }
32 });
33 return promise;
34 }
35 }

　　 initAndRegister() 方法通过 ChannelFactory 创建 NioServerSocketChannel。

1 final ChannelFuture initAndRegister() {
2 Channel channel = null;
3 try {
4 channel = channelFactory.newChannel();
5 init(channel);
6 } catch (Throwable t) {
7 if (channel != null) {
8 // channel can be null if newChannel crashed (eg SocketException("too many open files"))
9 channel.unsafe().closeForcibly();
10 // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
11 return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
12 }
13 // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
14 return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
15 }
16
17 ChannelFuture regFuture = config().group().register(channel);
18 if (regFuture.cause() != null) {
19 if (channel.isRegistered()) {
20 channel.close();
21 } else {
22 channel.unsafe().closeForcibly();
23 }
24 }
25
26 // If we are here and the promise is not failed, it's one of the following cases:
27 // 1) If we attempted registration from the event loop, the registration has been completed at this point.
28 // i.e. It's safe to attempt bind() or connect() now because the channel has been registered.
29 // 2) If we attempted registration from the other thread, the registration request has been successfully
30 // added to the event loop's task queue for later execution.
31 // i.e. It's safe to attempt bind() or connect() now:
32 // because bind() or connect() will be executed *after* the scheduled registration task is executed
33 // because register(), bind(), and connect() are all bound to the same thread.
34
35 return regFuture;
36 }

Channel 的四个状态：