Netty原始碼分析第4章(pipeline)---->第7節: 前章節內容回顧
Netty原始碼分析第四章: pipeline
第七節: 前章節內容回顧
我們在第一章和第三章中, 遺留了很多有關事件傳輸的相關邏輯, 這裡帶大家一一回顧
首先看兩個問題:
1.在客戶端接入的時候, NioMessageUnsafe的read方法中pipeline.fireChannelRead(readBuf.get(i))為什麼會呼叫到ServerBootstrap的內部類ServerBootstrapAcceptor中的channelRead()方法
2.客戶端handler是什麼時候被新增的?
首先看第一個問題:
1.在客戶端接入的時候, NioMessageUnsafe的read方法中pipeline.fireChannelRead(readBuf.get(i))為什麼會呼叫到ServerBootstrap的內部類ServerBootstrapAcceptor中的channelRead()方法?
我們首先看這段程式碼:
public void read() {
//必須是NioEventLoop方法呼叫的, 不能通過外部執行緒呼叫
assert eventLoop().inEventLoop();
//服務端channel的config
final ChannelConfig config = config();
//服務端channel的pipeline
final ChannelPipeline pipeline = pipeline();
//處理服務端接入的速率
final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
重點看pipeline.fireChannelRead(readBuf.get(i))
首先, 這裡pipeline是服務端channel的pipeline, 也就是NioServerSocketChannel的pipeline
我們學習過pipeline之後, 對這種寫法並不陌生, 就是傳遞channelRead事件, 這裡通過傳遞channelRead事件走到了ServerBootstrapAcceptor的channelRead()方法, 說明在這步之前, ServerBootstrapAcceptor作為一個handler新增到了服務端channel的pipeline中, 那麼這個handler什麼時候新增的呢?
我們回顧下第一章, 初始化NioServerSocketChannel的時候, 呼叫了ServerBootstrap的init方法:
void init(Channel channel) throws Exception {
//獲取使用者定義的選項(1)
final Map<ChannelOption<?>, Object> options = options0();
synchronized (options) {
channel.config().setOptions(options);
}
//獲取使用者定義的屬性(2)
final Map<AttributeKey<?>, Object> attrs = attrs0();
synchronized (attrs) {
for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
@SuppressWarnings("unchecked")
AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
channel.attr(key).set(e.getValue());
}
}
//獲取channel的pipline(3)
ChannelPipeline p = channel.pipeline();
//work執行緒組(4)
final EventLoopGroup currentChildGroup = childGroup;
//使用者設定的Handler(5)
final ChannelHandler currentChildHandler = childHandler;
final Entry<ChannelOption<?>, Object>[] currentChildOptions;
final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
//選項轉化為Entry物件(6)
synchronized (childOptions) {
currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
}
//屬性轉化為Entry物件(7)
synchronized (childAttrs) {
currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
}
//新增服務端handler(8)
p.addLast(new ChannelInitializer<Channel>() {
//初始化channel
@Override
public void initChannel(Channel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();
ChannelHandler handler = config.handler();
if (handler != null) {
pipeline.addLast(handler);
}
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
pipeline.addLast(new ServerBootstrapAcceptor(
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}
這個方法比較長, 我們重點關注第8步, 新增服務端channel, 這裡的pipeline, 是服務服務端channel的pipeline, 也就是NioServerSocketChannel繫結的pipeline, 這裡添加了一個ChannelInitializer型別的handler
我們看一下ChannelInitializer這個類的繼承關係:
public abstract class ChannelInitializer<C extends Channel> extends ChannelInboundHandlerAdapter {
//省略類體
}
我們看到其繼承了ChannelInboundHandlerAdapter, 說明是一個inbound型別的handler
這裡我們可能會想到, 新增完handler會執行handlerAdded, 然後再handlerAdded方法中做了新增ServerBootstrapAcceptor這個handler
但是, 實際上並不是這樣的, 當程式執行到這裡, 並沒有馬上執行handlerAdded, 我們緊跟addLast方法
最後會跟到DefualtChannelPipeline的一個addLast方法中去:
public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
final AbstractChannelHandlerContext newCtx;
synchronized (this) {
//判斷handler是否被重複新增(1)
checkMultiplicity(handler);
//建立一個HandlerContext並新增到列表(2)
newCtx = newContext(group, filterName(name, handler), handler);
//新增HandlerContext(3)
addLast0(newCtx);
//是否已註冊
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;
}
}
//回撥新增事件(4)
callHandlerAdded0(newCtx);
return this;
}
首先完成了handler的新增, 但是並沒有馬上執行回撥
這裡我們重點關注if (!registered)這個條件判斷, 其實在註冊完成, registered會變成true, 但是走到這一步的時候NioServerSockeChannel並沒有完成註冊(可以回顧第一章看註冊在哪一步), 所以會進到if裡並返回自身
我們重點關注callHandlerCallbackLater這個方法, 我們跟進去:
private void callHandlerCallbackLater(AbstractChannelHandlerContext ctx, boolean added) {
assert !registered;
//判斷是否已新增, 未新增, 進行新增, 已新增進行刪除
PendingHandlerCallback task = added ? new PendingHandlerAddedTask(ctx) : new PendingHandlerRemovedTask(ctx);
//獲取第一個Callback任務
PendingHandlerCallback pending = pendingHandlerCallbackHead;
//如果第一個Callback任務為空
if (pending == null) {
//將第一個任務設定為剛建立的任務
pendingHandlerCallbackHead = task;
} else {
while (pending.next != null) {
pending = pending.next;
}
pending.next = task;
}
}
因我們呼叫這個方法的時候added傳的true, 所以PendingHandlerCallback task賦值為new PendingHandlerAddedTask(ctx)
PendingHandlerAddedTask這個類, 我們從名字可以看出, 這是一個handler新增的延遲任務, 用於執行handler延遲新增的操作, 同樣也對應一個名字為PendingHandlerRemovedTask的類, 用於執行延遲刪除handler的操作, 這兩個類都繼承抽象類PendingHandlerCallback
我們看PendingHandlerAddedTask類構造方法:
PendingHandlerAddedTask(AbstractChannelHandlerContext ctx) {
super(ctx);
}
這裡呼叫了父類的構造方法, 再跟進去:
PendingHandlerCallback(AbstractChannelHandlerContext ctx) {
this.ctx = ctx;
}
在父類中, 儲存了要新增的context, 也就是ChannelInitializer型別的包裝類
回到callHandlerCallbackLater方法中:
PendingHandlerCallback pending = pendingHandlerCallbackHead;
這表示獲取第一個PendingHandlerCallback的任務, 其實PendingHandlerCallback是一個單向連結串列, 自身維護一個PendingHandlerCallback型別的next, 指向下一個任務, 在DefaultChannelPipeline這個類中, 定義了個PendingHandlerCallback型別的引用pendingHandlerCallbackHead, 用來指向延遲迴調任務的中的第一個任務
之後判斷這個任務是為空, 如果是第一次新增handler, 那麼這裡就是空, 所以將第一個任務賦值為我們剛建立的新增任務
如果不是第一次新增handler, 則將我們新建立的任務新增到連結串列的尾部, 因為這裡我們是第一次新增, 所以第一個回撥任務就指向了我們建立的新增handler的任務
完成這一系列操作之後, addLast方法返歸, 此時並沒有完成新增操作
而什麼時候完成新增操作的呢?
在服務端channel註冊時候的會走到AbstractChannel的register0方法:
private void register0(ChannelPromise promise) {
try {
//做實際的註冊(1)
doRegister();
neverRegistered = false;
registered = true;
//觸發事件(2)
pipeline.invokeHandlerAddedIfNeeded();
safeSetSuccess(promise);
//觸發註冊成功事件(3)
pipeline.fireChannelRegistered();
if (isActive()) {
if (firstRegistration) {
//傳播active事件(4)
pipeline.fireChannelActive();
} else if (config().isAutoRead()) {
beginRead();
}
}
} catch (Throwable t) {
//省略程式碼
}
}
重點關注第二步pipeline.invokeHandlerAddedIfNeeded(), 這裡已經通過doRegister()方法完成了實際的註冊, 我們跟到該方法中:
final void invokeHandlerAddedIfNeeded() {
assert channel.eventLoop().inEventLoop();
if (firstRegistration) {
firstRegistration = false;
callHandlerAddedForAllHandlers();
}
}
這裡會判斷是否第一次註冊, 這裡返回true, 然後會執行callHandlerAddedForAllHandlers()方法, 我們跟進去:
private void callHandlerAddedForAllHandlers() {
final PendingHandlerCallback pendingHandlerCallbackHead;
synchronized (this) {
assert !registered;
registered = true;
pendingHandlerCallbackHead = this.pendingHandlerCallbackHead;
this.pendingHandlerCallbackHead = null;
}
//獲取task
PendingHandlerCallback task = pendingHandlerCallbackHead;
while (task != null) {
//執行新增handler方法
task.execute();
task = task.next;
}
}
這裡拿到第一個延遲執行handler新增的task其實就是我們之前剖析過的, 延遲執行handler新增的task, 就是PendingHandlerAddedTask物件
在while迴圈中, 通過執行execute()方法將handler新增
我們跟到PendingHandlerAddedTask的execute()方法中:
void execute() {
//獲取當前eventLoop執行緒
EventExecutor executor = ctx.executor();
//是當前執行的執行緒
if (executor.inEventLoop()) {
callHandlerAdded0(ctx);
} else {
try {
//新增到佇列
executor.execute(this);
} catch (RejectedExecutionException e) {
//程式碼省略
}
}
}
終於在這裡, 我們看到了執行回撥的方法
再回到init方法中:
void init(Channel channel) throws Exception {
//獲取使用者定義的選項(1)
final Map<ChannelOption<?>, Object> options = options0();
synchronized (options) {
channel.config().setOptions(options);
}
//獲取使用者定義的屬性(2)
final Map<AttributeKey<?>, Object> attrs = attrs0();
synchronized (attrs) {
for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
@SuppressWarnings("unchecked")
AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
channel.attr(key).set(e.getValue());
}
}
//獲取channel的pipline(3)
ChannelPipeline p = channel.pipeline();
//work執行緒組(4)
final EventLoopGroup currentChildGroup = childGroup;
//使用者設定的Handler(5)
final ChannelHandler currentChildHandler = childHandler;
final Entry<ChannelOption<?>, Object>[] currentChildOptions;
final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
//選項轉化為Entry物件(6)
synchronized (childOptions) {
currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
}
//屬性轉化為Entry物件(7)
synchronized (childAttrs) {
currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
}
//新增服務端handler(8)
p.addLast(new ChannelInitializer<Channel>() {
//初始化channel
@Override
public void initChannel(Channel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();
ChannelHandler handler = config.handler();
if (handler != null) {
pipeline.addLast(handler);
}
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
pipeline.addLast(new ServerBootstrapAcceptor(
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}
我們繼續看第8步新增服務端handler
因為這裡的handler是ChannelInitializer, 所以完成新增之後會呼叫ChannelInitializer的handlerAdded方法
跟到handlerAdded方法:
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
//預設情況下, 會返回true
if (ctx.channel().isRegistered()) {
initChannel(ctx);
}
}
因為執行到這步服務端channel已經完成註冊, 所以會執行到initChannel方法
跟到initChannel方法:
private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
//這段程式碼是否被執行過
if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) {
try {
initChannel((C) ctx.channel());
} catch (Throwable cause) {
exceptionCaught(ctx, cause);
} finally {
//呼叫之後會刪除當前節點
remove(ctx);
}
return true;
}
return false;
}
我們關注initChannel這個方法, 這個方法是在ChannelInitializer的匿名內部來實現的, 這裡我們注意, 在initChannel方法執行完畢之後會呼叫remove(ctx)刪除當前節點
我們繼續跟進initChannel方法:
@Override
public void initChannel(Channel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();
ChannelHandler handler = config.handler();
if (handler != null) {
pipeline.addLast(handler);
}
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
pipeline.addLast(new ServerBootstrapAcceptor(
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
這裡首先新增使用者自定義的handler, 這裡如果使用者沒有定義, 則新增不成功, 然後, 會呼叫addLast將ServerBootstrapAcceptor這個handler添加了進去, 同樣這個handler也繼承了ChannelInboundHandlerAdapter, 在這個handler中, 重寫了channelRead方法, 所以, 這就是第一個問題的答案
緊接著我們看第二個問題:
2.客戶端handler是什麼時候被新增的?
我們這裡看ServerBootstrapAcceptor的channelRead方法:
public void channelRead(ChannelHandlerContext ctx, Object msg) {
final Channel child = (Channel) msg;
//新增channelHadler, 這個channelHandler, 就是使用者程式碼新增的ChannelInitializer
child.pipeline().addLast(childHandler);
//程式碼省略
try {
//work執行緒註冊channel
childGroup.register(child).addListener(new ChannelFutureListener() {
//程式碼省略
});
} catch (Throwable t) {
forceClose(child, t);
}
}
這裡真相可以大白了, 服務端再建立完客戶端channel之後, 將新建立的NioSocketChannel作為引數觸發channelRead事件(可以回顧NioMessageUnsafe的read方法, 程式碼這裡就不貼了), 所以這裡的引數msg就是NioSocketChannel
拿到channel時候再將客戶端的handler新增進去, 我們回顧客戶端handler的新增過程:
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new StringDecoder());
ch.pipeline().addLast(new StringEncoder());
ch.pipeline().addLast(new ServerHandler());
}
});
和服務端channel的邏輯一樣, 首先會新增ChannelInitializer這個handler但是沒有註冊所以沒有執行新增handler的回撥, 將任務儲存到一個延遲迴調的task中
等客戶端channel註冊完畢, 會將執行新增handler的回撥, 也就是handlerAdded方法, 在回撥中執行initChannel方法將客戶端handler新增進去, 然後刪除ChannelInitializer這個handler
因為在服務端channel中這塊邏輯已經進行了詳細的剖析, 所以這邊就不在贅述, 同學們可以自己跟進去走一遍流程
這裡注意, 因為每建立一個NioSoeketChannel都會呼叫服務端ServerBootstrapAcceptor的channelRead方法, 所以這裡會將每一個NioSocketChannel的handler進行新增
第四章總結
本章剖析了事件傳輸的相關邏輯, 包括handler的新增, 刪除, inbound和outbound以及異常事件的傳輸, 最後結合第一章和第三章, 剖析了服務端channel和客戶端channel的新增過程, 同學們可以課後跟進原始碼, 將這些功能自己再走一遍以加深印象.其他的有關事件傳輸的邏輯, 可以結合這一章的知識點進行自行剖析