Netty服务端启动属性绑定及Channel线程创建源码阅读

从绑定端口bind()方法开始

AbstractBootstrap.java 从构造方法一路跟踪

/**
 * Create a new {@link Channel} and bind it.
 */
public ChannelFuture bind(int inetPort) {
    return bind(new InetSocketAddress(inetPort));
}
/**
 * Create a new {@link Channel} and bind it.
 */
public ChannelFuture bind(String inetHost, int inetPort) {
    return bind(SocketUtils.socketAddress(inetHost, inetPort));
}
/**
 * Create a new {@link Channel} and bind it.
 */
public ChannelFuture bind(InetAddress inetHost, int inetPort) {
    return bind(new InetSocketAddress(inetHost, inetPort));
}
/**
 * Create a new {@link Channel} and bind it.
 */
public ChannelFuture bind(SocketAddress localAddress) {
    // 验证group与channelFactory属性是否为null
    validate();
    if (localAddress == null) {
        throw new NullPointerException("localAddress");
    }
    return doBind(localAddress);
}
private ChannelFuture doBind(final SocketAddress localAddress) {
    // 以异步的方式创建、初始化一个channel，并将其注册到selector
      // 准备跟踪initAndRegister() 方法
    final ChannelFuture regFuture = initAndRegister();
    // 从future中获取channel
    final Channel channel = regFuture.channel();
    // 若在执行异步操作过程中出现了异常，则直接返回这个future
    if (regFuture.cause() != null) {
        return regFuture;
    }
    // 处理当前异步操作完成（任务正常结束，或执行过程中发生异常，或任务被取消）的情况
    if (regFuture.isDone()) {
        // At this point we know that the registration was complete and successful.
        // 创建一个channelPromise实例
        ChannelPromise promise = channel.newPromise();
        // 继续绑定
        doBind0(regFuture, channel, localAddress, promise);
        return promise;
    } else {  // 处理当前异步操作目前尚未完成的情况
        // Registration future is almost always fulfilled already, but just in case it's not.
        // Pending，悬而未决的
        final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
        // 为future添加监听，当异步操作完成时，会触发该回调的执行
        regFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                Throwable cause = future.cause();
                if (cause != null) {
                    // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
                    // IllegalStateException once we try to access the EventLoop of the Channel.
                    // 修改promise的值
                    promise.setFailure(cause);
                } else {
                    // Registration was successful, so set the correct executor to use.
                    // See https://github.com/netty/netty/issues/2586
                    promise.registered();
                    doBind0(regFuture, channel, localAddress, promise);
                }
            }
        });
        return promise;
    }
}

初始化并注册ChannelFuture

final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
        // 创建parentChannel
        // 其实这里面是使用反射newInstance 创建出一个Channel无参对象
        channel = channelFactory.newChannel();
        // 对象创建完后，开始初始化该channel
        init(channel);
    } catch (Throwable t) {
        if (channel != null) { // 若条件为true，说明channel创建成功，但初始化时出现问题
            // channel can be null if newChannel crashed (eg SocketException("too many open files"))
            // 将channel强制关闭
            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);
        }
        // 代码能走到这里，说明创建channel过程中出现了问题
        // 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);
    }
    // 注册parentChannel(该过程中从group中选择出了eventLoop与channel进行了绑定，并创建启动了这个线程）
    ChannelFuture regFuture = config().group().register(channel);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }
    // If we are here and the promise is not failed, it's one of the following cases:
    // 1) If we attempted registration from the event loop, the registration has been completed at this point.
    //    i.e. It's safe to attempt bind() or connect() now because the channel has been registered.
    // 2) If we attempted registration from the other thread, the registration request has been successfully
    //    added to the event loop's task queue for later execution.
    //    i.e. It's safe to attempt bind() or connect() now:
    //         because bind() or connect() will be executed *after* the scheduled registration task is executed
    //         because register(), bind(), and connect() are all bound to the same thread.
    return regFuture;
}

我们现在查看的源代码是服务端的，所以选择ServerBootstrap.java进行跟踪init() 初始化方法

@Override
void init(Channel channel) throws Exception {
    // 获取ServerBootstrap中的options属性
    final Map<ChannelOption<?>, Object> options = options0();
    synchronized (options) {
        //这个地方就是将options属性初始化到channel中，而这个channel就是我们
        //.childHandler(new ChannelInitializer<SocketChannel>() {} 中设置channel
          //准备跟踪
        setChannelOptions(channel, options, logger);
    }
    // 获取ServerBootstrap中的attrs属性
    final Map<AttributeKey<?>, Object> attrs = attrs0();
    synchronized (attrs) {
        // 逐个将attr属性写入到channel
        for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
            @SuppressWarnings("unchecked")
            AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
            channel.attr(key).set(e.getValue());
        }
    }
    // 获取当前channel的pipeline
    ChannelPipeline p = channel.pipeline();
    // 将ServerBootstrap中所有以child开头的属性赋值给局部变量
    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(0));
    }
    synchronized (childAttrs) {
        currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0));
    }
    // 添加一个ChannelInitializer处理器到pipeline
    p.addLast(new ChannelInitializer<Channel>() {
        @Override
        public void initChannel(final Channel ch) throws Exception {
            final ChannelPipeline pipeline = ch.pipeline();
            // 获取serverBootstrap的handler()属性值，并添加到pipeline
            ChannelHandler handler = config.handler();
            if (handler != null) {
                pipeline.addLast(handler);
            }
            ch.eventLoop().execute(new Runnable() {
                @Override
                public void run() {
                    // ServerBootstrapAcceptor称为连接处理器
                    pipeline.addLast(new ServerBootstrapAcceptor(
                            ch, currentChildGroup, currentChildHandler, currentChildOptions,          currentChildAttrs));
                }
            });
        }
    });
}

设置options的属性

static void setChannelOptions(Channel channel, Map<ChannelOption<?>, Object> options, InternalLogger logger) {
    // 遍历options
    for (Map.Entry<ChannelOption<?>, Object> e: options.entrySet()) {
        // 将当前遍历的option初始化到channel
        setChannelOption(channel, e.getKey(), e.getValue(), logger);
    }
}
@SuppressWarnings("unchecked")
private static void setChannelOption(Channel channel, ChannelOption<?> option, Object value, InternalLogger logger) {
    try {
        // 将option写入到channel的config中
        if (!channel.config().setOption((ChannelOption<Object>) option, value)) {
            logger.warn("Unknown channel option '{}' for channel '{}'", option, channel);
        }
    } catch (Throwable t) {
        logger.warn(
                "Failed to set channel option '{}' with value '{}' for channel '{}'", option, value, channel, t);
    }
}
static void setChannelOptions(Channel channel, Map.Entry<ChannelOption<?>, Object>[] options, InternalLogger logger) {
    // 多个时，使用循环的方式增加
    for (Map.Entry<ChannelOption<?>, Object> e: options) {
        setChannelOption(channel, e.getKey(), e.getValue(), logger);
    }
}

属性设置这一块算是跟踪完毕了，继续回到initAndRegister() 方法中跟踪

ChannelFuture regFuture = config().group().register(channel); 跟踪register 的注册细节。

MultithreadEventLoopGroup.java

@Override
public ChannelFuture register(Channel channel) {
    // next()是从group中通过轮询方式选择出一个eventLoop
    return next().register(channel);
}

SingleThreadEventLoop.java

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

AbstractChannel.java

@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
    // 若eventLoop为null，则抛出异常。因为这里要将当前channel与eventLoop进行绑定
    if (eventLoop == null) {
        throw new NullPointerException("eventLoop");
    }
    // 若当前channel已经注册过了，则直接结束
    if (isRegistered()) {
        promise.setFailure(new IllegalStateException("registered to an event loop already"));
        return;
    }
    // 若当前eventLoop与当前channel不兼容，则直接结束
    if (!isCompatible(eventLoop)) {
        promise.setFailure(
                new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
        return;
    }
    // 当前channel与eventLoop的绑定就发生在这里
    AbstractChannel.this.eventLoop = eventLoop;
    if (eventLoop.inEventLoop()) {  // 判断当前正在执行的线程与eventLoop所绑定的线程是否是同一个
        register0(promise);
    } else {
        try {
            // eventLoop本质上是一个executor，这里调用的是它的execute()
            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) {
            try {
                // check if the channel is still open as it could be closed in the mean time when the register
                // call was outside of the eventLoop
                if (!promise.setUncancellable() || !ensureOpen(promise)) {
                    return;
                }
                boolean firstRegistration = neverRegistered;
                // 注册
                doRegister();
                neverRegistered = false;
                registered = true;
                // Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
                // user may already fire events through the pipeline in the ChannelFutureListener.
                pipeline.invokeHandlerAddedIfNeeded();
                safeSetSuccess(promise);
                pipeline.fireChannelRegistered();
                // Only fire a channelActive if the channel has never been registered. This prevents firing
                // multiple channel actives if the channel is deregistered and re-registered.
                if (isActive()) {
                    if (firstRegistration) {
                        pipeline.fireChannelActive();
                    } else if (config().isAutoRead()) {
                        // This channel was registered before and autoRead() is set. This means we need to begin read
                        // again so that we process inbound data.
                        //
                        // See https://github.com/netty/netty/issues/4805
                        beginRead();
                    }
                }
            } catch (Throwable t) {
                // Close the channel directly to avoid FD leak.
                closeForcibly();
                closeFuture.setClosed();
                safeSetFailure(promise, t);
            }
        }

AbstractNioChannel.java

@Override
protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            // 完成NIO原生channel向NIO原生Selector的注册
            // 第二个参数为0，表示当前channel没有关注的事件，为什么指定为0？
            // 两个原因：
            // 1)这是一个一般性方法，所有channel的注册均会调用该方法。所有channel包含三类：
            //    1.1 服务端的parentChannel，其关注的事件应该为OP_ACCEPT，接收连接就绪事件
            //    1.2 服务端的childChannel，其关注的事件应该为OP_READ或OP_WRITE，即读/写就绪事件
            //    1.3 客户端的channel，其关注的事件应该为OP_CONNECT，即连接就绪事件
            // 2)真正指定其所关注的事件，是在Netty封装的channel创建时指定的
            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;
            }
        }
    }  // end-for
}

回到AbstractChannel.java中，跟踪线程创建 eventLoop.execute(new Runnable()

SingleThreadEventExecutor.java

@Override
public void execute(Runnable task) {
    if (task == null) {
        throw new NullPointerException("task");
    }
    // 若当前线程与当前eventLoop所绑定线程是同一个线程，则返回true，否则返回false
    boolean inEventLoop = inEventLoop();
    // 将task任务添加到任务队列
    addTask(task);
    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 && wakesUpForTask(task)) {
        wakeup(inEventLoop);
    }
}
private void startThread() {
        if (state == ST_NOT_STARTED) {
            if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
                try {
                    doStartThread();
                } catch (Throwable cause) {
                    STATE_UPDATER.set(this, ST_NOT_STARTED);
                    PlatformDependent.throwException(cause);
                }
            }
        }
    }
private void doStartThread() {
        assert thread == null;
        // 调用子executor(即eventLoop所包含的executor)的execute()
        // 该execute()完成了两项工作：
        // 1）创建了一个线程
        // 2）启动了这个线程
        executor.execute(new Runnable() {
            @Override
            public void run() {
                thread = Thread.currentThread();
                if (interrupted) {
                    thread.interrupt();
                }
                boolean success = false;
                updateLastExecutionTime();
                try {
                    // 其会调用一个无限循环的for
                    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.
                        for (;;) {
                            if (confirmShutdown()) {
                                break;
                            }
                        }
                    } 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.release();
                            if (!taskQueue.isEmpty()) {
                                if (logger.isWarnEnabled()) {
                                    logger.warn("An event executor terminated with " +
                                            "non-empty task queue (" + taskQueue.size() + ')');
                                }
                            }
                            terminationFuture.setSuccess(null);
                        }
                    }
                }
            }
        });
    }

跟踪doStartThread() 方法中的 线程创建方法 execute

ThreadExecutorMap.java

public static Executor apply(final Executor executor, final EventExecutor eventExecutor) {
    ObjectUtil.checkNotNull(executor, "executor");
    ObjectUtil.checkNotNull(eventExecutor, "eventExecutor");
    // 通过匿名内部类创建的一个executor
    return new Executor() {
        @Override
        public void execute(final Runnable command) {
            // 调用总executor的execute()
            executor.execute(apply(command, eventExecutor));
        }
    };
}

继续跟execute()中的execute方法

ThreadPerTaskExecutor.java

@Override
public void execute(Runnable command) {
    // newThread() 创建一个新的线程
    // start() 启动该新线程，即调用该command的run()方法
    threadFactory.newThread(command).start();
}

创建这个任务线程并返回这个线程。到这里该线程就创建完毕了

DefaultThreadFactory.java

@Override
public Thread newThread(Runnable r) {
    // 创建了一个线程
    Thread t = newThread(FastThreadLocalRunnable.wrap(r), prefix + nextId.incrementAndGet());
    // 初始化线程
    try {
        if (t.isDaemon() != daemon) {
            t.setDaemon(daemon);
        }
        if (t.getPriority() != priority) {
            t.setPriority(priority);
        }
    } catch (Exception ignored) {
        // Doesn't matter even if failed to set.
    }
    return t;
}
protected Thread newThread(Runnable r, String name) {
     return new FastThreadLocalThread(threadGroup, r, name);
}