服务端的启动通过ServerBootstrap类来完成，ServerBootstrap内有以下主要属性
ServerBootstrap extends AbstractBootstrap {
//处理channel连接事件的线程组
EventLoopGroup group;
//处理channel其它事件的线程组
EventLoopGroup childGroup;
//创建channel的工厂类
ChannelFactory<? extends C> channelFactory；
//channel相关选项
Map<ChannelOption<?>, Object> options;
//channel相关属性
Map<AttributeKey<?>, Object> attrs；
//handler
ChannelHandler handler;
}
group()方法就是设置两个线程组属性。
channel()方法会new ReflectiveChannelFactory()的工厂赋值给channelFactory属性。
childHandler()设置childHandler属性。
另外还有一个重要的内部类ServerBootstrapAcceptor，
bind方法
bind方法绑定端口启动channel这里是重点,这里实际会调到doBind方法进行处理
来看doBind代码
private ChannelFuture doBind(final SocketAddress localAddress) {
//doBind-1
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
if (regFuture.cause() != null) {
return regFuture;
}
if (regFuture.isDone()) {
// At this point we know that the registration was complete and successful.
ChannelPromise promise = channel.newPromise();
doBind0(regFuture, channel, localAddress, promise);
return promise;
} else {
//...
}
}
初始化和注册
doBind-1会调用initAndRegister方法进行初始channel和注册事件
final ChannelFuture initAndRegister() {
Channel channel = null;
//step1 创建channel
channel = channelFactory.newChannel();
//step2 初始化channel
init(channel);
//step3 注册 这里的group是bossGroup
ChannelFuture regFuture = config().group().register(channel);
return regFuture;
}
step1、创建channel
创建channel是使用的channelFactory。我们上面有说这里工厂实例是ReflectiveChannelFactory。其newChannel就是调用入参class的无参构造函数创建实例。也就是我们传入的NioServerSocketChannel。这里NioServerSocketChannel无参构造方法我们要拿出来看一下。
这里会先根据SelectorProvider创建一个ServerSocketChannel，这都是jdk创建channel的方式。然后调用下面的构造方法
public NioServerSocketChannel(ServerSocketChannel channel) {
//调用父类初始化
super(null, channel, SelectionKey.OP_ACCEPT);
config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}
super调用父类构造方法是AbstractNioChannel类
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent);//这里parent是null
this.ch = ch;
//设置感兴趣的操作 这里是上面传入的SelectionKey.OP_ACCEPT
this.readInterestOp = readInterestOp;
//设置channel为非阻塞
ch.configureBlocking(false);
}
这里又调用父类AbstractChannel的构造方法
protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
//这里会创建一个NioMessageUnsafe类型的unsafe类
unsafe = newUnsafe();
//初始化pipeline
pipeline = newChannelPipeline();
}
其它的不看，先来看下初始化pipleline方法。其实就是创建了一个DefaultChannelPipeline实例。
protected DefaultChannelPipeline(Channel channel) {
this.channel = ObjectUtil.checkNotNull(channel, "channel");
succeededFuture = new SucceededChannelFuture(channel, null);
voidPromise = new VoidChannelPromise(channel, true);
//设置链表头尾
tail = new TailContext(this);
head = new HeadContext(this);
head.next = tail;
tail.prev = head;
}
我们知道Pipeline是一个双向链表，这里就会初始化tail和head。
到这里看到chanel创建好了，还是jdk的nio channel。设置为非阻塞模式，封装成NioServerSocketChannel。并且创建了默认的pipleline。
这里有三个点需要几下，readInterestOp=SelectionKey.OP_ACCEPT，unsafe和pipleline里的HeadContext后面会用到
step2、初始化channel
void init(Channel channel) {
setChannelOptions(channel, newOptionsArray(), logger);
setAttributes(channel, newAttributesArray());
ChannelPipeline p = channel.pipeline();
final EventLoopGroup currentChildGroup = childGroup;
final ChannelHandler currentChildHandler = childHandler;
final Entry<ChannelOption<?>, Object>[] currentChildOptions = newOptionsArray(childOptions);
final Entry<AttributeKey<?>, Object>[] currentChildAttrs = newAttributesArray(childAttrs);
//这里往pipeline里加一个ChannelInitializer
p.addLast(new ChannelInitializer<Channel>() {
//initChannel方法在
@Override
public void initChannel(final Channel ch) {
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(
ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}
这里基本上是把serverboot里的属性设置给channel，然后pipleline里加入一个ChannelInitializer。重写了其initChannel方法。目前不会被调到先不看。不过很重要。主要是ch.eventLoop().execute()这里。这里的ch就是我们的serverchannel，eventLoop是绑定的bossgroup里的一个eventloop。显然这里还没有初始化.
这里调用的pipleline.addLast()方法看一下，其中里面有一步逻辑
//这里的handler就是我们传入的ChannelInitializer
AbstractChannelHandlerContext newCtx = newContext(group, filterName(name, handler), handler)；
if (!registered) {//未注册，成立
newCtx.setAddPending();
callHandlerCallbackLater(newCtx, true);
return this;
}
在addlast方法里会判断是否还未注册，会调用callHandlerCallbackLater()
private void callHandlerCallbackLater(AbstractChannelHandlerContext ctx, boolean added) {
assert !registered;
//added = true
PendingHandlerCallback task = added ? new PendingHandlerAddedTask(ctx) : new PendingHandlerRemovedTask(ctx);
PendingHandlerCallback pending = pendingHandlerCallbackHead;
if (pending == null) {//赋值给pendingHandlerCallbackHead
pendingHandlerCallbackHead = task;
} else {
// Find the tail of the linked-list.
while (pending.next != null) {
pending = pending.next;
}
pending.next = task;
}
}
这里pendingHandlerCallbackHead = 包装(ChannelInitializer)。这一步后面的注册会有回调。
step3、注册channel
第三步config().group().register(channel)；
这里是调用的bossGroup的register方法。前面NioEventLoopGroup部分有说到其register方法。NioEventLoopGroup会拿出一个children也就是NioEventLoop进行与channel绑定。所以从SingleThreadEventLoop的register方法开始看
public ChannelFuture register(Channel channel) {
return register(new DefaultChannelPromise(channel, this));
}
@Override
public ChannelFuture register(final ChannelPromise promise) {
ObjectUtil.checkNotNull(promise, "promise");
//调用unsafe的register方法 这里实例是AbstractUnsafe，是一个AbstractChannel的内部类
promise.channel().unsafe().register(this, promise);
return promise;
}
这里unsafe我们在step1创建channel时候有看到是一个AbstractUnsafe类型，最后调用AbstractUnsafe.register方法
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
//eventLoop是从NioEventLoopGroup拿出来的一个child
AbstractChannel.this.eventLoop = eventLoop;
//判断当前线程和child线程是不是同一个线程 我们这里第一次是主线程 不成立
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
//执行这里
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
}
}
最后执行eventLoop.execute。eventLoop这里是拿出来的一个child是SingleThreadEventLoop extends SingleThreadEventExecutor。这里eventLoop和unsafe类互相调用。最后会调到下面SingleThreadEventExecutor类的重载execute方法
//这里task就是上面传入的runnable。immediate是true
private void execute(Runnable task, boolean immediate) {
boolean inEventLoop = inEventLoop();//还是false
addTask(task);//添加任务
if (!inEventLoop) {
startThread();//启动线程
}
//...
}
这里SingleThreadEventExecutor有一个任务队列Queue taskQueue。addTask就是先将任务加入该队列。然后startThread方法会调用doStartThread真正启动一个线程执行任务。
startThread方法也看一眼
private void startThread() {
if (state == ST_NOT_STARTED) {
if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
boolean success = false;
try {
doStartThread();
success = true;
} finally {
if (!success) {
STATE_UPDATER.compareAndSet(this, ST_STARTED, ST_NOT_STARTED);
}
}
}
}
}
我们看到这里会维护一个state用来标识起没启动过线程，保证只启用一个线程。
doStartThread方法
private void doStartThread() {
assert thread == null;
//这里的executor在创建NioEventLoop时指定的ThreadPerTaskExecutor
//其execut方法就是threadFactory.newThread(command).start();启动一个线程
executor.execute(new Runnable() {
@Override
public void run() {
thread = Thread.currentThread();
if (interrupted) {
thread.interrupt();
}
boolean success = false;
updateLastExecutionTime();
try {//重要的一句 这里this实例是NioEventLoop
SingleThreadEventExecutor.this.run();
success = true;
}
}
});
}
绕来绕去又是run方法又是execute方法。我们这里来总结一下，最后目前的状态。
下面是大致逻辑代码：
NioEventLoop{
ThreadPerTaskExecutor executor;
Queue<Runnable> taskQueue;
void execute(Runnable task){
addTask(task);
startThread();
}
void run(){
...
}
void startThread() {
executor.execute(new Runnable() {
@Override
public void run() {
this.run();
}
});
}
}
1、unsafe调用NioEventLoop.execute()方法执行register0()任务。
2、execute方法首先会将该任务放到taskQueue里。然后startThread启动一个线程。
3、startThread执行其属性executor.execute()方法。executor是ThreadPerTaskExecutor类型，其execute方法会创建并start运行传入的Runnable。所以就是运行起来NioEventLoop.run()方法。
这个时候NioEventLoop里的线程启动起来了，然后任务队列里有一个执行register0()待处理任务。
NioEventLoop.run方法内容：
protected void run() {
int selectCnt = 0;
for (;;) {//死循环，上面创建的线程一直运行
try {
int strategy;
try {
//计算策略值 如果有任务返回Selector.selectNow，否则返回SelectStrategy.SELECT
strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
switch (strategy) {
case SelectStrategy.CONTINUE://-2
continue;
case SelectStrategy.BUSY_WAIT://-3
case SelectStrategy.SELECT://-1
long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
if (curDeadlineNanos == -1L) {
curDeadlineNanos = NONE; // nothing on the calendar
}
nextWakeupNanos.set(curDeadlineNanos);
if (!hasTasks()) {
strategy = select(curDeadlineNanos);
}
}
}
selectCnt++;
cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;//这里默认值50
boolean ranTasks;
if (ioRatio == 100) {
try {
if (strategy > 0) {
processSelectedKeys();
}
} finally {
// Ensure we always run tasks.
ranTasks = runAllTasks();
}
} else if (strategy > 0) {
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
final long ioTime = System.nanoTime() - ioStartTime;
ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
} else {
ranTasks = runAllTasks(0); // This will run the minimum number of tasks
}
}
}
}
运行run方法逻辑，是一个for循环。计算strategy值，
第一次循环： task队列不为空，=Selector.selectNow()。这时候还没有channel注册到selector，selectorNow会返回0.跳过switch判断。ioRatio的判断也不成立，会走最后的else。执行runAllTasks(0)。这个时候才会执行我们第一次AbstractUnsafe.register往taskQueue加的任务,也就是register0方法。
register0方法 AbstractChannel.AbstractUnsafe.register0
private void register0(ChannelPromise promise) {
try {
firstRegistration = true;
//reg1-注册selector
doRegister();
registered = true;
//reg2-回调pipleline里handler的handlerAdded方法
pipeline.invokeHandlerAddedIfNeeded();
//reg3- 发布注册事件
pipeline.fireChannelRegistered();
//reg4- 发布active事件
if (isActive()) {
if (firstRegistration) {
pipeline.fireChannelActive();
} else if (config().isAutoRead()) {
beginRead();
}
}
//...
}
}
reg1-doRegister方法
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
//将channel注册到selector上 注意看这里ops的值是0
selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
return;
}
}
这个时候channel会注册到Selector上，但是关注的事件key值还是0。
reg2
注册完后会调用pipeline.invokeHandlerAddedIfNeeded()方法。第一次注册会调用callHandlerAddedForAllHandlers();方法
PendingHandlerCallback task = pendingHandlerCallbackHead;
while (task != null) {
task.execute();
task = task.next;
}
这里pendingHandlerCallbackHead就是我们step2初始化时候添加的ChannelInitializer。PendingHandlerAddedTask.execute()方法最后会执行到handler.handlerAdd()方法。我们addLast是加的ChannelInitializer类型。其handlerAdd方法如下
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
if (ctx.channel().isRegistered()) {
if (initChannel(ctx)) {//这里入参是handler
// We are done with init the Channel, removing the initializer now.
removeState(ctx);
}
}
}
会调用initChannel(ChannelHandlerContext)方法。
private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
if (initMap.add(ctx)) { // Guard against re-entrance.
try {
initChannel((C) ctx.channel());
} finally {
if (!ctx.isRemoved()) {//移除该handler
ctx.pipeline().remove(this);
}
}
return true;
}
return false;
}
调用initChannel(channel)这个方法是我们step2里从写的方法。然后执行完后会将该handler从pipline里删除
再回头看一下
public void initChannel(final Channel ch) {
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(
ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
主要是ch.eventLoop().execute这里。这里ch.eventLoop()是NioEventLoop，现在已经绑定好了。其execute前面已经介绍过会往任务队列里添加一个任务。
上面是执行runAllTasks第一个任务register0()，register0()最后执行完后又加入一个任务。runAllTasks是个循环只有取不到任务才会跳出，所以会执行第二个刚加入的任务，也就是 pipeline.addLast(new ServerBootstrapAcceptor）。往pipleline里加入请先记住这里pipeline里有一个ServerBootstrapAcceptor。
reg3-发布注册事件
reg4-发布active事件
目前pipleline里只有head和tail两个handler。fireChannelActive()最后会触发handler的channelActive()方法
然而在HeadContext.channelActive()方法最后会调用unsafe.beginRead()方法，然后调用doBeginRead()
protected void doBeginRead() throws Exception {
final SelectionKey selectionKey = this.selectionKey;
if (!selectionKey.isValid()) {
return;
}
readPending = true;
//这里是初始化的0
final int interestOps = selectionKey.interestOps();
//readInterestOp是构造函数设置的值，serverchannel是OP_READ,client创建的channel值是OP_READ
if ((interestOps & readInterestOp) == 0) {//与运算
selectionKey.interestOps(interestOps | readInterestOp);
}
}
这里会修改interestOps。这个时候才开始监听accept事件。就是要等到reg2步ServerBootstrapAcceptor被加入到pipline里之后。后面连接建立时候会有说明为什么。
所有任务执行完成，执行第二次循环
第二次循环：任务已经执行完成为空，这时候strategy = SelectStrategy.SELECT.
会走switch SELECT分支，最后走到select(-1)。内部实现执行selector.select()方法。这个时候就阻塞等待事件的发生。有事件发生，会继续往下走到processSelectedKeys()方法。实际在processSelectedKeysOptimized()方法处理selectKeys。最后具体处理一个selectKey方法是processSelectedKey
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
//
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
try {//不同的key事件处理
int readyOps = k.readyOps();
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
ch.unsafe().forceFlush();
}
if ((readyOps & (SelectionKey.OP_READ| | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
到这里服务端就启动完成了，等待客户端发起i连接事件。
step4、连接处理
这个时候我们启动一个client来连接server，selector就会监听到SelectionKey.OP_ACCEPT事件，就会走unsafe.read()方法。这里server端unsafe实例是NioMessageUnsafe.read方法：
@Override
public void read() {
assert eventLoop().inEventLoop();
final ChannelConfig config = config();
final ChannelPipeline pipeline = pipeline();
final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
allocHandle.reset(config);
boolean closed = false;
//读取数据，这里的readBuf是客户端连接channel
try {
try {
do {
//readBuf是List<NioSocketChannel> 类型，获取所有新连接
int localRead = doReadMessages(readBuf);
if (localRead == 0) {
break;
}
if (localRead < 0) {
closed = true;
break;
}
allocHandle.incMessagesRead(localRead);
} while (continueReading(allocHandle));
} catch (Throwable t) {
exception = t;
}
int size = readBuf.size();
//逐个channel进行处理
for (int i = 0; i < size; i ++) {
readPending = false;
//调用pileline的read方法
pipeline.fireChannelRead(readBuf.get(i));
}
readBuf.clear();
allocHandle.readComplete();
//调用pipleline的ReadComplete
pipeline.fireChannelReadComplete();
}
}
}
doReadMessages就不看了，还是调用nio的accept方法建立channel连接。这里包装成了NioSocketChannel。readInterestOp属性设置的是SelectionKey.OP_READ。
将新建的客户端连接逐个触发ChannelRead方法。这里回想下没有特殊处理现在pipeline里最少有
HeadContext、ServerBootstrapAcceptor、TailContext
ServerBootstrapAcceptor是一个ChannelInboundHandlerAdapter类型的handler。其channelRead方法如下
public void channelRead(ChannelHandlerContext ctx, Object msg) {
//这里的child是新建的客户端channel
final Channel child = (Channel) msg;
//这里的childHandler是我们调用serverchanel.childHandler()方法显示设置的
child.pipeline().addLast(childHandler);
//options 和attrs都是serverchannel初始化显示设置的
setChannelOptions(child, childOptions, logger);
setAttributes(child, childAttrs);
try {//childGroup是workergroup
childGroup.register(child).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (!future.isSuccess()) {
forceClose(child, future.cause());
}
}
});
} catch (Throwable t) {
forceClose(child, t);
}
}
这里的register方法和前面bossgroup的register方法实现是一致的。因为两个group都是NioEventLoopGroup类型。只不过这里是从workgroup拿出来一个child走regiser0进行和新创建的客户端chanel进行绑定，关注的是OP_READ事件。最后一直监听read事件。
最后就是server端会使用bossgroup进行线程channel绑定，监听OP_ACCEPT事件。
clientchannel会和workgroup中的线程进行绑定。监听OP_READ事件。workgroup一个child可以绑定多个channel。同时监听多个channel的READ事件。
启动流程总结：
qa：
1、workgroup是怎么绑定多个clientchannel的？
前面我们知道，新clientchanel连接来了workgroup会分配一个child进行处理。child是怎么分配的呢。
workgroup的next()方法
return executors[(int) Math.abs(idx.getAndIncrement() % executors.length)];
idx绑定一个channel递增一个值，用这个数与child数组长度取余。
这个时候拿出的child会有两种情况
1、未绑定channel还未初始化
这个就是走创建新线程.start()执行NioEventLoop.run()方法。和我们上面分析的服务端启动过程一致，没有问题。
2、child已经使用中，绑定过clientchannel。这个时候有可能处于select()方法阻塞状态。
那么我们新的register被加到taskQueue里岂不是要一直等待执行？
其实不然，这里有唤醒select()逻辑，只是上面没有说。
回到child的execute方法
//immediate 是否立即执行，这里是true
private void execute(Runnable task, boolean immediate) {
boolean inEventLoop = inEventLoop();
addTask(task);
if (!inEventLoop) {
startThread();
//...
}
if (!addTaskWakesUp && immediate) {
//如果有必要，这里唤醒select()阻塞
wakeup(inEventLoop);
}
}
wakeup方法
protected void wakeup(boolean inEventLoop) {
if (!inEventLoop && nextWakeupNanos.getAndSet(AWAKE) != AWAKE) {
selector.wakeup();
}
}
这里就打断了selector.select()的阻塞。然后进入run方法下一次循环判断，会先执行taskQueue里的任务。
