netty原始碼分析之pipeline(一)
通過前面的原始碼系列文章中的nettyreactor執行緒三部曲，我們已經知道，netty的reactor執行緒就像是一個發動機，驅動著整個netty框架的執行，而服務端的繫結和新連線的建立正是發動機的導火線，將發動機點燃
netty在服務端埠繫結和新連線建立的過程中會建立相應的channel，而與channel的動作密切相關的是pipeline這個概念，pipeline像是可以看作是一條流水線，原始的原料(位元組流)進來，經過加工，最後輸出
本文，我將以新連線的建立為例分為以下幾個部分給你介紹netty中的pipeline是怎麼玩轉起來的
- pipeline 初始化
- pipeline 新增節點
- pipeline 刪除節點
pipeline 初始化
在新連線的建立這篇文章中，我們已經知道了建立 NioSocketChannel 的時候會將netty的核心元件創建出來
pipeline是其中的一員，在下面這段程式碼中被建立
AbstractChannel
protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline();
}
複製程式碼
AbstractChannel
protected DefaultChannelPipeline newChannelPipeline() {
return new DefaultChannelPipeline(this);
}
複製程式碼
DefaultChannelPipeline
protected DefaultChannelPipeline(Channel channel) {
this.channel = ObjectUtil.checkNotNull(channel, "channel");
tail = new TailContext(this);
head = new HeadContext(this);
head.next = tail;
tail.prev = head;
}
複製程式碼
pipeline中儲存了channel的引用，建立完pipeline之後，整個pipeline是這個樣子的
pipeline中的每個節點是一個 ChannelHandlerContext 物件，每個context節點儲存了它包裹的執行器 ChannelHandler 執行操作所需要的上下文，其實就是pipeline，因為pipeline包含了channel的引用，可以拿到所有的context資訊
預設情況下，一條pipeline會有兩個節點，head和tail，後面的文章我們具體分析這兩個特殊的節點，今天我們重點放在pipeline
pipeline新增節點
下面是一段非常常見的客戶端程式碼
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new Spliter())
p.addLast(new Decoder());
p.addLast(new BusinessHandler())
p.addLast(new Encoder());
}
});
複製程式碼
首先，用一個spliter將來源TCP資料包拆包，然後將拆出來的包進行decoder，傳入業務處理器BusinessHandler，業務處理完encoder，輸出
整個pipeline結構如下
我用兩種顏色區分了一下pipeline中兩種不同型別的節點，一個是 ChannelInboundHandler ，處理inBound事件，最典型的就是讀取資料流，加工處理；還有一種型別的Handler是 ChannelOutboundHandler , 處理outBound事件，比如當呼叫 writeAndFlush() 類方法時，就會經過該種類型的handler
不管是哪種型別的handler，其外層物件 ChannelHandlerContext 之間都是通過雙向連結串列連線，而區分一個 ChannelHandlerContext 到底是in還是out，在新增節點的時候我們就可以看到netty是怎麼處理的
DefaultChannelPipeline
@Override
public final ChannelPipeline addLast(ChannelHandler... handlers) {
return addLast(null, handlers);
}
複製程式碼
@Override
public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
for (ChannelHandler h: handlers) {
addLast(executor, null, h);
}
return this;
}
複製程式碼
public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
final AbstractChannelHandlerContext newCtx;
synchronized (this) {
// 1.檢查是否有重複handler
checkMultiplicity(handler);
// 2.建立節點
newCtx = newContext(group, filterName(name, handler), handler);
// 3.新增節點
addLast0(newCtx);
}
// 4.回撥使用者方法
callHandlerAdded0(handler);
return this;
}
複製程式碼
這裡簡單地用 synchronized 方法是為了防止多執行緒併發操作pipeline底層的雙向連結串列
我們還是逐步分析上面這段程式碼
1.檢查是否有重複handler
在使用者程式碼新增一條handler的時候，首先會檢視該handler有沒有新增過
private static void checkMultiplicity(ChannelHandler handler) {
if (handler instanceof ChannelHandlerAdapter) {
ChannelHandlerAdapter h = (ChannelHandlerAdapter) handler;
if (!h.isSharable() && h.added) {
throw new ChannelPipelineException(
h.getClass().getName() +
" is not a @Sharable handler, so can't be added or removed multiple times.");
}
h.added = true;
}
}
複製程式碼
netty使用一個成員變數 added 標識一個channel是否已經新增，上面這段程式碼很簡單，如果當前要新增的Handler是非共享的，並且已經新增過，那就丟擲異常，否則，標識該handler已經新增
由此可見，一個Handler如果是sharable的，就可以無限次被新增到pipeline中，我們客戶端程式碼如果要讓一個Handler被共用，只需要加一個@Sharable標註即可，如下
@Sharable
public class BusinessHandler {
}
複製程式碼
而如果Handler是sharable的，一般就通過spring的注入的方式使用，不需要每次都new 一個
isSharable() 方法正是通過該Handler對應的類是否標註@Sharable來實現的
ChannelHandlerAdapter
public boolean isSharable() {
Class<?> clazz = getClass();
Map<Class<?>, Boolean> cache = InternalThreadLocalMap.get().handlerSharableCache();
Boolean sharable = cache.get(clazz);
if (sharable == null) {
sharable = clazz.isAnnotationPresent(Sharable.class);
cache.put(clazz, sharable);
}
return sharable;
}
複製程式碼
這裡也可以看到，netty為了效能優化到極致，還使用了ThreadLocal來快取Handler的狀態，高併發海量連線下，每次有新連線新增Handler都會建立呼叫此方法
2.建立節點
回到主流程，看建立上下文這段程式碼
newCtx = newContext(group, filterName(name, handler), handler); 複製程式碼
這裡我們需要先分析 filterName(name, handler) 這段程式碼，這個函式用於給handler建立一個唯一性的名字
private String filterName(String name, ChannelHandler handler) {
if (name == null) {
return generateName(handler);
}
checkDuplicateName(name);
return name;
}
複製程式碼
顯然，我們傳入的name為null，netty就給我們生成一個預設的name，否則，檢查是否有重名，檢查通過的話就返回
netty建立預設name的規則為 簡單類名#0 ，下面我們來看些具體是怎麼實現的
private static final FastThreadLocal<Map<Class<?>, String>> nameCaches =
new FastThreadLocal<Map<Class<?>, String>>() {
@Override
protected Map<Class<?>, String> initialValue() throws Exception {
return new WeakHashMap<Class<?>, String>();
}
};
private String generateName(ChannelHandler handler) {
// 先檢視快取中是否有生成過預設name
Map<Class<?>, String> cache = nameCaches.get();
Class<?> handlerType = handler.getClass();
String name = cache.get(handlerType);
// 沒有生成過，就生成一個預設name，加入快取
if (name == null) {
name = generateName0(handlerType);
cache.put(handlerType, name);
}
// 生成完了，還要看預設name有沒有衝突
if (context0(name) != null) {
String baseName = name.substring(0, name.length() - 1);
for (int i = 1;; i ++) {
String newName = baseName + i;
if (context0(newName) == null) {
name = newName;
break;
}
}
}
return name;
}
複製程式碼
netty使用一個 FastThreadLocal (後面的文章會細說)變數來快取Handler的類和預設名稱的對映關係，在生成name的時候，首先檢視快取中有沒有生成過預設name( 簡單類名#0 )，如果沒有生成，就呼叫 generateName0() 生成預設name，然後加入快取
接下來還需要檢查name是否和已有的name有衝突，呼叫 context0() ，查詢pipeline裡面有沒有對應的context
private AbstractChannelHandlerContext context0(String name) {
AbstractChannelHandlerContext context = head.next;
while (context != tail) {
if (context.name().equals(name)) {
return context;
}
context = context.next;
}
return null;
}
複製程式碼
context0() 方法連結串列遍歷每一個 ChannelHandlerContext ，只要發現某個context的名字與待新增的name相同，就返回該context，最後丟擲異常，可以看到，這個其實是一個線性搜尋的過程
如果 context0(name) != null 成立，說明現有的context裡面已經有了一個預設name，那麼就從 簡單類名#1 往上一直找，直到找到一個唯一的name，比如 簡單類名#3
如果使用者程式碼在新增Handler的時候指定了一個name，那麼要做到事僅僅為檢查一下是否有重複
private void checkDuplicateName(String name) {
if (context0(name) != null) {
throw new IllegalArgumentException("Duplicate handler name: " + name);
}
}
複製程式碼
處理完name之後，就進入到建立context的過程，由前面的呼叫鏈得知， group 為null，因此 childExecutor(group) 也返回null
DefaultChannelPipeline
private AbstractChannelHandlerContext newContext(EventExecutorGroup group, String name, ChannelHandler handler) {
return new DefaultChannelHandlerContext(this, childExecutor(group), name, handler);
}
private EventExecutor childExecutor(EventExecutorGroup group) {
if (group == null) {
return null;
}
//..
}
複製程式碼
DefaultChannelHandlerContext
DefaultChannelHandlerContext(
DefaultChannelPipeline pipeline, EventExecutor executor, String name, ChannelHandler handler) {
super(pipeline, executor, name, isInbound(handler), isOutbound(handler));
if (handler == null) {
throw new NullPointerException("handler");
}
this.handler = handler;
}
複製程式碼
建構函式中， DefaultChannelHandlerContext 將引數回傳到父類，儲存Handler的引用，進入到其父類
AbstractChannelHandlerContext
AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor, String name,
boolean inbound, boolean outbound) {
this.name = ObjectUtil.checkNotNull(name, "name");
this.pipeline = pipeline;
this.executor = executor;
this.inbound = inbound;
this.outbound = outbound;
}
複製程式碼
netty中用兩個欄位來表示這個 channelHandlerContext 屬於 inBound 還是 outBound ，或者兩者都是，兩個boolean是通過下面兩個小函式來判斷(見上面一段程式碼)
DefaultChannelHandlerContext
private static boolean isInbound(ChannelHandler handler) {
return handler instanceof ChannelInboundHandler;
}
private static boolean isOutbound(ChannelHandler handler) {
return handler instanceof ChannelOutboundHandler;
}
複製程式碼
通過 instanceof 關鍵字根據介面型別來判斷，因此，如果一個Handler實現了兩類介面，那麼他既是一個inBound型別的Handler，又是一個outBound型別的Handler，比如下面這個類
常用的，將decode操作和encode操作合併到一起的codec，一般會繼承 MessageToMessageCodec ，而 MessageToMessageCodec 就是繼承 ChannelDuplexHandler
MessageToMessageCodec
public abstract class MessageToMessageCodec<INBOUND_IN, OUTBOUND_IN> extends ChannelDuplexHandler {
protected abstract void encode(ChannelHandlerContext ctx, OUTBOUND_IN msg, List<Object> out)
throws Exception;
protected abstract void decode(ChannelHandlerContext ctx, INBOUND_IN msg, List<Object> out)
throws Exception;
}
複製程式碼
context 建立完了之後，接下來終於要將建立完畢的context加入到pipeline中去了
3.新增節點
private void addLast0(AbstractChannelHandlerContext newCtx) {
AbstractChannelHandlerContext prev = tail.prev;
newCtx.prev = prev; // 1
newCtx.next = tail; // 2
prev.next = newCtx; // 3
tail.prev = newCtx; // 4
}
複製程式碼
用下面這幅圖可見簡單的表示這段過程，說白了，其實就是一個雙向連結串列的插入操作
操作完畢，該context就加入到pipeline中
到這裡，pipeline新增節點的操作就完成了，你可以根據此思路掌握所有的addxxx()系列方法
4.回撥使用者方法
AbstractChannelHandlerContext
private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
ctx.handler().handlerAdded(ctx);
ctx.setAddComplete();
}
複製程式碼
到了第四步，pipeline中的新節點新增完成，於是便開始回撥使用者程式碼 ctx.handler().handlerAdded(ctx); ，常見的使用者程式碼如下
AbstractChannelHandlerContext
public class DemoHandler extends SimpleChannelInboundHandler<...> {
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
// 節點被新增完畢之後回撥到此
// do something
}
}
複製程式碼
接下來，設定該節點的狀態
AbstractChannelHandlerContext
final void setAddComplete() {
for (;;) {
int oldState = handlerState;
if (oldState == REMOVE_COMPLETE || HANDLER_STATE_UPDATER.compareAndSet(this, oldState, ADD_COMPLETE)) {
return;
}
}
}
複製程式碼
用cas修改節點的狀態至：REMOVE_COMPLETE（說明該節點已經被移除） 或者 ADD_COMPLETE
pipeline刪除節點
netty 有個最大的特性之一就是Handler可插拔，做到動態編織pipeline，比如在首次建立連線的時候，需要通過進行許可權認證，在認證通過之後，就可以將此context移除，下次pipeline在傳播事件的時候就就不會呼叫到許可權認證處理器
下面是許可權認證Handler最簡單的實現，第一個資料包傳來的是認證資訊，如果校驗通過，就刪除此Handler，否則，直接關閉連線
public class AuthHandler extends SimpleChannelInboundHandler<ByteBuf> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf data) throws Exception {
if (verify(authDataPacket)) {
ctx.pipeline().remove(this);
} else {
ctx.close();
}
}
private boolean verify(ByteBuf byteBuf) {
//...
}
}
複製程式碼
重點就在 ctx.pipeline().remove(this) 這段程式碼
@Override
public final ChannelPipeline remove(ChannelHandler handler) {
remove(getContextOrDie(handler));
return this;
}
複製程式碼
remove操作相比add簡單不少，分為三個步驟：
1.找到待刪除的節點 2.調整雙向連結串列指標刪除 3.回撥使用者函式
1.找到待刪除的節點
DefaultChannelPipeline
private AbstractChannelHandlerContext getContextOrDie(ChannelHandler handler) {
AbstractChannelHandlerContext ctx = (AbstractChannelHandlerContext) context(handler);
if (ctx == null) {
throw new NoSuchElementException(handler.getClass().getName());
} else {
return ctx;
}
}
@Override
public final ChannelHandlerContext context(ChannelHandler handler) {
if (handler == null) {
throw new NullPointerException("handler");
}
AbstractChannelHandlerContext ctx = head.next;
for (;;) {
if (ctx == null) {
return null;
}
if (ctx.handler() == handler) {
return ctx;
}
ctx = ctx.next;
}
}
複製程式碼
這裡為了找到Handler對應的context，照樣是通過依次遍歷雙向連結串列的方式，直到某一個context的Handler和當前Handler相同，便找到了該節點
2.調整雙向連結串列指標刪除
DefaultChannelPipeline
private AbstractChannelHandlerContext remove(final AbstractChannelHandlerContext ctx) {
assert ctx != head && ctx != tail;
synchronized (this) {
// 2.調整雙向連結串列指標刪除
remove0(ctx);
}
// 3.回撥使用者函式
callHandlerRemoved0(ctx);
return ctx;
}
private static void remove0(AbstractChannelHandlerContext ctx) {
AbstractChannelHandlerContext prev = ctx.prev;
AbstractChannelHandlerContext next = ctx.next;
prev.next = next; // 1
next.prev = prev; // 2
}
複製程式碼
經歷的過程要比新增節點要簡單，可以用下面一幅圖來表示
最後的結果為
結合這兩幅圖，可以很清晰地瞭解許可權驗證Handler的工作原理，另外，被刪除的節點因為沒有物件引用到，果過段時間就會被gc自動回收
3.回撥使用者函式
private void callHandlerRemoved0(final AbstractChannelHandlerContext ctx) {
try {
ctx.handler().handlerRemoved(ctx);
} finally {
ctx.setRemoved();
}
}
複製程式碼
到了第三步，pipeline中的節點刪除完成，於是便開始回撥使用者程式碼 ctx.handler().handlerRemoved(ctx); ，常見的程式碼如下
AbstractChannelHandlerContext
public class DemoHandler extends SimpleChannelInboundHandler<...> {
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
// 節點被刪除完畢之後回撥到此，可做一些資源清理
// do something
}
}
複製程式碼
最後，將該節點的狀態設定為removed
final void setRemoved() {
handlerState = REMOVE_COMPLETE;
}
複製程式碼
removexxx系列的其他方法族大同小異，你可以根據上面的思路展開其他的系列方法，這裡不再贅述
總結
1.以新連線建立為例，新連線建立的過程中建立channel，而在建立channel的過程中建立了該channel對應的pipeline，建立完pipeline之後，自動給該pipeline添加了兩個節點，即ChannelHandlerContext，ChannelHandlerContext中有用pipeline和channel所有的上下文資訊。
2.pipeline是雙向個連結串列結構，新增和刪除節點均只需要調整連結串列結構
3.pipeline中的每個節點包著具體的處理器 ChannelHandler ，節點根據 ChannelHandler 的型別是 ChannelInboundHandler 還是 ChannelOutboundHandler 來判斷該節點屬於in還是out或者兩者都是
下一篇文章將繼續pipeline的分析，敬請期待！
如果你想系統地學Netty，我的小冊 ofollow,noindex">《Netty 入門與實戰：仿寫微信 IM 即時通訊系統》 可以幫助你，如果你想系統學習Netty原理，那麼你一定不要錯過我的Netty原始碼分析系列視訊： coding.imooc.com/class/230.h…