Netty4学习笔记（1）-- ChannelPipeline

青旅半醒 2022-06-03 09:57 226阅读 0赞

## Netty4 ##

[Netty][]是一个和[MINA][]类似的Java NIO框架，目前的最新版本是[4.0.13][]，这两个框架的主要作者好像都是[同一个韩国人][Link 1]。

## Channel ##

[Channel][]是Netty最核心的接口，一个Channel就是一个联络Socket的通道，通过Channel，你可以对Socket进行各种操作。

## ChannelHandler ##

用Netty编写网络程序的时候，你很少直接操纵Channel，而是通过[ChannelHandler][]来间接操纵Channel。

## ChannelPipeline ##

[ChannelPipeline][]实际上应该叫做ChannelHandlerPipeline，可以把ChannelPipeline看成是一个ChandlerHandler的链表，当需要对Channel进行某种处理的时候，Pipeline负责依次调用每一个Handler进行处理。每个Channel都有一个属于自己的Pipeline，调用Channel\#pipeline()方法可以获得Channel的Pipeline，调用Pipeline\#channel()方法可以获得Pipeline的Channel。

![SouthEast][]

ChannelPipeline的方法有很多，其中一部分是用来管理ChannelHandler的，如下面这些：

\[java\] [view plain][] [copy][view plain]

1.  ChannelPipeline addFirst(String name, ChannelHandler handler); 
2.  ChannelPipeline addLast(String name, ChannelHandler handler); 
3.  ChannelPipeline addBefore(String baseName, String name, ChannelHandler handler); 
4.  ChannelPipeline addAfter(String baseName, String name, ChannelHandler handler); 
5.  ChannelPipeline remove(ChannelHandler handler); 
6.  ChannelHandler remove(String name); 
7.  ChannelHandler removeFirst(); 
8.  ChannelHandler removeLast(); 
9.  ChannelPipeline replace(ChannelHandler oldHandler, String newName, ChannelHandler newHandler); 
10. ChannelHandler replace(String oldName, String newName, ChannelHandler newHandler); 
11. ChannelHandler first(); 
12. ChannelHandler last(); 
13. ChannelHandler get(String name);

根据上面的方法，能够大概想象的到Pipeline按照什么样的方式组织Handler。

## ChannelHandlerContext ##

ChannelPipeline并不是直接管理ChannelHandler，而是通过[ChannelHandlerContext][]来间接管理，这一点通过ChannelPipeline的默认实现DefaultChannelPipeline可以看出来。

![SouthEast 1][]

调用ChannelHandlerContext\#channel()方法可以得到和Context绑定的Channel，调用ChannelHandlerContext\#handler()方法可以得到和Context绑定的Handler。

##  ##

ChannelPipeline和ChannelHandlerContext默认实现

DefaultChannelHandlerContext和DefaultChannelPipeline是ChannelHandlerContext和ChannelPipeline的默认实现，下面是它们的部分代码：

\[java\] [view plain][] [copy][view plain]

1.  finalclass DefaultChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext \{ 
2.  
3.  volatile DefaultChannelHandlerContext next; 
4.  volatile DefaultChannelHandlerContext prev; 
5.  
6.  privatefinalboolean inbound; 
7.  privatefinalboolean outbound; 
8.  privatefinal AbstractChannel channel; 
9.  privatefinal DefaultChannelPipeline pipeline; 
10. privatefinal String name; 
11. privatefinal ChannelHandler handler; 
12. privateboolean removed; 
13. 
14. // ...
15. \}

\[java\] [view plain][] [copy][view plain]

1.  finalclass DefaultChannelPipeline implements ChannelPipeline \{ 
2.  // ...
3.  
4.  final DefaultChannelHandlerContext head; 
5.  final DefaultChannelHandlerContext tail; 
6.  
7.  // ...
8.  \}

从上面的代码可以看出，在DefaultPipeline内部，DefaultChannelHandlerContext组成了一个双向链表：

![SouthEast 2][]

再来看看DefaultChannelPipeline的构造函数：

\[java\] [view plain][] [copy][view plain]

1.  public DefaultChannelPipeline(AbstractChannel channel) \{ 
2.  if (channel == null) \{ 
3.  thrownew NullPointerException("channel"); 
4.   \} 
5.  this.channel = channel; 
6.  
7.   TailHandler tailHandler = new TailHandler(); 
8.   tail = new DefaultChannelHandlerContext(this, null, generateName(tailHandler), tailHandler); 
9.  
10.  HeadHandler headHandler = new HeadHandler(channel.unsafe()); 
11.  head = new DefaultChannelHandlerContext(this, null, generateName(headHandler), headHandler); 
12. 
13.  head.next = tail; 
14.  tail.prev = head; 
15. \}

可以看到，DefaultChinnelPipeline内部使用了两个特殊的Handler来表示Handler链的头和尾：

![SouthEast 3][]

## ChannelHandler的种类 ##

从上面DefaultChannelHandlerContext代码可以知道，Handler实际上分为两种，Inbound和Outbound，这一点也可以从ChannelHandler接口的子接口得到证明：

\[java\] [view plain][] [copy][view plain]

1.  publicinterface ChannelInboundHandler extends ChannelHandler \{ 
2.  // ...
3.  \} 
4.  
5.  publicinterface ChannelOutboundHandler extends ChannelHandler \{ 
6.  // ...
7.  \}

## 事件的传播 ##

为了搞清楚事件如何在Pipeline里传播，让我们从Channel的抽象子类AbstractChannel开始，下面是AbstractChannel\#write()方法的实现：

\[java\] [view plain][] [copy][view plain]

1.  publicabstractclass AbstractChannel extends DefaultAttributeMap implements Channel \{ 
2.  // ...
3.  @Override
4.  public Channel write(Object msg) \{ 
5.  return pipeline.write(msg); 
6.   \} 
7.  // ...
8.  \}

AbstractChannel直接调用了Pipeline的write()方法：

![SouthEast 4][]

再看DefaultChannelPipeline的write()方法实现：

\[java\] [view plain][] [copy][view plain]

1.  finalclass DefaultChannelPipeline implements ChannelPipeline \{ 
2.  // ...
3.  @Override
4.  public ChannelFuture write(Object msg) \{ 
5.  return tail.write(msg); 
6.   \} 
7.  // ...
8.  \}

因为write是个outbound事件，所以DefaultChannelPipeline直接找到tail部分的context，调用其write()方法：

![SouthEast 5][]

接着看DefaultChannelHandlerContext的write()方法：

\[java\] [view plain][] [copy][view plain]

1.  finalclass DefaultChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext \{ 
2.  // ...
3.  @Override
4.  public ChannelFuture write(Object msg) \{ 
5.  return write(msg, newPromise()); 
6.   \} 
7.  
8.  @Override
9.  public ChannelFuture write(final Object msg, final ChannelPromise promise) \{ 
10. if (msg == null) \{ 
11. thrownew NullPointerException("msg"); 
12.  \} 
13. 
14.  validatePromise(promise, true); 
15. 
16.  write(msg, false, promise); 
17. 
18. return promise; 
19.  \} 
20. 
21. privatevoid write(Object msg, boolean flush, ChannelPromise promise) \{ 
22.  DefaultChannelHandlerContext next = findContextOutbound(); 
23.  next.invokeWrite(msg, promise); 
24. if (flush) \{ 
25.  next.invokeFlush(); 
26.  \} 
27.  \} 
28. 
29. private DefaultChannelHandlerContext findContextOutbound() \{ 
30.  DefaultChannelHandlerContext ctx = this; 
31. do \{ 
32.  ctx = ctx.prev; 
33.  \} while (!ctx.outbound); 
34. return ctx; 
35.  \} 
36. 
37. privatevoid invokeWrite(Object msg, ChannelPromise promise) \{ 
38. try \{ 
39.  ((ChannelOutboundHandler) handler).write(this, msg, promise); 
40.  \} catch (Throwable t) \{ 
41.  notifyOutboundHandlerException(t, promise); 
42.  \} 
43.  \} 
44. 
45. // ...
46. \}

context的write()方法沿着context链往前找，直至找到一个outbound类型的context为止，然后调用其invokeWrite()方法：

![SouthEast 6][]

invokeWrite()接着调用handler的write()方法：

![SouthEast 7][]

最后看看ChannelOutboundHandlerAdapter的write()方法实现：

\[java\] [view plain][] [copy][view plain]

1.  publicclass ChannelOutboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelOutboundHandler \{ 
2.  // ...
3.  @Override
4.  publicvoid write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception \{ 
5.   ctx.write(msg, promise); 
6.   \} 
7.  // ...
8.  \}

默认的实现调用了context的write()方法而不做任何处理，这样write事件就沿着outbound链继续传播：

![SouthEast 8][]

可见，Pipeline的事件传播，是靠Pipeline，Context和Handler共同协作完成的。

[Netty]: http://netty.io/
[MINA]: http://mina.apache.org/
[4.0.13]: http://dl.bintray.com/netty/downloads/netty-4.0.13.Final.tar.bz2
[Link 1]: http://stackoverflow.com/users/55808/trustin
[Channel]: http://netty.io/4.0/api/io/netty/channel/Channel.html
[ChannelHandler]: http://netty.io/4.0/api/io/netty/channel/ChannelHandler.html
[ChannelPipeline]: http://netty.io/4.0/api/io/netty/channel/ChannelPipeline.html
[SouthEast]: /images/20220603/70fe1bf7c0cd4cb3a792b4eb220836f1.png
[view plain]: http://blog.csdn.net/zxhoo/article/details/17264263#
[ChannelHandlerContext]: http://netty.io/4.0/api/io/netty/channel/ChannelHandlerContext.html
[SouthEast 1]: /images/20220603/6f0d7620bb7b469a8e53ff0e877f8c85.png
[SouthEast 2]: /images/20220603/d43a57d228a9423fba6c7f3bc6e4ac43.png
[SouthEast 3]: /images/20220603/eb5cdced955e4e30abc9b5186abfa808.png
[SouthEast 4]: /images/20220603/dac7ad6a15024a499132d7f8503dd6a9.png
[SouthEast 5]: /images/20220603/432ba8c69a2b424489952192c8211ed6.png
[SouthEast 6]: /images/20220603/b9a8a6030dc04fa3b38e8a6ca9ba051a.png
[SouthEast 7]: /images/20220603/2692dcf2212d4f40a9b8116bc88191be.png
[SouthEast 8]: /images/20220603/a8ab3faee8fa47e1820c459c33ed872c.png