通过走读netty源码来学习netty（二）（基于netty-4.0.19.Final）

今天继续走读netty源码

 

 上次走读到NioEventLoop的register方法，进入NioEventLoop类中，发现register方法并不在该类中，而是在父类SingleThreadEventLoop中，进入父类的对应方法后，发现最终调用的是unsafe的register方法

channel.unsafe().register(this, promise);

 有两个参数的register方法是在AbstractChannel类中定义，继续进入AbstractChannel类中查看，发现调用的是unsafe的register方法

channel.unsafe().register(this, promise);

 而unsafe的register方法定义在AbstractUnsafe内部类中，进入到AbstractUnsafe类的对应方法中就会看到如下判断

if (eventLoop.inEventLoop()) {
	register0(promise);
} else {
	try {
		eventLoop.execute(new OneTimeTask() {
			@Override
			public void run() {
				register0(promise);
			}
		});
	} catch (Throwable t) {	
	}
}

先看inEventLoop方法干了什么事

inEventLoop方法是定义在AbstractEventExecutor类中的，而AbstractEventExecutor类中对应的方法又调用了子类SingleThreadEventExecutor的inEventLoop方法，如下是SingleThreadEventExecutor的inEventLoop方法

 public boolean inEventLoop(Thread thread) {
        return thread == this.thread;
    }

 其中参数thread是代表当前线程，所以该方法其实在判断当前线程是否是SingleThreadEventExecutor类中定义的thread。由于我们现在是从main方法中进入的，当然不是了，所以inEventLoop方法返回false

 

 程序应该进入else分支了，else分支调用的是execute方法，传入的是OneTimeTask类的实例。

从NioEventLoop方法中查找execute方法，就会发现execute方法在SingleThreadEventExecutor类中定义

 

boolean inEventLoop = inEventLoop();
if (inEventLoop) {
	addTask(task);
} else {
	startThread();
	addTask(task);
	if (isShutdown() && removeTask(task)) {
		reject();
	}
}

if (!addTaskWakesUp && wakesUpForTask(task)) {
	wakeup(inEventLoop);
}

 刚才查看inEventLoop方法，返回的是false，所以直接进入else分支执行

 

先看startThread方法

 

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();
		}
	}
}

先判断线程有没有启动，如果没有启动，现将STATE_UPDATER以CAS方式设置为启动，成功后在delayedTaskQueue中增加一个任务，然后启动thread（注意，现在程序除了主线程外，增加了thread线程）

 

然后再看addTask方法，这个方法很简单，将传入的OneTimeTask增加到taskQueue队列中

最后看是否需要调用wakeup方法

  1、看addTaskWakesUp变量是否为true，还是需要回到NioEventLoopGroup类的newChild方法中去寻找，最后发现addTaskWakesUp变量为false，

    2、看wakesUpForTask方法返回是否为true，wakesUpForTask方法在SingleThreadEventExecutor中定义，但该方法是protected，所以还需要查找是否有覆盖（这个是走读代码经常会遇到的问题，eclipse中在调用的方法上直接按Ctrl+T），发现SingleThreadEventLoop中重写了该方法

protected boolean wakesUpForTask(Runnable task) {
	return !(task instanceof NonWakeupRunnable);
}

 task当然不是NonWakeupRunnable类型了，所以该方法返回true

 

进入wakeup方法，发现wakeup做的唯一一件事是taskQueue中增加了一个WAKEUP_TASK，而WAKEUP_TASK的run方法什么都不做。

 

终于register方法走读完了，这个应该算是netty中一个比较重要的方法吧

 

现在回到SingleThreadEventLoop的register方法，发现register方法返回的是DefaultChannelPromise的实例，而这个promise一直以参数方式传递到到AbstractUnsafe类中register0方法中，而register0方法是在thread线程中调用，所以promise是在线程中设置值的

 

现在回到了AbstractBootstrap类的doBind方法，查看doBind，发现返回的promise并不是register方法返回的promise，而是一个新的promise（同样是DefaultChannelPromise类的实例）

 

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 DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE);
	regFuture.addListener(new ChannelFutureListener() {
		@Override
		public void operationComplete(ChannelFuture future) throws Exception {
			doBind0(regFuture, channel, localAddress, promise);
		}
	});
}

 由于返回的regFuture是在单独的线程中设置值的，所以我们可以先假定isDone方法返回的是false（线程还没有设置对应的值），所以进入else分支，else分支中主要是regFuture增加了一个监听器

 

 

终于又回到了EchoServer类的run方法，bind方法调用完成后开始调用sync方法

 ChannelFuture f = b.bind(port).sync();

 检查代码，发现sync调用最终会调用到DefaultPromise类的await方法，而await方法最终调用了Object的wait方法等待

synchronized (this) {
	while (!isDone()) {
		checkDeadLock();
		incWaiters();
		try {
			wait();
		} finally {
			decWaiters();
		}
	}
}

 

至此，主线程的调用进入了等待状态。

下次开始走读主线程调用register方法启动的SingleThreadEventExecutor类中thread线程代码