netty: 4.1.9.Final
核心概念：
- Channel 通道，贯穿一切。
- ChannelHandler 处理通道事件或响应。
- ChannelHandlerInitializer 用来初始化一个ChannelHandler。
- ChannelHandlerContext 务必注意，这个不是在pipeline中通用的上下文，而是仅限于某个ChannelHandler的上下文。
- ChannelPipeline - 每个Channel都有自己的Pipeline；每个Pipeline都有一堆ChannelHandler。实现了拦截过滤器模式。
所以层级关系是这样的： `Channel > ChannelPipeline > ChannelHandler`
其他还有
- ChannelInboundHandler
- ChannelOutboundHandler
- ChannelInboundInvoker
- ChannelOutboundInvoker
## 关于ctx主动触发事件
### 以`fireChannelRegistered`为例，具体流程是：
1. ctx#fireChannelRegistered 当前ctx主动触发事件
2. ctx#findContextInbound 当前ctx找到下一个ctx
3. ctx#invokeChannelRegistered(next)
4. next#invokeChannelRegistered 主动调用下一个ctx的invokeChannelRegistered
1 //ctx主动调用
2 @Override
3 public ChannelHandlerContext fireChannelRegistered() {
4 //先找到下一个ctx - findContextInbound()
5 invokeChannelRegistered(findContextInbound());
6 return this;
7 }
8
9 //核心：next#invokeChannelRegistered
10 static void invokeChannelRegistered(final AbstractChannelHandlerContext next) {
11 EventExecutor executor = next.executor();
12 if (executor.inEventLoop()) {
13 next.invokeChannelRegistered();
14 } else {
15 executor.execute(new Runnable() {
16 @Override
17 public void run() {
18 next.invokeChannelRegistered();
19 }
20 });
21 }
22 }
23
24 //触发handler的回调
25 private void invokeChannelRegistered() {
26 if (invokeHandler()) {
27 try {
28 //这里触发的是 hanlder#channelRegistered
29 ((ChannelInboundHandler) handler()).channelRegistered(this);
30 } catch (Throwable t) {
31 notifyHandlerException(t);
32 }
33 } else {
34 fireChannelRegistered();
35 }
36 }
View Code
### 再以`fireChannelRead`为例，具体流程是：
1 @Override
2 public ChannelHandlerContext fireChannelRead(final Object msg) {
3 //先找到下一个ctx
4 invokeChannelRead(findContextInbound(), msg);
5 return this;
6 }
7
8 //触发下一个ctx的invokeChannelRead
9 static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
10 final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
11 EventExecutor executor = next.executor();
12 if (executor.inEventLoop()) {
13 next.invokeChannelRead(m);
14 } else {
15 executor.execute(new Runnable() {
16 @Override
17 public void run() {
18 next.invokeChannelRead(m);
19 }
20 });
21 }
22 }
23
24 //最终触发的是 handler#channelRead
25 private void invokeChannelRead(Object msg) {
26 if (invokeHandler()) {
27 try {
28 ((ChannelInboundHandler) handler()).channelRead(this, msg);
29 } catch (Throwable t) {
30 notifyHandlerException(t);
31 }
32 } else {
33 fireChannelRead(msg);
34 }
35 }
View Code
## 关于ctx的prev/next
1 private AbstractChannelHandlerContext findContextInbound() {
2 AbstractChannelHandlerContext ctx = this;
3 do {
4 ctx = ctx.next;
5 } while (!ctx.inbound);
6 return ctx;
7 }
8
9 private AbstractChannelHandlerContext findContextOutbound() {
10 AbstractChannelHandlerContext ctx = this;
11 do {
12 ctx = ctx.prev;
13 } while (!ctx.outbound);
14 return ctx;
15 }
View Code
其实ctx是在注册ChannelHandler的时候创建的，同时将其prev设置成之前的ctx。见`DefaultChannelPipeline`，如下：
1 final AbstractChannelHandlerContext head;
2 final AbstractChannelHandlerContext tail;
3 private final Channel channel;
4
5 //构造器中就创建是head、tail
6 protected DefaultChannelPipeline(Channel channel) {
7 this.channel = ObjectUtil.checkNotNull(channel, "channel");
8 succeededFuture = new SucceededChannelFuture(channel, null);
9 voidPromise = new VoidChannelPromise(channel, true);
10
11 tail = new TailContext(this);
12 head = new HeadContext(this);
13
14 head.next = tail;
15 tail.prev = head;
16 }
17
18 //当添加handler的时候，
19 @Override
20 public final ChannelPipeline addFirst(String name, ChannelHandler handler) {
21 return addFirst(null, name, handler);
22 }
23
24 //当添加handler的时候，就会创建用于该handler的ctx，同时
25 @Override
26 public final ChannelPipeline addFirst(EventExecutorGroup group, String name, ChannelHandler handler) {
27 final AbstractChannelHandlerContext newCtx;
28 synchronized (this) {
29 checkMultiplicity(handler);
30 name = filterName(name, handler);
31
32 newCtx = newContext(group, name, handler);
33
34 addFirst0(newCtx);
35
36 // If the registered is false it means that the channel was not registered on an eventloop yet.
37 // In this case we add the context to the pipeline and add a task that will call
38 // ChannelHandler.handlerAdded(...) once the channel is registered.
39 if (!registered) {
40 newCtx.setAddPending();
41 callHandlerCallbackLater(newCtx, true);
42 return this;
43 }
44
45 EventExecutor executor = newCtx.executor();
46 if (!executor.inEventLoop()) {
47 newCtx.setAddPending();
48 executor.execute(new Runnable() {
49 @Override
50 public void run() {
51 callHandlerAdded0(newCtx);
52 }
53 });
54 return this;
55 }
56 }
57 //最后调用下ChannelHandler#handlerAdded
58 callHandlerAdded0(newCtx);
59 return this;
60 }
61
62 //同时将ctx的prev/next设置好
63 private void addFirst0(AbstractChannelHandlerContext newCtx) {
64 AbstractChannelHandlerContext nextCtx = head.next;
65 newCtx.prev = head;
66 newCtx.next = nextCtx;
67 head.next = newCtx;
68 nextCtx.prev = newCtx;
69 }
70
71 //最后调用下ChannelHandler#handlerAdded
72 private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
73 try {
74 ctx.handler().handlerAdded(ctx);
75 ctx.setAddComplete();
76 } catch (Throwable t) {
77 boolean removed = false;
78 try {
79 remove0(ctx);
80 try {
81 ctx.handler().handlerRemoved(ctx);
82 } finally {
83 ctx.setRemoved();
84 }
85 removed = true;
86 } catch (Throwable t2) {
87 if (logger.isWarnEnabled()) {
88 logger.warn("Failed to remove a handler: " + ctx.name(), t2);
89 }
90 }
91
92 if (removed) {
93 fireExceptionCaught(new ChannelPipelineException(
94 ctx.handler().getClass().getName() +
95 ".handlerAdded() has thrown an exception; removed.", t));
96 } else {
97 fireExceptionCaught(new ChannelPipelineException(
98 ctx.handler().getClass().getName() +
99 ".handlerAdded() has thrown an exception; also failed to remove.", t));
100 }
101 }
102 }
View Code
## pipeline的流程
ChannelPipeline注释里自带的
I/O Request
via Channel or
ChannelHandlerContext
|
+---------------------------------------------------+---------------+
| ChannelPipeline | |
| \|/ |
| +---------------------+ +-----------+----------+ |
| | Inbound Handler N | | Outbound Handler 1 | |
| +----------+----------+ +-----------+----------+ |
| /|\ | |
| | \|/ |
| +----------+----------+ +-----------+----------+ |
| | Inbound Handler N-1 | | Outbound Handler 2 | |
| +----------+----------+ +-----------+----------+ |
| /|\ . |
| . . |
| ChannelHandlerContext.fireIN_EVT() ChannelHandlerContext.OUT_EVT()|
| [ method call] [method call] |
| . . |
| . \|/ |
| +----------+----------+ +-----------+----------+ |
| | Inbound Handler 2 | | Outbound Handler M-1 | |
| +----------+----------+ +-----------+----------+ |
| /|\ | |
| | \|/ |
| +----------+----------+ +-----------+----------+ |
| | Inbound Handler 1 | | Outbound Handler M | |
| +----------+----------+ +-----------+----------+ |
| /|\ | |
+---------------+-----------------------------------+---------------+
| \|/
+---------------+-----------------------------------+---------------+
| | | |
| [ Socket.read() ] [ Socket.write() ] |
| |
| Netty Internal I/O Threads (Transport Implementation) |
+-------------------------------------------------------------------+
看的时候最好读写分开，否则会稍稍混淆。
inbound是读取外来内容，所以是从`socket#read`开始，逐个inbound handler调用。
outbound是将内存数据发送出去，所以先要经过一系列outbound handler处理，再由`socket#write`出去。
## 其他
其他诸如 ByteBuf/AttributeKey等，可以看相应的源码。
标签：Netty,newCtx,AbstractChannelHandlerContext,ctx,next,handler,executor,2022.10,源码 From： https://www.cnblogs.com/larryzeal/p/16793758.html