Netty 的 Channel、Promise、Pipeline 详解
- 2019 年 11 月 11 日
- 筆記
Netty Demo 示例
首先通过一个示例来分析，创建一个 NioServerSocketChannel 监听本机端口 11111 的 Socket 连接，将收到的消息原样返回；然后再创建一个 NioSocketChannel，发起对本机的 11111 端口的 Socket 连接，发送字符串 ”Netty rocks!“。预期能收到服务端返回的 “Netty rocks!” 响应。
Maven 依赖
本文使用的 Netty 版本是 5.0.0.Alpha2，与 4.x 版本相比变化还是挺大的。pom 文件添加：
<dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>5.0.0.Alpha2</version> </dependency>
创建一个 Server
创建一个 NioServerSocketChannel，监听本机端口 11111 的 Socket 连接。
public class EchoServer { private final int port; public EchoServer(int port) { this.port = port; } public static void main(String[] args) throws InterruptedException { new EchoServer(11111).start(); } public void start() throws InterruptedException { final EchoServerHandler serverHandler = new EchoServerHandler(); NioEventLoopGroup group = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(group).channel(NioServerSocketChannel.class) .localAddress(new InetSocketAddress(port)) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { socketChannel.pipeline().addLast(serverHandler); } }); ChannelFuture channelFuture = b.bind().sync(); channelFuture.channel().closeFuture().sync(); } finally { group.shutdownGracefully().sync(); } } }
EchoServerHandler 的实现如下，在 channelRead 时将数据写入 ChannelHandlerContext，并将数据输出到控制台。
@ChannelHandler.Sharable public class EchoServerHandler extends ChannelHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ByteBuf in = (ByteBuf) msg; System.out.println("Server received : " + in.toString(CharsetUtil.UTF_8)); ctx.write(in); } @Override public void channelReadComplete(ChannelHandlerContext ctx) { ctx.writeAndFlush(Unpooled.EMPTY_BUFFER) .addListener(ChannelFutureListener.CLOSE); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { cause.printStackTrace(); ctx.close(); } }
创建一个 Client
创建一个 NioSocketChannel，发起对本机的 11111 端口的 Socket 连接。
public class EchoClient { private final String host; private final int port; public EchoClient(String host, int port) { this.host = host; this.port = port; } public static void main(String[] args) throws InterruptedException { new EchoClient("localhost", 11111).start(); } public void start() throws InterruptedException { NioEventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap bootstrap = new Bootstrap(); bootstrap.group(group).channel(NioSocketChannel.class) .remoteAddress(new InetSocketAddress(host, port)) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { socketChannel.pipeline().addLast(new EchoClientHandler()); } }); ChannelFuture channelFuture = bootstrap.connect().sync(); channelFuture.channel().closeFuture().sync(); }finally { group.shutdownGracefully().sync(); } } }
EchoClientHandler 的实现如下，messageReceived（在 Netty 4.x 为 channelRead0）对于泛型 I（本例中是 ByteBuf）进行处理，将数据输出到控制台。
@ChannelHandler.Sharable public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> { @Override public void channelActive(ChannelHandlerContext ctx) { ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!", CharsetUtil.UTF_8)); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { cause.printStackTrace(); ctx.close(); } @Override public void messageReceived(ChannelHandlerContext ctx, ByteBuf msg) throws Exception { System.out.println( "Client received: " + msg.toString(CharsetUtil.UTF_8)); } }
io.netty.channel.Channel 类
上面 demo 中直接使用到的 2 个类是：NioServerSocketChannel 和 NioSocketChannel，这两个类底层都是实现了 Channel 接口，注意这个 Channel 接口是 io.netty.channel.Channel，而不是 JDK 自带的 java.nio.channels.Channel！两个类的继承关系如下：
NioServerSocketChannel
NioSocketChannel
Channel 提供应用程序网络套接字或其他组件连接，提供读、写、连接和绑定等 I/O 操作。
- Channel 的当前状态（开启、关闭）
- Channel 的配置参数（接收缓冲区大小）
- I/O 操作（读、写、连接、绑定）
- ChannelPipeline，处理所有与 Channel 绑定的 I/O 事件和请求
所有 I/O 操作都是异步的
Netty 中所有 I/O 操作都是异步的。这意味着所有的 I/O 调用都会立即返回，不能保证在调用结束时请求的 I/O 操作是否完成。调用者会得到一个 ChannelFuture 实例，该实例会在请求的 I/O 操作成功、失败、取消时通知调用者。
Channel 是分层级的
Channel 可以有 parent，这取决于 Channel 的创建方式。例如被 ServerSocketChannel 接收的 SocketChannel，会得到一个 ServerSocketChannel 作为它的 parent。
释放资源
使用完毕后，调用 close() 或 close(ChannelPromise) 释放资源非常重要。
ChannelFuture
ChannelFuture 是一个异步 Channel I/O 操作的结果。如上面所说，Netty 中所有 I/O 操作都是异步的。这意味着所有的 I/O 调用都会立即返回，不能保证在调用结束时请求的 I/O 操作是否完成。调用者会得到一个 ChannelFuture 实例。
ChannelFuture 只有 2 种状态：未完成、已完成。I/O 操作开始时，将会创建一个新的 ChannelFuture 对象，初始时是未完成状态 —— 不是成功、失败或取消的任何一种状态，因为 I/O 操作还没有完成。如果 I/O 操作结束（无论成功、失败、取消），ChannelFuture 都会处于完成状态。注意即使是失败也属于完成状态。
+---------------------------+ | Completed successfully | +---------------------------+ +----> isDone() = true | +--------------------------+ | | isSuccess() = true | | Uncompleted | | +===========================+ +--------------------------+ | | Completed with failure | | isDone() = false | | +---------------------------+ | isSuccess() = false |----+----> isDone() = true | | isCancelled() = false | | | cause() = non-null | | cause() = null | | +===========================+ +--------------------------+ | | Completed by cancellation | | +---------------------------+ +----> isDone() = true | | isCancelled() = true | +---------------------------+
我们还可以添加 ChannelFutureListener，以便在 I/O 操作完成时收到通知。
使用 addListener(GenericFutureListener) 而不是 await()
addListener(GenericFutureListener) 是非阻塞的，只需要将特定的 ChannelFutureListener 添加到ChannelFuture 即可，I/O 线程会在 ChannelFuture 绑定的 I/O 操作完成时通知监听器。ChannelFutureListener 完全非阻塞，因此效率极高。
而 await() 是阻塞操作，一旦调用，调用者线程就会阻塞直到操作完成。使用 await() 操作更容易，但是成本更高。此外，在特定的情况下还可能出现死锁。
使用 ChannelHandler 而不是 await()
ChannelHandler 中的事件处理方法通常由 I/O 线程调用，如果 await() 是由事件处理方法（I/O 线程）调用的，那么它正在等待的 I/O 操作可能永远也不会完成，因为 await() 方法可以阻止它正在等待的 I/O 操作，也就是发生了死锁。
// BAD - NEVER DO THIS @Override public void channelRead(ChannelHandlerContext ctx, GoodByeMessage msg) { ChannelFuture future = ctx.channel().close(); future.awaitUninterruptibly(); // Perform post-closure operation // ... } // GOOD @Override public void channelRead(ChannelHandlerContext ctx, GoodByeMessage msg) { ChannelFuture future = ctx.channel().close(); future.addListener(new ChannelFutureListener() { public void operationComplete(ChannelFuture future) { // Perform post-closure operation // ... } }); }
创建 Channel
在 Bootstrap（客户端） 和 ServerBootstrap（服务端） 的启动过程中都会调用 AbstractBootstrap#channel(…) 方法（参考文章开头的 Demo）：
public B channel(Class<? extends C> channelClass) { if (channelClass == null) { throw new NullPointerException("channelClass"); } return channelFactory(new ReflectiveChannelFactory<C>(channelClass)); }
对于上一行的 return 语句，首先看里面的 ReflectiveChannelFactory 对象，它是一个 ChannelFactory，通过反射调用对应 Class 的默认构造函数来实例化新的 Channel。其定义如下：
public class ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T> { private final Class<? extends T> clazz; ... }
clazz 是 Channel 的子类，其中 newChannel 方法仅仅是调用的 Class 的 newInstance() 方法。
再来看 return 语句中的 channelFactory 方法：
public B channelFactory(io.netty.channel.ChannelFactory<? extends C> channelFactory) { return channelFactory((ChannelFactory<C>) channelFactory); }
下面代码可以看出，就是将上面通过 ReflectiveChannelFactory 创建出来的 channelFactory 赋值到对应字段。注意此时并没有创建 Channel，而是在：
- 对于 NioSocketChannel，由于它充当客户端的功能，它的创建时机在 connect(…) 的时候；
- 对于 NioServerSocketChannel 来说，它充当服务端功能，它的创建时机在绑定端口 bind(…) 的时候。
public B channelFactory(ChannelFactory<? extends C> channelFactory) { if (channelFactory == null) { throw new NullPointerException("channelFactory"); } if (this.channelFactory != null) { throw new IllegalStateException("channelFactory set already"); } this.channelFactory = channelFactory; return (B) this; }
接下来，我们看下 ServerBootstrap 是如何创建 NioServerSocketChannel 的，以及 NioSocketChannel 是如何与 JDK 交互的。
public ChannelFuture bind() { validate(); SocketAddress localAddress = this.localAddress; if (localAddress == null) { throw new IllegalStateException("localAddress not set"); } return doBind(localAddress); }
首先是 validate() 方法，验证 group 和 channelFactory 不能为 null，否则会抛出异常。
public B validate() { if (group == null) { throw new IllegalStateException("group not set"); } if (channelFactory == null) { throw new IllegalStateException("channel or channelFactory not set"); } return (B) this; }
然后是验证 localAddress 不为空，之后就是核心的 doBind() 逻辑了。
private ChannelFuture doBind(final SocketAddress localAddress) { final ChannelFuture regFuture = initAndRegister(); 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.executor = channel.eventLoop(); } doBind0(regFuture, channel, localAddress, promise); } }); return promise; } }
首先就是 initAndRegister()，返回一个注册的 ChannelFuture，通过它来获取 Channel。
final ChannelFuture initAndRegister() { final Channel channel = channelFactory().newChannel(); try { init(channel); } catch (Throwable t) { 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); } ChannelFuture regFuture = group().register(channel); if (regFuture.cause() != null) { if (channel.isRegistered()) { channel.close(); } else { channel.unsafe().closeForcibly(); } } return regFuture; }
其中比较重要的是 init(channel)，它有 ServerBootstrap 和 Bootstrap 两个实现，这里就不细致展开。
还有一个重要的语句是：ChannelFuture regFuture = group().register(channel);，其中 group() 就是 ServerBootstrap 或 Bootstrap 的 EventLoopGroup 成员变量，每个 Bootstrap 还有一个 private volatile EventLoopGroup childGroup; 。
io.netty.channel.AbstractChannel.AbstractUnsafe#register 方法实现如下，
@Override public final void register(EventLoop eventLoop, final ChannelPromise promise) { ... // 需要重用 eventloop 对象，否则用户就会得到具有不同状态的多个对象 if (AbstractChannel.this.eventLoop == null) { AbstractChannel.this.eventLoop = new PausableChannelEventLoop(eventLoop); } else { AbstractChannel.this.eventLoop.unwrapped = eventLoop; } // 当前线程是否被 event loop 执行 if (eventLoop.inEventLoop()) { register0(promise); } else { try { eventLoop.execute(new OneTimeTask() { @Override public void run() { register0(promise); } }); } catch (Throwable t) { ... } } }
NioSocketChannel
NioSocketChannel 的无参构造参数：
public NioSocketChannel() { this(DEFAULT_SELECTOR_PROVIDER); } public NioSocketChannel(SelectorProvider provider) { // newSocket(provider) 方法会创建 JDK 的 SocketChannel this(newSocket(provider)); }
其最终会调用：io.netty.channel.nio.AbstractNioChannel#AbstractNioChannel
protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) { // 客户端关心 OP_READ 事件，等待读取服务端返回数据 super(parent, ch, SelectionKey.OP_READ); } protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) { super(parent); this.ch = ch; this.readInterestOp = readInterestOp; try { // 将 SelectableChannel 配置为「非阻塞」模式 ch.configureBlocking(false); } catch (IOException e) { ... } }
ServerSocketChannel
ServerSocketChannel 是一个 TCP/IP ServerChannel，处理 TCP/IP 连接请求。
NioServerSocketChannel
NioServerSocketChannel 是一个 ServerSocketChannel 的实现，基于 NIO 选择器来接收新的连接。
ChannelPipeline
ChannelHandler 列表处理和拦截 Channel 的传入事件和传出操作。ChannelPipeline 是 Intercepting Filter 模式的扩展，用户可以完全控制事件的处理方式和管道中 ChannelHandlers 如何交互。
Pipeline 创建
每个 Channel 都有自己的 Pipeline，并且在创建 Channel 时会自动创建 Pipeline。
事件是如何在 Pipeline 中传递的
下图描述了 ChannelPipeline 是如何处理 ChannelHandler 的 I/O 事件的。 I/O 事件由 ChannelInboundHandler 或 ChannelOutboundHandler 处理，并通过调用 ChannelHandlerContext 中定义的事件传播方法（例如 ChannelHandlerContext.fireChannelRead(Object) 和 ChannelOutboundInvoker.write(Object)）转发到其最近的 handler。
I/O Request via Channel or ChannelHandlerContext | +---------------------------------------------------+---------------+ | ChannelPipeline | | | |/ | | +----------------------------------------------+----------+ | | | ChannelHandler N | | | +----------+-----------------------------------+----------+ | | /| | | | | |/ | | +----------+-----------------------------------+----------+ | | | ChannelHandler N-1 | | | +----------+-----------------------------------+----------+ | | /| . | | . . | | ChannelHandlerContext.fireIN_EVT() ChannelHandlerContext.OUT_EVT()| | [method call] [method call] | | . . | | . |/ | | +----------+-----------------------------------+----------+ | | | ChannelHandler 2 | | | +----------+-----------------------------------+----------+ | | /| | | | | |/ | | +----------+-----------------------------------+----------+ | | | ChannelHandler 1 | | | +----------+-----------------------------------+----------+ | | /| | | +---------------+-----------------------------------+---------------+ | |/ +---------------+-----------------------------------+---------------+ | | | | | [ Socket.read() ] [ Socket.write() ] | | | | Netty Internal I/O Threads (Transport Implementation) | +-------------------------------------------------------------------+
比如下面的例子，以 Inbound 开头的类表示是入站处理程序，以 Outbound 开头的类表示是出站处理程序。
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());
上面的示例配置中，事件进入时处理顺序是1，2，3，4，5；事件出站顺序为5，4，3，2，1。
- 3 和 4 没有实现 ChannelInboundHandler，因此入站事件实际顺序是 1，2，5
- 1 和 2 没有实现 ChannelOutboundHandler，因此出站事件实际顺序是 5，4，3
- 5 同时实现了 ChannelInboundHandler 和 ChannelOutboundHandler
将事件转发到下一个 Handler
处理程序必须调用 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); } }
建立 Pipeline
假定用户在 Pipeline 中具有一个或多个 ChannelHandler，用于处理 I/O 事件。比如：
- 协议解码器：将二进制数据（例如 ByteBuf）转换为 Java 对象。
- 协议编码器：将 Java 对象转换为二进制数据。
- 业务逻辑处理程序：执行实际的业务逻辑（数据库访问）。
一个代码示例：
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());
线程安全性
ChannelHandler 可以在任何时候添加到 ChannelPipeline中，也可以随时从 ChannelPipeline 中移出，它是线程安全的。
EventExecutorGroup
定义如下：
public interface EventExecutorGroup extends java.util.concurrent.ScheduledExecutorService, java.lang.Iterable<EventExecutor>
EventExecutorGroup 顾名思义，就是 EventExecutor 的 group，负责通过其 next() 方法提供要使用的 EventExecutor。除此之外，它还负责处理 EventExecutor 的生命周期，并允许以全局方式关闭它们。
EventExecutor
EventExecutor 是一个特殊的 EventExecutorGroup，它提供一些方便的方法来查看某个线程是否在事件循环中执行。
image
EventLoop
在一个 Channel 注册后，将处理这个 Channel 的所有 I/O 操作。一个 EventLoop 实例通常将处理多个 Channel。
Promise
可写的 Future。
ChannelPromise
可写的 ChannelFuture。下图中上面的 Future 是 JDK 的 Future，下面的 Future 是 Netty 自定义的 Future，具有异步操作的结果。
image
Netty 自定义的 Future 接口其方法如下图，增加了很多方法：
- addListener：添加 listener
- await：等待 future 完成
- sync：等待 future 完成，如果失败会抛出失败原因
