一、基本说明
Netty 中的 ChannelPipeline、ChannelHandler 和 ChannelHandlerContext 是非常核心的组件,我们从源码来分析 Netty 是如何设计这三个核心组件的,并分析是如何创建和协调工作的。
二、 三大核心组件介绍
2.1 三者关系
1)、每当 ServerSocket 创建一个新的连接,就会创建一个 Socket,对应的就是目标客户端。
2)、每一个新创建的 Socket 都将会分配一个全新的 ChannelPipeline(简称pipeline)。
3)、每一个 ChannelPipeline 内部都含有多个 ChannelHandlerContext (简称Context)。
4)、它们一起组成了双向链表,这些 Context 用于包装我们调用 addLast 方法时添加的 ChannelHandler(简称handler)
- 从图中可以看出,ChannelSocket 和 ChannelPipeline 是一对一的关联关系,而 pipeline 内部由多个 Context 形成了链表,Context 只是对 handler 的封装。
- 当有一个请求进来的时候,会进入 Socket 对应的 pipeline,并经过 pipeline 所有的 handler,这就是设计模式中的过滤器模式。
2.2 ChannelPipeline 作用及设计
可以看到 ChannelPipeline 接口继承了 Inbound、Outbound、Iterable 接口,表示它可以调用数据出站和入站的方法,同时也能遍历内部的链表,它的几个代表性的方法,基本上都是针对 handler 链表的插入、追加、删除、替换操作,类似是一个 LinkedList。同时,它也能够返回 channel:Channel channel()(也就是 socket)
1)、在 pipeline 接口文档上,提供了一幅图
说明:
- 这是一个 handler 的 list,handler 用于处理或拦截入站事件和出站事件,pipeline 实现了过滤器的高级形式,以便用户控制事件如何处理 handler 在 pipeline 中的交互。
- 该图描述了一个典型的 handler 在 pipeline 中处理 I/O 事件的方式,I/O 事件由 InboundHandler 或者 outboundHandler 处理,并通过调用 ChannelHandlerContext.fireChannelRead 方法转发给其最近的处理程序。
- 如图所示,入站事件由入站处理程序以自下而上的方向处理。入站处理程序通常处理由图底部的 I/O 线程生成的入站数据。入站数据通常从 SocketChannel.read(ByteBuffer) 获取。
- 通常一个 pipeline 有多个 handler,例如,一个典型的服务器在每个通道的管道中都会有一下处理程序:协议解码器(将二进制数据转换为Java对象)、协议编码器(将Java对象转换为二进制数据)、业务逻辑处理程序。
- 注意:业务程序不能将线程阻塞,会影响 IO 的速度,进而影响整个 Netty 程序的性能。如果业务程序很快,就可以放在 IO 线程中,反之,需要异步执行。或者在添加 handler 的时候添加一个线程池,例如:
// 下面这个任务执行的时候,将不会阻塞 IO 线程,执行的线程来自 group 线程池
pipeline.addLast(group,"handler",new MyBusinessLogicHandler());
2.3 ChannelHandler 作用及设计
1)、源码
public interface ChannelHandler {
// 当把 ChannelHandler 添加到 pipeline 时被调用
void handlerAdded(ChannelHandlerContext ctx) throws Exception;
// 当从 pipeline 中移除时被调用
void handlerRemoved(ChannelHandlerContext ctx) throws Exception;
// 当处理过程中在 pipeline 发生异常时调用
@Deprecated
void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;
}
2)、ChannelHandler 的作用就是处理 IO 事件或拦截 IO 事件,并将其转发给下一个处理程序 ChannelHandler 。Handler 处理程序时分入站和出站,两个方向的操作都是不同的,因此,Netty 定义了两个子接口继承 ChannelHandler :ChannelInboundHandler(入站事件接口)和 ChannelOutboundHandler(出站事件接口)
3)、ChannelInboundHandler(入站事件接口)
- channelActive:用于当 Channel 处于活动状态时被调用
- channelRead:当从 Channel 读取数据时被调用
- 我们需要重写里面的方法,当发生关注的事件,需要在对应的方法实现业务逻辑,因为,当事件发生时,Netty 会回调对应的方法
4)、ChannelOutboundHandler(出站事件接口)
- bind:当请求将 Channel 绑定到本地地址时调用
- close:当请求关闭 Channel 时调用
- 出站操作都是一些连接和写出数据类似的方法
5)、ChannelDuplexHandler 处理出站和入站事件
- ChannelDuplexHandler 间接实现了入站接口并且实现了出站接口
- 是一个通用的能够同时处理入站事件和出站事件的类
2.4 ChannelHandlerContext 作用及设计
ChannelHandlerContext 继承了出站方法调用接口和入站方法调用接口
1)、ChannelInboundInvoker 和 ChannelOutboundInvoker 部分源码
这两个invoker 就是针对入站或出站方法来的,就是在 入站或出站 handler 的外层再包装一层,达到在方法前后拦截并做一些特定操作的目的。
2)、ChannelHandlerContext 部分源码
- ChannelHandlerContext 不仅仅继承了ChannelInboundInvoker 和ChannelOutboundInvoker 的方法,同时也定义了一些自己的方法。
- 这些方法能够获取 Context 上下文环境中对应的数据,比如:channel、executor、handler、pipeline、内存分配器、关联的handler是否被删除。
- Context 就是包装了 handler 相关的一切,以方便 Context 可以在 pipeline 中方便的操作 handler
2.5 pipeline、handler、context 创建过程
- 任何一个 ChannelSocket 创建的同时都会创建一个 pipeline
- 当用户或系统内部调用 pipeline 的 add**** 方法添加 handler 时,都会创建一个包装这个 handler 的 Context
- 这些 Context 在 pipeline 中组成了双向链表
1)、Socket 创建的时候创建 pipeline,在 SocketChannel 的抽象父类 AbstractChannel 的构造方法中
protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline(); // 断点
}
通过debug,继续追踪到
protected DefaultChannelPipeline(Channel channel) {
this.channel = ObjectUtil.checkNotNull(channel, "channel");
succeededFuture = new SucceededChannelFuture(channel, null);
voidPromise = new VoidChannelPromise(channel, true);
tail = new TailContext(this);
head = new HeadContext(this);
head.next = tail;
tail.prev = head;
}
- 将 channel 赋值给 channel 字段,用于 pipeline 操作 channel
- 创建了一个 future 和 promise,用于异步回调使用
- 创建一个 inbound 的 TailContext,创建一个既是 inbound 类型又是 outbound 类型的 HeadContext
- 最后,将两个 Context 互相连接,形成双向链表
- TailContext 和 HeadContext 非常重要,所有的 pipeline 中的事件都会流经它们
2)、在 add**** 添加处理器的时候创建 Context,可以看下 DefaultChannelPipeline 的 addLast 方法,代码如下:
@Override
public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
if (handlers == null) {
throw new NullPointerException("handlers");
}
for (ChannelHandler h: handlers) {
if (h == null) {
break;
}
addLast(executor, null, h); // 进入
}
return this;
}
@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;
}
- pipeline 添加 handler,参数是线程池,name 是 null,handler 是自定义或者系统传入的 handler。Netty 为了防止多个线程导致安全问题,同步了这段代码
- 检查这个 handler 实例是否是共享的,如果不是,并且已经被别的 pipeline 使用了,则抛出异常
- 调用 newContext(group, filterName(name, handler), handler) 方法,创建一个 Context。从这里可以看出,每次添加一个 handler 都会创建一个关联 Context
- 调用 addLast 方法,将 Context 追加到链表中
- 如果这个通道还没有注册到 selector 上,就将这个 Context 添加到这个 pipeline 的待办任务中。当注册好了以后,就会调用 callHandlerAdded0() 方法(默认是什么都不做,用户可以实现这个方法)
- 到这里,针对三个对象创建过程,了解的差不多了,和最初说的一样,每当创建 ChannelSocket 的时候都会创建一个绑定的 pipeline,一对一的关系,创建 pipeline 的时候也会创建 tail 节点和 head 节点,形成最初的链表。tail 是入站 inbound 类型的 handler,head 既是 inbound 也是 outbound 类型的 handler。在调用 pipeline 的 addLast 方法的时候,会根据给定的 handler 创建一个 Context,然后,将这个 Context 插入到链表的尾部(tail前面)
版权声明:本文不是「本站」原创文章,版权归原作者所有 | 原文地址: