bossGroup

向bossGroup里注册通道

流程图

 书接上文

ServerBootstrap.bind(hostname, port)

final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            channel = channelFactory.newChannel();
            init(channel);
        } catch (Throwable t) {
            if (channel != null) {
                // channel can be null if newChannel crashed (eg SocketException("too many open files"))
                channel.unsafe().closeForcibly();
                // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
                return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
            }
            // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
            return new DefaultChannelPromise(new FailedChannel(), 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;
    }

上篇文章说了init方法，接下去就是注册通道

config().group().register(channel)方法

首先调用的是MultithreadEventLoopGroup.register(Channel channel)方法

public ChannelFuture register(Channel channel) {
        return next().register(channel);
    }

public EventLoop next() {
        return (EventLoop) super.next();
    }

调用MultithreadEventExecutorGroup的next()方法

public EventExecutor next() {
    return chooser.next();
}

chooser选择器。在MultithreadEventExecutorGroup构造方法中实例化。

protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                            EventExecutorChooserFactory chooserFactory, Object... args) {
        if (nThreads <= 0) {
            throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
        }

        if (executor == null) {
            executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
        }

        children = new EventExecutor[nThreads];

        for (int i = 0; i < nThreads; i ++) {
            boolean success = false;
            try {
                children[i] = newChild(executor, 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) {
                            // Let the caller handle the interruption.
                            Thread.currentThread().interrupt();
                            break;
                        }
                    }
                }
            }
        }

        chooser = chooserFactory.newChooser(children);//实例化选择器

        final FutureListener<Object> terminationListener = new FutureListener<Object>() {
            @Override
            public void operationComplete(Future<Object> future) throws Exception {
                if (terminatedChildren.incrementAndGet() == children.length) {
                    terminationFuture.setSuccess(null);
                }
            }
        };

        for (EventExecutor e: children) {
            e.terminationFuture().addListener(terminationListener);
        }

        Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
        Collections.addAll(childrenSet, children);
        readonlyChildren = Collections.unmodifiableSet(childrenSet);
    }

chooserFactory.newChooser(children);

其实就前面讲过的选择器工厂DefaultEventExecutorChooserFactory，如何选择一个执行器。

public EventExecutorChooser newChooser(EventExecutor[] executors) {
        if (isPowerOfTwo(executors.length)) {
            return new PowerOfTwoEventExecutorChooser(executors);
        } else {
            return new GenericEventExecutorChooser(executors);
        }
    }

chooser.next();

private static final class PowerOfTwoEventExecutorChooser implements EventExecutorChooser {
        private final AtomicInteger idx = new AtomicInteger();
        private final EventExecutor[] executors;

        PowerOfTwoEventExecutorChooser(EventExecutor[] executors) {
            this.executors = executors;
        }

        @Override
        public EventExecutor next() {
            return executors[idx.getAndIncrement() & executors.length - 1];
        }
    }

    private static final class GenericEventExecutorChooser implements EventExecutorChooser {
        // Use a 'long' counter to avoid non-round-robin behaviour at the 32-bit overflow boundary.
        // The 64-bit long solves this by placing the overflow so far into the future, that no system
        // will encounter this in practice.
        private final AtomicLong idx = new AtomicLong();
        private final EventExecutor[] executors;

        GenericEventExecutorChooser(EventExecutor[] executors) {
            this.executors = executors;
        }

        @Override
        public EventExecutor next() {
            return executors[(int) Math.abs(idx.getAndIncrement() % executors.length)];
        }
    }

 选到一个NioEventLoop，执行注册，调用的是NioEventLoop的register。子类没有这个方法，由父类SingleThreadEventLoop实现。

public ChannelFuture register(Channel channel) {
        return register(new DefaultChannelPromise(channel, this));
    }

创建了一个DefaultChannelPromise把通道和NioEventLoop都封装进去，DefaultChannelPromise是对Future做了加强，也是异步回调，可以设置很多监听，也可以将结果写入。

public ChannelFuture register(final ChannelPromise promise) {
        ObjectUtil.checkNotNull(promise, "promise");
        promise.channel().unsafe().register(this, promise);
        return promise;
    }

Channel.Unsafe

这里的Unsafe不是JDK的Unsafe，是netty封装的Channel的内部类，因为他是一些操作NIO底层的方法，不建议外部用的，所有也叫Unsafe。我们所说的把通道注册进bossGroup实际上就是调用他的实现类。promise.channel()方法返回NioServerSocketChannel,他的unsafe方法由它的父类AbstractNioChannel实现。

public NioUnsafe unsafe() {

    return (NioUnsafe) super.unsafe();
}
//父类unsafe方法
public Unsafe unsafe() {

    return unsafe;
}

返回值 在初始化AbstractChannel时 指定

protected AbstractChannel(Channel parent) {

    this.parent = parent;
   id = newId();
   unsafe = newUnsafe();
   pipeline = newChannelPipeline();
}

newUnsafe(),又由子类AbstractNioMessageChannel实现

protected AbstractNioUnsafe newUnsafe() {

    return new NioMessageUnsafe();
}

所以config().group().register(channel)调用的是NioMessageUnsafe.register方法,但是自己没有这个方法,所以最终调用父类AbstractUnsafe的register方法,AbstractUnsafe为AbstractChannel的内部类.

这个就是真正的注册方法：

public final void register(EventLoop eventLoop, final ChannelPromise promise) {
            ObjectUtil.checkNotNull(eventLoop, "eventLoop");
            if (isRegistered()) {
                promise.setFailure(new IllegalStateException("registered to an event loop already"));
                return;
            }
            if (!isCompatible(eventLoop)) {
                promise.setFailure(
                        new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
                return;
            }

            AbstractChannel.this.eventLoop = eventLoop;//设置通道的事件循环，1对1，只设置一次
　　　　　　　//只能当前线程是eventLoop的线程才可以注册，防止多线程并发问题，所以即使多线程来操作，也是安全的，会按照一定顺序提交到任务队列里
            if (eventLoop.inEventLoop()) {
                register0(promise);
            } else {//否则就当做任务提交给eventLoop的任务队列
                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);
                }
            }
        }

这里有个很巧妙的点，就是eventLoop.inEventLoop()这个的判断，就是判断调用这个方法的是不是eventLoop的线程，如果是，那就是同一个线程调用，直接就注册，否则属于多线程调用，可能会有问题，所以还是提交一个任务给eventLoop的线程去执行，这样就是单线程，不会有线程安全问题。

此时当前线程是main肯定不是事件循环里的线程，事件循环里的线程还没创建呢，所以会提交到队列。eventLoop为NioEventLoop，执行父类SingleThreadEventExecutor的execute方法。

public void execute(Runnable task) {

    ObjectUtil.checkNotNull(task, "task");
   execute(task, !(task instanceof LazyRunnable) && wakesUpForTask(task));
}

private void execute(Runnable task, boolean immediate) {
        boolean inEventLoop = inEventLoop();
        addTask(task);//添加任务到taskQueue
        if (!inEventLoop) {//当前线程不是事件循环线程
            startThread();//让线程工厂开启线程
            if (isShutdown()) {
                boolean reject = false;
                try {
                    if (removeTask(task)) {
                        reject = true;
                    }
                } catch (UnsupportedOperationException e) {
                    // The task queue does not support removal so the best thing we can do is to just move on and
                    // hope we will be able to pick-up the task before its completely terminated.
                    // In worst case we will log on termination.
                }
                if (reject) {
                    reject();
                }
            }
        }

        if (!addTaskWakesUp && immediate) {
            wakeup(inEventLoop);//唤醒线程，添加了一个空的任务去唤醒
        }
    }

task为register0(promise);方法,addTask是给taskqueue添加task,

startThread();

private static final int ST_NOT_STARTED = 1;//没启动
    private static final int ST_STARTED = 2;//启动了
    private static final int ST_SHUTTING_DOWN = 3;//正在关闭中
    private static final int ST_SHUTDOWN = 4;//关闭了
    private static final int ST_TERMINATED = 5;//终止了

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

doStartThread()

其实里面就是调用SingleThreadEventExecutor.this.run();，这里其实executor.execute已经启动了新的线程来执行new Runnable()里的任务，也就是执行NioEventLoop. run()。

private void doStartThread() {
        assert thread == null;
        executor.execute(new Runnable() {
            @Override
            public void run() {
                thread = Thread.currentThread();
                if (interrupted) {
                    thread.interrupt();
                }

                boolean success = false;
                updateLastExecutionTime();
                try {
                    SingleThreadEventExecutor.this.run();
                    success = true;
                } catch (Throwable t) {
                    logger.warn("Unexpected exception from an event executor: ", t);
                } finally {
                    for (;;) {
                        int oldState = state;
                        if (oldState >= ST_SHUTTING_DOWN || STATE_UPDATER.compareAndSet(
                                SingleThreadEventExecutor.this, oldState, ST_SHUTTING_DOWN)) {
                            break;
                        }
                    }

                    // Check if confirmShutdown() was called at the end of the loop.
                    if (success && gracefulShutdownStartTime == 0) {
                        if (logger.isErrorEnabled()) {
                            logger.error("Buggy " + EventExecutor.class.getSimpleName() + " implementation; " +
                                    SingleThreadEventExecutor.class.getSimpleName() + ".confirmShutdown() must " +
                                    "be called before run() implementation terminates.");
                        }
                    }

                    try {
                        // Run all remaining tasks and shutdown hooks. At this point the event loop
                        // is in ST_SHUTTING_DOWN state still accepting tasks which is needed for
                        // graceful shutdown with quietPeriod.
                        for (;;) {
                            if (confirmShutdown()) {
                                break;
                            }
                        }

                        // Now we want to make sure no more tasks can be added from this point. This is
                        // achieved by switching the state. Any new tasks beyond this point will be rejected.
                        for (;;) {
                            int oldState = state;
                            if (oldState >= ST_SHUTDOWN || STATE_UPDATER.compareAndSet(
                                    SingleThreadEventExecutor.this, oldState, ST_SHUTDOWN)) {
                                break;
                            }
                        }

                        // We have the final set of tasks in the queue now, no more can be added, run all remaining.
                        // No need to loop here, this is the final pass.
                        confirmShutdown();
                    } finally {
                        try {
                            cleanup();
                        } finally {
                            // Lets remove all FastThreadLocals for the Thread as we are about to terminate and notify
                            // the future. The user may block on the future and once it unblocks the JVM may terminate
                            // and start unloading classes.
                            // See https://github.com/netty/netty/issues/6596.
                            FastThreadLocal.removeAll();

                            STATE_UPDATER.set(SingleThreadEventExecutor.this, ST_TERMINATED);
                            threadLock.countDown();
                            int numUserTasks = drainTasks();
                            if (numUserTasks > 0 && logger.isWarnEnabled()) {
                                logger.warn("An event executor terminated with " +
                                        "non-empty task queue (" + numUserTasks + ')');
                            }
                            terminationFuture.setSuccess(null);
                        }
                    }
                }
            }
        });
    }

executor在SingleThreadEventExecutor构造方法中实例化，this.executor = ThreadExecutorMap.apply(executor, this);

public static Executor apply(final Executor executor, final EventExecutor eventExecutor) {
        ObjectUtil.checkNotNull(executor, "executor");
        ObjectUtil.checkNotNull(eventExecutor, "eventExecutor");
        return new Executor() {
            @Override
            public void execute(final Runnable command) {
                executor.execute(apply(command, eventExecutor));
            }
        };
    }

apply方法中的executor为SingleThreadEventExecutor构造方法的入参参数,根据继承关系可知SingleThreadEventExecutor构造方法在NioEventLoop构造方法中调用了,而NioEventLoop构造方法在之前已说明,在

MultithreadEventExecutorGroup的构造方法中调用所以apply方法中的executor为ThreadPerTaskExecutor

ThreadPerTaskExecutor该类的execute比较简单,利用DefaultThreadFactory实话为thread实现异步执行,但是thread为FastThreadLocalThread而不是原生的thread.

apply方法中的eventExecutor为SingleThreadEventExecutor对象eventExecutor

apply方法内部的apply

public static Runnable apply(final Runnable command, final EventExecutor eventExecutor) {
        ObjectUtil.checkNotNull(command, "command");
        ObjectUtil.checkNotNull(eventExecutor, "eventExecutor");
        return new Runnable() {
            @Override
            public void run() {
                setCurrentEventExecutor(eventExecutor);
                try {
                    command.run();
                } finally {
                    setCurrentEventExecutor(null);
                }
            }
        };
    }

setCurrentEventExecutor(eventExecutor);设置本地线程变量,利用了FastThreadLocal

command.run()方法调用的为doStartThread()方法内部的异步执行方法体SingleThreadEventExecutor.this.run();下文详解

 

sync()

main线程如果完成了注册流程后，就会调用sync()尝试阻塞，调用的就是：

ChannelFuture channelFuture = b.bind(hostname, port).sync();

最终到了DefaultPromise的await()方法：

 1 public Promise<V> await() throws InterruptedException {
 2         if (isDone()) {//完成了就返回
 3             return this;
 4         }
 5 
 6         if (Thread.interrupted()) {
 7             throw new InterruptedException(toString());
 8         }
 9 
10         checkDeadLock();
11 
12         synchronized (this) {
13             while (!isDone()) {//判断是否完成
14                 incWaiters();
15                 try {
16                     wait();//阻塞
17                 } finally {
18                     decWaiters();
19                 }
20             }
21         }
22         return this;
23     }

如果这个时候注册和绑定端口完成了，就会返回，否则就会wait();阻塞。

safeSetSuccess(promise);

绑定是事件循环的线程去做的，完成绑定后会调用safeSetSuccess(promise);方法，设置成功的结果，唤醒阻塞的主线程。最终是调用DefaultPromise的checkNotifyWaiters：

protected final void safeSetSuccess(ChannelPromise promise) {
            if (!(promise instanceof VoidChannelPromise) && !promise.trySuccess()) {
                logger.warn("Failed to mark a promise as success because it is done already: {}", promise);
            }
        }

调用DefaultChannelPromise的trySuccess()

private boolean setValue0(Object objResult) {
        if (RESULT_UPDATER.compareAndSet(this, null, objResult) ||
            RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) {
            if (checkNotifyWaiters()) {
                notifyListeners();
            }
            return true;
        }
        return false;
    }

方法checkNotifyWaiters()

private synchronized boolean checkNotifyWaiters() {
        if (waiters > 0) {
            notifyAll();
        }
        return listeners != null;
    }

方法notifyListeners()

private void notifyListeners() {
        EventExecutor executor = executor();
        if (executor.inEventLoop()) {
            final InternalThreadLocalMap threadLocals = InternalThreadLocalMap.get();
            final int stackDepth = threadLocals.futureListenerStackDepth();
            if (stackDepth < MAX_LISTENER_STACK_DEPTH) {
                threadLocals.setFutureListenerStackDepth(stackDepth + 1);
                try {
                    notifyListenersNow();
                } finally {
                    threadLocals.setFutureListenerStackDepth(stackDepth);
                }
                return;
            }
        }

        safeExecute(executor, new Runnable() {
            @Override
            public void run() {
                notifyListenersNow();
            }
        });
    }

注册绑定监听并阻塞

阻塞被唤醒后，就注册一个监听的回调，然后阻塞关闭事件，其实就意味着不关闭就永远运行着了：

ChannelFuture cf = bootstrap.bind(8888).sync();
            cf.addListener((ChannelFutureListener) future -> {
                if (cf.isSuccess()) {
                    System.out.println("监听端口 8888 成功");
                } else {
                    System.out.println("监听端口 8888 失败");
                }
            });
            System.out.println("服务器开始提供服务");
            cf.channel().closeFuture().sync();

当然也可以这样写，先添加监听，然后阻塞，反正结果都是一样的：

ChannelFuture cf = bootstrap.bind(8888);
            cf.addListener((ChannelFutureListener) future -> {
                if (cf.isSuccess()) {
                    System.out.println("监听端口 8888 成功");
                } else {
                    System.out.println("监听端口 8888 失败");
                }
            });
            System.out.println("服务器开始提供服务");
            cf.sync();

如果addListener是添加在同步后面的，此时可能事件循环线程已经完成了注册和绑定，addListener里面就会提交一个任务，然后事件循环会去执行这个任务。

控制台打印结果

服务器开始提供服务
监听端口 8888 成功

至此注册流程基本结束，后续就看事件循环中的新线程如何运行了。