Netty 源碼解析
寫在開始
這篇文章的主要目的是探尋Netty的啟動(dòng)過程，目的是搞清楚下邊兩個(gè)問題胜嗓「咧埃看這篇文章至少需要具備一些Java知識，并且大概對Netty有一些了解辞州。
1.ServerSocketChannel是何時(shí)怔锌，如何初始化的？
2.ServerSocketChannel是何時(shí)register到Eventloop变过？
準(zhǔn)備工作
- 下載源碼：
git@github.com:netty/netty.git产禾。如果想看某個(gè)分支的代碼可以checkout到那個(gè)分支。 - 導(dǎo)入idea牵啦，idea會自動(dòng)識別maven項(xiàng)目亚情，下載依賴需要一段時(shí)間。
- 編譯哈雏，方便debug楞件。首先build common：
cd common && mvn clean install，然后就可以在idea中運(yùn)行程序了裳瘪。
源碼分析
Netty源碼中土浸，有一個(gè)子項(xiàng)目叫examples。這里邊自帶了非常多Netty的例子彭羹，學(xué)習(xí)Netty黄伊，這些例子是最好的入門資料。我們打開DiscardServer這個(gè)示例派殷，這個(gè)例子是最簡單的入門示例还最。相當(dāng)于學(xué)習(xí)一門新語言時(shí)的HelloWorld。從哪里開始看起呢毡惜？請看下邊代碼：
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(PORT).sync();
ServerBootstrap的bind方法是整個(gè)邏輯的入口拓轻。我們進(jìn)入bind的實(shí)現(xiàn)，會跳轉(zhuǎn)到AbstractBootstrap這個(gè)類经伙，經(jīng)過兩個(gè)方法跳轉(zhuǎn)我們開始看AbstractBootstrap.doBind方法：
private ChannelFuture doBind(final SocketAddress localAddress) {
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
if (regFuture.cause() != null) {
return regFuture;
}
if (regFuture.isDone()) {
ChannelPromise promise = channel.newPromise();
doBind0(regFuture, channel, localAddress, promise);
return promise;
} else {
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
Throwable cause = future.cause();
if (cause != null) {
promise.setFailure(cause);
} else {
promise.registered();
doBind0(regFuture, channel, localAddress, promise);
}
}
});
return promise;
}
}
Netty代碼質(zhì)量非常高扶叉，通過讀代碼我們就能大概知道這個(gè)函數(shù)的功能（當(dāng)然需要你有一點(diǎn)Netty相關(guān)的知識儲備）：
初始化一個(gè)Channel，并把它register到Eventloop上。
綁定Address枣氧。
如果我們不想探究細(xì)節(jié)的話溢十，其實(shí)前兩個(gè)問題我們現(xiàn)在已經(jīng)可以解答了：AbstractBootstrap類的doBind方法，通過調(diào)用initAndRegister創(chuàng)建一個(gè)Channel达吞，并把它Register到Eventloop上茶宵。但是這樣回答時(shí)我們還是非常不自信的，因?yàn)閕nitAndRegister干了什么宗挥，doBind0又干了啥乌庶，我們沒有一點(diǎn)頭緒。所以契耿，我們再接著向下深入一層（好的代碼是有層次感的瞒大，而不是一個(gè)全能的大方法搞定一切）。
我們接下來展開initAndRegister方法搪桂，看這個(gè)方法有兩個(gè)目的：channel如何init透敌，如何register？
final ChannelFuture initAndRegister() {
Channel channel = null;
try {
channel = channelFactory.newChannel();
init(channel);
} catch (Throwable t) {
if (channel != null) {
channel.unsafe().closeForcibly();
}
return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
}
ChannelFuture regFuture = config().group().register(channel);
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
return regFuture;
}
首先看try里邊的邏輯：channelFactory.newChannel()踢械。channelFactory是啥時(shí)初始化的呢酗电？我們再回到DiscardServer類。
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
ChannelPipeline p = ch.pipeline();
if (sslCtx != null) {
p.addLast(sslCtx.newHandler(ch.alloc()));
}
p.addLast(new DiscardServerHandler());
}
});
我們跟進(jìn)一下channel這個(gè)方法内列。
public B channel(Class<? extends C> channelClass) {
if (channelClass == null) {
throw new NullPointerException("channelClass");
}
return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
}
public B channelFactory(ChannelFactory<? extends C> channelFactory) {
if (channelFactory == null) {
throw new NullPointerException("channelFactory");
}
if (this.channelFactory != null) {
throw new IllegalStateException("channelFactory set already");
}
this.channelFactory = channelFactory;
return (B) this;
}
channel調(diào)用了channelFactory這個(gè)方法撵术，并把NioServerSocketChannel的ReflectiveChannelFactory傳遞給它。channelFactory把這個(gè)值傳給了channelFactory屬性话瞧∧塾耄看到這其實(shí)已經(jīng)有點(diǎn)譜了，channel是ReflectiveChannelFactory通過反射方式創(chuàng)建的NioServerSocketChannel的實(shí)例交排。
接下來我們開始走register這條線划滋。執(zhí)行register的邏輯是：ChannelFuture regFuture = config().group().register(channel)。我們跟進(jìn)group()方法埃篓，發(fā)現(xiàn)最終這個(gè)方法返回的是AbstractBootstrap的group的屬性处坪。找一下這個(gè)group是何時(shí)set的〖茏ǎ回到DiscardServer的代碼中：b.group(bossGroup, workerGroup)同窘，我們跟進(jìn)這個(gè)group方法。進(jìn)去后發(fā)現(xiàn)super.group(parentGroup);胶征，調(diào)用了AbstractBootstrap的group方法塞椎，在這個(gè)方法中對group屬性進(jìn)行了賦值。所以睛低，config().group()返回的是bossGroup。
接下來跟進(jìn)register方法。register是EventLoopGroup接口中的方法钱雷。究竟該看哪個(gè)實(shí)現(xiàn)呢骂铁？再回到DiscardServer類，我們發(fā)現(xiàn)bossGroup是一個(gè)NioEventLoopGroup的實(shí)例罩抗。但是NioEventLoopGroup中并沒有register方法拉庵。看一下NioEventLoopGroup的繼承關(guān)系套蒂，如果它沒實(shí)現(xiàn)肯定是在其父類中實(shí)現(xiàn)了钞支。我們看一下它的繼承關(guān)系：
去MultithreadEventLoopGroup中，確實(shí)有register方法的實(shí)現(xiàn)：
public ChannelFuture register(Channel channel) {
return next().register(channel);
}
看一下next方法操刀，調(diào)用super的next方法并返回一個(gè)EventLoop烁挟。然后調(diào)用EventLoop的register方法，把Channel注冊到EventLoop上骨坑。OK撼嗓，目前為止，前兩個(gè)問題算是搞清楚了欢唾。
我們可以就此止步且警，也可以繼續(xù)探究下去。因?yàn)檠矍斑€有兩個(gè)問題: next方法是如何實(shí)現(xiàn)的礁遣？register過程具體如何執(zhí)行斑芜？我們尚不清楚。接下來看一下next方法是如何實(shí)現(xiàn)的：
public EventLoop next() {
return (EventLoop) super.next();
}
在MultithreadEventLoopGroup的next方法直接調(diào)用了其父類MultithreadEventExecutorGroup的next方法祟霍，我門來看一下它是如何實(shí)現(xiàn)的：
@Override
public EventExecutor next() {
return chooser.next();
}
在這個(gè)類中有一個(gè)選擇器chooser押搪，chooser的賦值時(shí)機(jī)是在初始化這個(gè)類的時(shí)候，我們看其構(gòu)造函數(shù)：
protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
}
DefaultEventExecutorChooserFactory中實(shí)現(xiàn)了兩個(gè)Chooser：PowerOfTwoEventExecutorChooser和GenericEventExecutorChooser浅碾。選擇chooser時(shí)會進(jìn)行判斷大州，如果是executor的size是2的冪的話會選擇使用第一種，這樣更快一點(diǎn)垂谢，使用按位操作厦画，否則的話逐個(gè)從線程池中拿。
chooser.next()選擇出來的其實(shí)是NioEventExecutor滥朱，在這會進(jìn)行強(qiáng)轉(zhuǎn)根暑，變成EventLoop類型？這里我們看一下EventLoop的繼承關(guān)系徙邻。
我們回來接著再看register的邏輯排嫌。根據(jù)NioEventLoop的繼承關(guān)系我們可以從SingleThreadEventLoop中找到register的實(shí)現(xiàn)＄掷纾看一下代碼：
public ChannelFuture register(Channel channel) {
return register(new DefaultChannelPromise(channel, this));
}
public ChannelFuture register(final ChannelPromise promise) {
ObjectUtil.checkNotNull(promise, "promise");
promise.channel().unsafe().register(this, promise);
return promise;
}
主要邏輯在下邊的register方法淳地。拿到channel怖糊，然后在調(diào)用channal的unsafe方法拿到一個(gè)Unsafe實(shí)例，然后由Unsafe執(zhí)行register邏輯颇象。又一個(gè)難題伍伤，Unsafe由是干啥的啊遣钳？ 都走到這兒了扰魂，硬著頭皮往下看吧。我們跟進(jìn)一下unsafe方法蕴茴，發(fā)現(xiàn)這個(gè)方法是在Channel類中定義的劝评。我們知道Channel提供了網(wǎng)絡(luò)層的抽象，定義了基本的IO操作：bind倦淀，connect蒋畜，read，write晃听。我們打開Channel源碼百侧，發(fā)現(xiàn)其內(nèi)部定義了Unsafe這個(gè)類。這個(gè)類的注釋是這樣寫的：
/**
* <em>Unsafe</em> operations that should <em>never</em> be called from user-code. These methods
* are only provided to implement the actual transport, and must be invoked from an I/O thread except for the
* following methods:
* <ul>
* <li>{@link #localAddress()}</li>
* <li>{@link #remoteAddress()}</li>
* <li>{@link #closeForcibly()}</li>
* <li>{@link #register(EventLoop, ChannelPromise)}</li>
* <li>{@link #deregister(ChannelPromise)}</li>
* <li>{@link #voidPromise()}</li>
* </ul>
*/
interface Unsafe {
和JDK的Unsafe一樣能扒，這個(gè)東西不是給普通用戶提供的佣渴。那我們找一下它是什么時(shí)候，在什么地方實(shí)例化的初斑。先看一下Channel的子類都有哪些：
先從第一個(gè)子類AbstractChannel看起辛润。果然這個(gè)類定義了unsafe的變量，并且在構(gòu)造器中進(jìn)行了初始化：
private final Unsafe unsafe;
protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline();
}
稍微回想一下见秤，前邊我們看過和channel相關(guān)的代碼是通過ReflectiveChannelFactory創(chuàng)建一個(gè)NioServerSocketChannel實(shí)例砂竖。嚴(yán)重懷疑它們之間有密不可分的關(guān)系【榇穑看一下NioServerSocketChannel的繼承關(guān)系：
果然乎澄，網(wǎng)上數(shù)三輩兒，就是AbstractChannel测摔。我們回退到Factory創(chuàng)建channel的代碼置济，如果不猜錯(cuò)我們肯定還能從那兒進(jìn)入到AbstractChannel。在ReflectiveChannelFactory中我們找到了newChannel的實(shí)現(xiàn)邏輯：
public T newChannel() {
try {
return clazz.newInstance();
} catch (Throwable t) {
throw new ChannelException("Unable to create Channel from class " + clazz, t);
}
}
我們已經(jīng)知道這里clazz其實(shí)就是NioServerSocketChannel锋八。我們進(jìn)入NioServerSocketChannel浙于，找到它的無參的構(gòu)造函數(shù)：
public NioServerSocketChannel() {
this(newSocket(DEFAULT_SELECTOR_PROVIDER));
}
先撇開newSocket這個(gè)方法，因?yàn)檫@需要Java的NIO知識挟纱。先搞清楚Channel的初始化邏輯是第一要?jiǎng)?wù)羞酗。這個(gè)構(gòu)造函數(shù)調(diào)用了另一個(gè)構(gòu)造函數(shù)：
public NioServerSocketChannel(ServerSocketChannel channel) {
super(null, channel, SelectionKey.OP_ACCEPT);
config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}
這個(gè)構(gòu)造函數(shù)里，調(diào)用了父類紊服。我們繼續(xù)跟進(jìn)檀轨，進(jìn)入AbstractNioMessageChannel胸竞。它的構(gòu)造函數(shù)實(shí)現(xiàn)如下：
protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent, ch, readInterestOp);
}
啥也沒干，繼續(xù)向上走裤园，進(jìn)入AbstractNioChannel撤师。
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent);
this.ch = ch;
// SelectionKey.OP_ACCEPT
this.readInterestOp = readInterestOp;
try {
ch.configureBlocking(false);
} catch (IOException e) {
try {
ch.close();
} catch (IOException e2) {
if (logger.isWarnEnabled()) {
logger.warn(
"Failed to close a partially initialized socket.", e2);
}
}
throw new ChannelException("Failed to enter non-blocking mode.", e);
}
}
AbstractNioChannel的構(gòu)造函數(shù)也調(diào)用了其父類剂府，我們在向上走一步拧揽，終于進(jìn)到了AbstractChannel。
protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline();
}
挑重點(diǎn)：unsafe = newUnsafe()腺占。但是newUnsafe方法卻是一個(gè)抽象方法淤袜，我們還得退回到子類去找。在AbstractNioMessageChannel中發(fā)現(xiàn)了newUnsafe的定義：
@Override
protected AbstractNioUnsafe newUnsafe() {
return new NioMessageUnsafe();
}
private final class NioMessageUnsafe extends AbstractNioUnsafe {
///
}
所以衰伯，unsafe其實(shí)是一個(gè)NioMessageUnsafe的實(shí)例铡羡。不要忘了我們的初衷：我們是來看register實(shí)現(xiàn)的。但是NioMessageUnsafe里沒有實(shí)現(xiàn)這個(gè)方法意鲸，只好在去遞歸它的父類烦周。穿過它的父類：AbstractNioUnsafe，我們來到AbstractUnsafe怎顾，終于發(fā)現(xiàn)register方法读慎。不容易啊槐雾！
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
AbstractChannel.this.eventLoop = eventLoop;
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
logger.warn(
"Force-closing a channel whose registration task was not accepted by an event loop: {}",
AbstractChannel.this, t);
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
}
private void register0(ChannelPromise promise) {
doRegister();
}
register首先把eventloop傳給了AbstractChannel的eventloop夭委；然后調(diào)用register0方法，register0方法又調(diào)用doRegister()方法募强。我們跟進(jìn)doRegister方法株灸，發(fā)現(xiàn)它其實(shí)是AbstractChannel中定義的抽象方法。哇擎值，又來慌烧。。鸠儿。繼續(xù)在其子類中搜索屹蚊。在AbstractNioChannel中register方法在靜靜的等著我們：
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
return;
} catch (CancelledKeyException e) {
if (!selected) {
// Force the Selector to select now as the "canceled" SelectionKey may still be
// cached and not removed because no Select.select(..) operation was called yet.
eventLoop().selectNow();
selected = true;
} else {
// We forced a select operation on the selector before but the SelectionKey is still cached
// for whatever reason. JDK bug ?
throw e;
}
}
}
}
javaChannel()拿的是最開始通過factory創(chuàng)建的channel，通過看AbstractNioChannel構(gòu)造器可以知道捆交；eventloop是我們調(diào)用register時(shí)的SingleThreadEventLoop淑翼。看到這才知道最底層原來是這樣的：Channel是注冊到了Selector上品追。
至此玄括，channel的init和register兩個(gè)過程我們都在代碼里找到了線索。而且我們還搞清楚了register的真相：channel register到Selector上肉瓦。差點(diǎn)忘了遭京，我們還有一個(gè)bind邏輯還沒有分析呢胃惜。但是理解bind邏輯需要對Netty的Pipeline有一些了解。所以哪雕，下一篇文章中會重點(diǎn)介紹一下Netty的Pipeline船殉，以及完成bind邏輯的解讀。
