一起學Netty（十八）netty原始碼學習之netty server端原始碼初讀（上）
阿新 • • 發佈：2018-12-29
server端是使用了Reactor模式對nio進行了一些封裝，Reactor模式網上有很多資料，不贅述，瞭解了這個模式開始看原始碼
netty的版本是4.0.21.Final
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.0.21.Final</version>
</dependency>netty端server端的比較經典的程式碼如下：
（程式碼一）
public void start(){ EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap sbs = new ServerBootstrap(); sbs.group(bossGroup,workerGroup); sbs.channel(NioServerSocketChannel.class); sbs.localAddress(new InetSocketAddress(port)); sbs.childHandler(new ChannelInitializer<SocketChannel>() { protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast("decoder", new StringDecoder()); ch.pipeline().addLast("encoder", new StringEncoder()); ch.pipeline().addLast(new HelloWorldServerHandler()); }; }).option(ChannelOption.SO_BACKLOG, 128) .childOption(ChannelOption.SO_KEEPALIVE, true); // 繫結埠，開始接收進來的連線 ChannelFuture future = sbs.bind(port).sync(); System.out.println("Server start listen at " + port ); future.channel().closeFuture().sync(); } catch (Exception e) { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } }
step1：
這是一段普通不能再普通的程式碼了，先看EventLoopGroup這個類，這個類是EventLoop的一個集合，我們先簡單的看下EventLoopGroup的一個具體實現NioEventLoopGroup
public NioEventLoopGroup() { this(0); } public NioEventLoopGroup(int nThreads) { this(nThreads, null); } public NioEventLoopGroup(int nThreads, ThreadFactory threadFactory) { this(nThreads, threadFactory, SelectorProvider.provider()); } public NioEventLoopGroup( int nThreads, ThreadFactory threadFactory, final SelectorProvider selectorProvider) { super(nThreads, threadFactory, selectorProvider); }
NioEventLoopGroup建構函式最後呼叫了它父類MultithreadEventLoopGroup的建構函式：
（程式碼二）
protected MultithreadEventLoopGroup(int nThreads, ThreadFactory threadFactory, Object... args) {
super(nThreads == 0? DEFAULT_EVENT_LOOP_THREADS : nThreads, threadFactory, args);
}其中DEFAULT_EVENT_LOOP_THREADS設定了這個group中的個數，如果我們沒有預設傳遞引數，系統預設會是：
DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
"io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2));EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();看名字就知道哪個EventLoopGroup對應哪一個Reactor模型中的角色了
回到上面一個話題中，我們接著【程式碼二】看，它接著呼叫它的父類MultithreadEventExecutorGroup的建構函式
protected MultithreadEventExecutorGroup(int nThreads, ThreadFactory threadFactory, Object... args) {
if (nThreads <= 0) {
throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
}
if (threadFactory == null) {
threadFactory = newDefaultThreadFactory();
}
children = new SingleThreadEventExecutor[nThreads];
if (isPowerOfTwo(children.length)) {
chooser = new PowerOfTwoEventExecutorChooser();
} else {
chooser = new GenericEventExecutorChooser();
}
for (int i = 0; i < nThreads; i ++) {
boolean success = false;
try {
children[i] = newChild(threadFactory, args);
success = true;
} catch (Exception e) {
// TODO: Think about if this is a good exception type
throw new IllegalStateException("failed to create a child event loop", e);
} finally {
if (!success) {
for (int j = 0; j < i; j ++) {
children[j].shutdownGracefully();
}
for (int j = 0; j < i; j ++) {
EventExecutor e = children[j];
try {
while (!e.isTerminated()) {
e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
}
} catch (InterruptedException interrupted) {
Thread.currentThread().interrupt();
break;
}
}
}
}
}private final EventExecutor[] children;SingleThreadEventExecutor片段一：
private final EventExecutorGroup parent;
private final Queue<Runnable> taskQueue;
final Queue<ScheduledFutureTask<?>> delayedTaskQueue = new PriorityQueue<ScheduledFutureTask<?>>();
private final Thread thread;
private final Semaphore threadLock = new Semaphore(0);
private final Set<Runnable> shutdownHooks = new LinkedHashSet<Runnable>();
private final boolean addTaskWakesUp;SingleThreadEventExecutor片段二：
thread = threadFactory.newThread(new Runnable() {
@Override
public void run() {
boolean success = false;
updateLastExecutionTime();
try {
SingleThreadEventExecutor.this.run();
success = true;
} catch (Throwable t) {
logger.warn("Unexpected exception from an event executor: ", t);
} 回到上文children的初始化的程式碼中來，初始化好之後，開始為一個child賦值：
children[i] = newChild(threadFactory, args);@Override
protected EventExecutor newChild(
ThreadFactory threadFactory, Object... args) throws Exception {
return new NioEventLoop(this, threadFactory, (SelectorProvider) args[0]);
}
NioEventLoop(NioEventLoopGroup parent, ThreadFactory threadFactory, SelectorProvider selectorProvider) {
super(parent, threadFactory, false);
if (selectorProvider == null) {
throw new NullPointerException("selectorProvider");
}
provider = selectorProvider;
selector = openSelector();
}
protected SingleThreadEventLoop(EventLoopGroup parent, ThreadFactory threadFactory, boolean addTaskWakesUp) {
super(parent, threadFactory, addTaskWakesUp);
}好了，到此為止NioEventLoopGroup的原始碼就已經分析完了，此時NioEventLoopGroup中的每一個NioEventLoop中繫結的thread還沒有start，我們接著看
step2：
回到程式碼一：
ServerBootstrap sbs = new ServerBootstrap();
sbs.group(bossGroup,workerGroup);廢話不多說，workerGroup賦值給ServerBootstrap例項的childGroup，bossGroup賦值給AbstractBootstrap的group,具體程式碼如下：
ServerBootstrap.java
private volatile EventLoopGroup childGroup;
public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
super.group(parentGroup);
if (childGroup == null) {
throw new NullPointerException("childGroup");
}
if (this.childGroup != null) {
throw new IllegalStateException("childGroup set already");
}
this.childGroup = childGroup;
return this;
}private volatile EventLoopGroup group;
AbstractBootstrap(AbstractBootstrap<B, C> bootstrap) {
group = bootstrap.group;
channelFactory = bootstrap.channelFactory;
handler = bootstrap.handler;
localAddress = bootstrap.localAddress;
synchronized (bootstrap.options) {
options.putAll(bootstrap.options);
}
synchronized (bootstrap.attrs) {
attrs.putAll(bootstrap.attrs);
}
}sbs.channel(NioServerSocketChannel.class);去掉一些handler類，我們看：
ChannelFuture future = sbs.bind(port).sync(); public ChannelFuture bind(int inetPort) {
return bind(new InetSocketAddress(inetPort));
}
public ChannelFuture bind(SocketAddress localAddress) {
validate();
if (localAddress == null) {
throw new NullPointerException("localAddress");
}
return doBind(localAddress);
}
private ChannelFuture doBind(final SocketAddress localAddress) {
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
if (regFuture.cause() != null) {
return regFuture;
}
final ChannelPromise promise;
if (regFuture.isDone()) {
promise = channel.newPromise();
doBind0(regFuture, channel, localAddress, promise);
} else {
// Registration future is almost always fulfilled already, but just in case it's not.
promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
doBind0(regFuture, channel, localAddress, promise);
}
});
}
return promise;
}最後呼叫了AbstractBootstrap（這個我們剛才做過處理，把bossGroup賦值了AbstractBootstrap的group，也就是說這個AbstractBootstrap獲取到一個thread執行器了）
言歸正傳，我們看doBind方法：
（程式碼三）
final ChannelFuture regFuture = initAndRegister();final ChannelFuture initAndRegister() {
final Channel channel = channelFactory().newChannel();
try {
init(channel);
} catch (Throwable t) {
channel.unsafe().closeForcibly();
return channel.newFailedFuture(t);
}
ChannelFuture regFuture = group().register(channel);
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
return regFuture;
}（程式碼四）
@Override
public T newChannel() {
try {
return clazz.newInstance();
} catch (Throwable t) {
throw new ChannelException("Unable to create Channel from class " + clazz, t);
}
}protected AbstractChannel(Channel parent) {
this.parent = parent;
unsafe = newUnsafe();
pipeline = new DefaultChannelPipeline(this);
}原始碼位於io.netty.channel.Channel的Unsafe
interface Unsafe {
/**
* Return the {@link SocketAddress} to which is bound local or
* {@code null} if none.
*/
SocketAddress localAddress();
/**
* Return the {@link SocketAddress} to which is bound remote or
* {@code null} if none is bound yet.
*/
SocketAddress remoteAddress();
/**
* Register the {@link Channel} of the {@link ChannelPromise} with the {@link EventLoop} and notify
* the {@link ChannelFuture} once the registration was complete.
*/
void register(EventLoop eventLoop, ChannelPromise promise);我列出部分程式碼的意思就是這個unsafe其實是束縛於Channel的，但也提供了channel的一些方法，比如register(EventLoop eventLoop, ChannelPromise promise)這個方法，將channel註冊到eventLoop上去，這是很關鍵的，我們再回歸到AbstractChannel的newUnsafe方法上來，追蹤原始碼最終發現：
@Override
protected AbstractNioUnsafe newUnsafe() {
return new NioMessageUnsafe();
}好了，最終NioServerSocketChannel這個channel對應的Unsafe是NioMessageUnsafe，關於AbstractChannel的newUnsafe這只是一個插曲（下文會用到），我們還是回到【程式碼三】
final Channel channel = channelFactory().newChannel();
init(channel);init的方法，很重要，這裡面規範了channel的一些屬性，我們看看位於ServerBootstrap中的init方法
@Override
void init(Channel channel) throws Exception {
final Map<ChannelOption<?>, Object> options = options();
synchronized (options) {
channel.config().setOptions(options);
}
final Map<AttributeKey<?>, Object> attrs = attrs();
synchronized (attrs) {
for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
@SuppressWarnings("unchecked")
AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
channel.attr(key).set(e.getValue());
}
}
ChannelPipeline p = channel.pipeline();
if (handler() != null) {
p.addLast(handler());
}
final EventLoopGroup currentChildGroup = childGroup;
final ChannelHandler currentChildHandler = childHandler;
final Entry<ChannelOption<?>, Object>[] currentChildOptions;
final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
synchronized (childOptions) {
currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
}
synchronized (childAttrs) {
currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
}
p.addLast(new ChannelInitializer<Channel>() {
@Override
public void initChannel(Channel ch) throws Exception {
ch.pipeline().addLast(new ServerBootstrapAcceptor(
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}回到initAndRegister這個方法中來，init完了channel，最最關鍵的一步就是
ChannelFuture regFuture = group().register(channel);還是不忘初心，在NIO的程式設計中有一個很重要的步驟，我們可以將這個步驟寫出來：
SocketChannel channel = SocketChannel.open();
channel.configureBlocking(false);
InetSocketAddress s = new InetSocketAddress("localhost", 2000);
channel.connect(s);
Selector selector = Selector.open();
channel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ);這是一段很經典的純的Java的NIO的的client程式碼。
這邊我們看channel.register這行程式碼，我們需要將channel註冊到selector上，並且傳遞自己感興趣的事件引數，這樣當有selector上有事件發生的是，就會通知註冊的channel觸發對應的事件
我們在分析netty的時候，我們知道在分析NioEventLoopGroup原始碼的時候，我們知道每一個NioEventLoopGroup都維護了一個selector，再後文中我們也分析了一個channel的建立過程，但是還沒有進行selector和selector進行register
這邊的group().register(channel)最後呼叫的是：
channel.unsafe().register(this, promise);我們也注意下register的入參，this指代的是SingleThreadEventLoop這個例項（其實我要的不是這個例項，而是這個例項中的selector（位於NioEventLoop.java中））
最後的註冊動作是在AbstractNioChannel.java中完成的
@Override
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
selectionKey = javaChannel().register(eventLoop().selector, 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;
}
}
}
}protected SelectableChannel javaChannel() {
return ch;
}private final SelectableChannel ch;估計這邊大家要問這個ch是什麼時候賦值的，請看【程式碼四】
在newInstant的時候建立NioServerSocketChannel例項的時候，進行了賦值，這邊ch其實就是NioServerSocketChannel
好了，到此為止，我們基本分析了Netty的server端的啟動過程，你也許會問server的thread貌似還沒有start吧，其實不然，你可以仔細觀察一下register0這個方法，是這麼呼叫的：
try {
eventLoop.execute(new OneTimeTask() {
@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);
}@Override
public void execute(Runnable task) {
if (task == null) {
throw new NullPointerException("task");
}
boolean inEventLoop = inEventLoop();
if (inEventLoop) {
addTask(task);
} else {
startThread();
addTask(task);
if (isShutdown() && removeTask(task)) {
reject();
}
}
if (!addTaskWakesUp && wakesUpForTask(task)) {
wakeup(inEventLoop);
}
}最後會呼叫startThread方法，歸於自然，歸於本心，呼叫SingleThreadEventExecutor的
private void startThread() {
if (STATE_UPDATER.get(this) == ST_NOT_STARTED) {
if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
delayedTaskQueue.add(new ScheduledFutureTask<Void>(
this, delayedTaskQueue, Executors.<Void>callable(new PurgeTask(), null),
ScheduledFutureTask.deadlineNanos(SCHEDULE_PURGE_INTERVAL), -SCHEDULE_PURGE_INTERVAL));
thread.start();
}
}
}thrad的start呼叫的方法就是我們剛才片段二種的
SingleThreadEventExecutor.this.run();好了，先分析這麼多吧，我們進行總結一下，分析太多，我自己也記不過來
1）Netty的server端程式碼一開始初始化了兩個EventLoopGroup，其實就是初始化EventLoop，每一個EventLoop的具體實現就是維護了一個任務佇列，一個延遲任務佇列，一個thread，並且每一個EventLoop都有一個屬於自己的Executor執行器，這樣做的好處就是每一個唯一的thread去不停的迴圈呼叫，去執行任務佇列和延遲任務佇列中的task，沒有了上下文的切換們還要記得每一個EventLoop還初始化了一個selector，關於selector的建立，netty做了很大的優化，使其與CPU更加親和
（中間還忘記分析了，CPU是2的倍數的時候，Netty的優化，大家可以自己看下）
2）關於serverBootstrap的初始化，主要就是做了channel的建立，channel的執行器的繫結，option屬性的配置，繫結埠，這些配置好了之後就是channel和selector的繫結，繫結的時候，順帶啟動一些AbstractBootstrap中的thread，讓其進行無限迴圈中
3）關於細節
①serverBootstrap中在channelPipeline中偷偷地加了一個ServerBootstrapAcceptor
②serverBootstrap中的boss執行緒對應的unsafe物件是NioMessageUnsafe例項