敖丙肝了一个月的Netty知识点

public class NettyServer {
    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup parentGroup = new NioEventLoopGroup();
        EventLoopGroup childGroup = new NioEventLoopGroup();
        try {

            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(parentGroup, childGroup)
                     .channel(NioServerSocketChannel.class)
                     .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {

                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            pipeline.addLast(new SomeSocketServerHandler());
                         }
                    });

            ChannelFuture future = bootstrap.bind(8888).sync();
            System.out.println("服务器已启动。。。");

            future.channel().closeFuture().sync();
        } finally {
            parentGroup.shutdownGracefully();
            childGroup.shutdownGracefully();
        }
    }
}

public class DemoSocketServerHandler
                       extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {
        System.out.println("Client Address ====== " + ctx.channel().remoteAddress());
        ctx.channel().writeAndFlush("from server:" + UUID.randomUUID());
        ctx.fireChannelActive();
        TimeUnit.MILLISECONDS.sleep(500);
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx,
                                Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

public class NettyClient {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
                            pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
                            pipeline.addLast(new DemoSocketClientHandler());
                        }
                    });

            ChannelFuture future = bootstrap.connect("localhost", 8888).sync();
            future.channel().closeFuture().sync();
        } finally {
            if(eventLoopGroup != null) {
                eventLoopGroup.shutdownGracefully();
            }
        }
    }
}

public class DemoSocketClientHandler
               extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {
        System.out.println(msg);
        ctx.channel().writeAndFlush("from client: " + System.currentTimeMillis());
        TimeUnit.MILLISECONDS.sleep(5000);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx)
            throws Exception {
        ctx.channel().writeAndFlush("from client：begin talking");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx,
                                Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

    public void execute(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }

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

    public NioEventLoopGroup() {
        this(0);
    }

    public NioEventLoopGroup(int nThreads) {
        // 第二个参数是这个group所包含的executor
        this(nThreads, (Executor) null);
    }

    public NioEventLoopGroup(int nThreads, Executor executor) {
        // 第三个参数是provider，其用于提供selector及selectable的channel，
        // 这个provider是当前JVM中唯一的一个单例的provider
        this(nThreads, executor, SelectorProvider.provider());
    }

    public NioEventLoopGroup(
            int nThreads, Executor executor, final SelectorProvider selectorProvider) {
        // 第四个参数是一个选择策略工厂实例
        this(nThreads, executor, selectorProvider, DefaultSelectStrategyFactory.INSTANCE);
    }

    public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,
                             final SelectStrategyFactory selectStrategyFactory) {
        super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());
    }

    protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    }

    protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
        // 第三个参数是选择器工厂实例
        this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
    }

    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是group所包含的executor，其将来会为其所包含的每个eventLoop创建一个线程
            executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
        }

        children = new EventExecutor[nThreads];

        for (int i = 0; i < nThreads; i ++) {
            boolean success = false;
            try {
                // 创建eventLoop
                children[i] = newChild(executor, args);
                success = true;
            } catch (Exception e) {
                throw new IllegalStateException("failed to create a child event loop", e);
            } finally {
                // 在创建这些eventLoop过程中，只要有一个创建失败，则关闭之前所有已经创建好的eventLoop
                if (!success) {
                    // 关闭之前所有已经创建好的eventLoop
                    for (int j = 0; j < i; j ++) {
                        children[j].shutdownGracefully();
                    }

                    // 终止所有eventLoop上所执行的任务
                    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;
                        }
                    }
                }
            }
        }

        // 创建一个选择器
        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);
    }

executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());

children = new EventExecutor[nThreads];

// 逐个创建nioEventLoop实例
for (int i = 0; i < nThreads; i ++) {
    boolean success = false;
    try {
        // 创建eventLoop
        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 {
        // 在创建这些eventLoop过程中，只要有一个创建失败，则关闭之前所有已经创建好的eventLoop
        if (!success) {
            // 闭之前所有已经创建好的eventLoop
            for (int j = 0; j < i; j ++) {
                children[j].shutdownGracefully();
            }

            // 终止所有eventLoop上所执行的任务
            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;
                }
            }
        }
    }
}

@Override
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
    return new NioEventLoop(this, executor, (SelectorProvider) args[0],
            ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
}

NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
             SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {
    super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);
    if (selectorProvider == null) {
        throw new NullPointerException("selectorProvider");
    }
    if (strategy == null) {
        throw new NullPointerException("selectStrategy");
    }
    provider = selectorProvider;
    // 创建一个selector的二元组
    final SelectorTuple selectorTuple = openSelector();
    selector = selectorTuple.selector;
    unwrappedSelector = selectorTuple.unwrappedSelector;
    selectStrategy = strategy;
}

// io.netty.channel.SingleThreadEventLoop#SingleThreadEventLoop
protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,
                                boolean addTaskWakesUp, int maxPendingTasks,
                                RejectedExecutionHandler rejectedExecutionHandler) {
    super(parent, executor, addTaskWakesUp, maxPendingTasks, rejectedExecutionHandler);
    // 创建一个收尾队列
    tailTasks = newTaskQueue(maxPendingTasks);
}

// io.netty.util.concurrent.SingleThreadEventExecutor#SingleThreadEventExecutor
protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
                                    boolean addTaskWakesUp, int maxPendingTasks,
                                    RejectedExecutionHandler rejectedHandler) {
    super(parent);
    this.addTaskWakesUp = addTaskWakesUp;
    this.maxPendingTasks = Math.max(16, maxPendingTasks);
    // 这是当前NioEventLoop所包含的executor
    this.executor = ThreadExecutorMap.apply(executor, this);
    // 创建一个任务队列
    taskQueue = newTaskQueue(this.maxPendingTasks);
    rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
}

this.executor = ThreadExecutorMap.apply(executor, this);

    public static Executor apply(final Executor executor, final EventExecutor eventExecutor) {
        ObjectUtil.checkNotNull(executor, "executor");
        ObjectUtil.checkNotNull(eventExecutor, "eventExecutor");
        // 这里创建的executor是子executor
        return new Executor() {
            // 这个execute()是子executor的execute()
            @Override
            public void execute(final Runnable command) {
                // 这里调用了NioEventLoopGroup所包含的executor的execute()
                // 即调用了“总的executor”的execute()
                executor.execute(apply(command, eventExecutor));
            }
        };
    }

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

private static final class SelectorTuple {
    final Selector unwrappedSelector;  // NIO原生selector
    final Selector selector;  // 优化过的selector

    SelectorTuple(Selector unwrappedSelector) {
        this.unwrappedSelector = unwrappedSelector;
        this.selector = unwrappedSelector;
    }

    SelectorTuple(Selector unwrappedSelector, Selector selector) {
        this.unwrappedSelector = unwrappedSelector;
        this.selector = selector;
    }
}

//2.创建服务端启动引导/辅助类：ServerBootstrap
ServerBootstrap b = new ServerBootstrap();
//3.给引导类配置两大线程组,确定了线程模型
b.group(bossGroup, workerGroup)
    // (非必备)打印日志
    .handler(new LoggingHandler(LogLevel.INFO))
    // 4.指定 IO 模型
    .channel(NioServerSocketChannel.class)
    .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) {
            ChannelPipeline p = ch.pipeline();
            //5.可以自定义客户端消息的业务处理逻辑
            p.addLast(new HelloServerHandler());
        }
    });

Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup)
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
            pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
            pipeline.addLast(new SomeSocketClientHandler());
        }
    });

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

b.bind(port).sync();

// io.netty.bootstrap.AbstractBootstrap#bind(int)
public ChannelFuture bind(int inetPort) {
    return bind(new InetSocketAddress(inetPort));
}

// 继续跟进
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，返回一个异步结果
    final ChannelFuture regFuture = initAndRegister();
    // 从异步结果中获取channel
    final Channel channel = regFuture.channel();
    // 若异步操作执行过程中出现了异常，则直接返回异步对象（直接结束）
    if (regFuture.cause() != null) {
        return regFuture;
    }

    // 处理异步操作完成的情况（可能是正常结束，或发生异常，或任务取消，这些情况都属于有结果的情况）
    if (regFuture.isDone()) {
        ChannelPromise promise = channel.newPromise();
        // 绑定指定的端口
        doBind0(regFuture, channel, localAddress, promise);
        return promise;
    } else {  // 处理异步操作尚未有结果的情况
        final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
        // 为异步操作添加监听
        regFuture.addListener(new ChannelFutureListener() {
            // 若异步操作具有了结果（即完成），则触发该方法的执行
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                Throwable cause = future.cause();
                if (cause != null) { // 异步操作执行过程中出现了问题
                    promise.setFailure(cause);
                } else {  // 异步操作正常结果
                    promise.registered();
                    // 绑定指定的端口
                    doBind0(regFuture, channel, localAddress, promise);
                }
            }
        });
        return promise;
    }
}

final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
        // 创建channel
        channel = channelFactory.newChannel();
        // 初始化channel
        init(channel);
    } catch (Throwable t) {
        if (channel != null) {
            channel.unsafe().closeForcibly();
            return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
        }
        return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
    }

    // 将channel注册到selector
    ChannelFuture regFuture = config().group().register(channel);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }
    return regFuture;
}

@Override
public T newChannel() {
    try {
        // 调用无参构造器创建channel
        return constructor.newInstance();
    } catch (Throwable t) {
        throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
    }
}

public B channel(Class<? extends C> channelClass) {
    if (channelClass == null) {
        throw new NullPointerException("channelClass");
    }
    return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
}

public ReflectiveChannelFactory(Class<? extends T> clazz) {
    ObjectUtil.checkNotNull(clazz, "clazz");
    try {
        // 将NioServerSocketChannel的无参构造器初始化到constructor
        this.constructor = clazz.getConstructor();
    } catch (NoSuchMethodException e) {
        throw new IllegalArgumentException("Class " + StringUtil.simpleClassName(clazz) +
                                           " does not have a public non-arg constructor", e);
    }
}

// NIO中的provider，其用于创建selector与channel。并且是单例的
private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();

public NioServerSocketChannel() {
    // DEFAULT_SELECTOR_PROVIDER 静态变量
    this(newSocket(DEFAULT_SELECTOR_PROVIDER));
}

private static ServerSocketChannel newSocket(SelectorProvider provider) {
    try { 
        // 创建NIO原生的channel => ServerSocketChannel
        return provider.openServerSocketChannel();
    } catch (IOException e) {
        throw new ChannelException(
            "Failed to open a server socket.", e);
    }
}

public NioServerSocketChannel(ServerSocketChannel channel) {
    // 参数1：父channel
    // 参数2：NIO原生channel
    // 参数3：指定当前channel所关注的事件为  接受连接
    super(null, channel, SelectionKey.OP_ACCEPT);
    // 用于对channel进行配置的属性集合
    config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}

// io.netty.channel.nio.AbstractNioMessageChannel#AbstractNioMessageChannel
protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
    super(parent, ch, readInterestOp);
}

// io.netty.channel.nio.AbstractNioChannel#AbstractNioChannel
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
    super(parent);
    // 这里的this.ch为NIO原生channel
    this.ch = ch;
    this.readInterestOp = readInterestOp;
    try {
        // NIO，非阻塞
        ch.configureBlocking(false);
    } catch (IOException e) {
        try {
            ch.close();
        } catch (IOException e2) {
            if (logger.isWarnEnabled()) {
                logger.warn(
                    "Failed to close a partially initialized socket.", e2);
            }
        }
        throw new ChannelException("Failed to enter non-blocking mode.", e);
    }
}

// io.netty.channel.AbstractChannel#AbstractChannel(io.netty.channel.Channel)
protected AbstractChannel(Channel parent) {
    this.parent = parent;
    // 为channel生成id，由五部分构成
    id = newId();
    // 生成一个底层操作对象unsafe
    unsafe = newUnsafe();
    // 创建与这个channel相绑定的channelPipeline
    pipeline = newChannelPipeline();
}

void init(Channel channel) throws Exception {
    // 获取serverBootstrap中的options属性
    final Map<ChannelOption<?>, Object> options = options0();
    // 将options属性设置到channel
    synchronized (options) {
        setChannelOptions(channel, options, logger);
    }

    // 获取serverBootstrap中的attrs属性
    final Map<AttributeKey<?>, Object> attrs = attrs0();
    synchronized (attrs) {
        // 遍历attrs属性
        for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
            @SuppressWarnings("unchecked")
            AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
            // 将当前遍历的attr初始化到channel
            channel.attr(key).set(e.getValue());
        }
    }

    // 获取channel的pipeline
    ChannelPipeline p = channel.pipeline();

    // 将serverBootstrap中所有以child开头的属性写入到局部变量，
    // 然后将它们初始化到childChannel中
    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));
    }

    p.addLast(new ChannelInitializer<Channel>() {
        @Override
        public void initChannel(final Channel ch) throws Exception {
            final ChannelPipeline pipeline = ch.pipeline();
            ChannelHandler handler = config.handler();
            if (handler != null) {
                pipeline.addLast(handler);
            }

            ch.eventLoop().execute(new Runnable() {
                @Override
                public void run() {
                    // 将ServerBootstrapAcceptor处理器添加到pipeline
                    // ServerBootstrapAcceptor处理器用于接收ServerBootstrap中的属性值，
                    // 我们通常称其为连接处理器
                    pipeline.addLast(new ServerBootstrapAcceptor(
                        ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                }
            });
        }
    });
}

p.addLast(new ChannelInitializer<Channel>() {
        @Override
        public void initChannel(final Channel ch) throws Exception {
            final ChannelPipeline pipeline = ch.pipeline();
            ChannelHandler handler = config.handler();
            if (handler != null) {
                pipeline.addLast(handler);
            }

            ch.eventLoop().execute(new Runnable() {
                @Override
                public void run() {
                    // 将ServerBootstrapAcceptor处理器添加到pipeline
                    // ServerBootstrapAcceptor处理器用于接收ServerBootstrap中的属性值，
                    // 我们通常称其为连接处理器
                    pipeline.addLast(new ServerBootstrapAcceptor(
                        ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                }
            });
        }
    });

ServerBootstrapAcceptor(
    final Channel channel, EventLoopGroup childGroup, ChannelHandler childHandler,
    Entry<ChannelOption<?>, Object>[] childOptions, Entry<AttributeKey<?>, Object>[] childAttrs) {
    this.childGroup = childGroup;
    this.childHandler = childHandler;
    this.childOptions = childOptions;
    this.childAttrs = childAttrs;

    // See https://github.com/netty/netty/issues/1328
    enableAutoReadTask = new Runnable() {
        @Override
        public void run() {
            channel.config().setAutoRead(true);
        }
    };
}

public void channelRead(ChannelHandlerContext ctx, Object msg) {
    // msg为客户端发送来的数据，其为NioSocketChannel，即子channel，childChannel
    final Channel child = (Channel) msg;

    // 将来自于ServerBootstrap的child开头属性初始化到childChannel中（childHandler、childOptions、childAttrs）
    child.pipeline().addLast(childHandler);
    setChannelOptions(child, childOptions, logger);
    for (Entry<AttributeKey<?>, Object> e: childAttrs) {
        child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
    }

    try {
        // 将childChannel注册到selector 需要注意的是，这里的selector与父channel所注册的selector不是同一个
        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);
    }
}

// 
config().group().register(channel);

public ChannelFuture register(Channel channel) {
    // next() 从eventLoop数组中选择一个eventLoop
    return next().register(channel);
}

public ChannelFuture register(Channel channel) {
    // 创建一个 ChannelPromise 然后注册
    return register(new DefaultChannelPromise(channel, this));
}

// ----> 这里继续调用 unsafe 的 register
public ChannelFuture register(final ChannelPromise promise) {
    ObjectUtil.checkNotNull(promise, "promise");
    promise.channel().unsafe().register(this, promise);
    return promise;
}

@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
    // 对异常情况的处理
    if (eventLoop == null) {
        throw new NullPointerException("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;
    }

    // channel与eventLoop的绑定就发生在这里，
    // 需要注意，这里的eventLoop还没有绑定线程，因为这个线程还没有创建
    AbstractChannel.this.eventLoop = eventLoop;
    // 判断当前线程与eventLoop所绑定线程是否是同一个线程
    if (eventLoop.inEventLoop()) {
        register0(promise);
    } else {
        try {
            // 执行当前线程所绑定的eventLoop的execute(), 这个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 {
        if (!promise.setUncancellable() || !ensureOpen(promise)) {
            return;
        }
        boolean firstRegistration = neverRegistered;
        doRegister();  // 绑定
        neverRegistered = false;
        registered = true;
        
        pipeline.invokeHandlerAddedIfNeeded();

        safeSetSuccess(promise);
        pipeline.fireChannelRegistered();
        
        if (isActive()) {
            if (firstRegistration) {
                pipeline.fireChannelActive();
            } else if (config().isAutoRead()) {
                beginRead();
            }
        }
    } catch (Throwable t) {
        closeForcibly();
        closeFuture.setClosed();
        safeSetFailure(promise, t);
    }
}

protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            // 在这里进行了注册，将NIO原生channel注册到了NIO原生selector
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
            return;
        } catch (CancelledKeyException e) {
            if (!selected) {
                eventLoop().selectNow();
                selected = true;
            } else {
                throw e;
            }
        }
    }
}

// 刚刚往里面跟的是 register0 现在再说一下 execute 
eventLoop.execute(new Runnable() {
    @Override
    public void run() {
        register0(promise);
    }
});

public void execute(Runnable task) {
    if (task == null) {
        throw new NullPointerException("task");
    }

    // 判断当前线程与eventLoop所绑定线程是否是同一个
    boolean inEventLoop = inEventLoop();
    // 将任务添加到任务队列
    addTask(task);
    if (!inEventLoop) {
        // 创建并启动一个线程
        startThread();
        if (isShutdown()) {
            boolean reject = false;
            try {
                if (removeTask(task)) {
                    reject = true;
                }
            } catch (UnsupportedOperationException e) {
            }
            if (reject) {
                reject();
            }
        }
    }

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

private void startThread() {
    // 若当前eventLoop所绑定线程尚未启动
    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;
    // 调用NioEventLoop所包含的executor的execute()
    // 这个execute()会创建并启动一个线程
    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 {
                // 省略......
        }
    });
}

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) {
            // 这里调用了NioEventLoopGroup所包含的executor的execute()
            executor.execute(apply(command, eventExecutor));
        }
    };
}

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

public void execute(Runnable command) {
    // 创建并启动一个线程
    threadFactory.newThread(command).start();
}

//  对应的代码位置 io.netty.channel.AbstractChannel.AbstractUnsafe#register
eventLoop.execute(new Runnable() {
     @Override
     public void run() {
         register0(promise);
     }
 });

 private void doStartThread() {
        assert thread == null;
        // 调用NioEventLoop所包含的executor的execute() 这个execute()会创建并启动一个线程
        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 {
               // 省略。。。

protected void run() {
    for (;;) {
        try {
            try {
                // 选择就绪的channel
                switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                    case SelectStrategy.CONTINUE:    // NioEventLoop不支持
                        continue;
                    case SelectStrategy.BUSY_WAIT:  // NioEventLoop不支持

                    case SelectStrategy.SELECT:  // SELECT = -1 能走到这里，说明当前任务队列中没有任务
                        // 进行阻塞式选择
                        select(wakenUp.getAndSet(false));
                        if (wakenUp.get()) {
                            selector.wakeup();
                        }
                    default:
                }
            } catch (IOException e) {
                rebuildSelector0();
                handleLoopException(e);
                continue;
            }

            cancelledKeys = 0;
            needsToSelectAgain = false;
            // 该变量用于设置“处理就绪channel的IO所使用的时间”与“处理任务队列中任务使用时间”的比例 该值为整型，不大于100
            final int ioRatio = this.ioRatio;
            if (ioRatio == 100) {
                try {
                    processSelectedKeys();
                } finally {
                    runAllTasks();
                }
            } else {
                // 记录处理就绪channel的IO开始执行的时间点
                final long ioStartTime = System.nanoTime();
                try {
                    // 处理就绪channel的IO
                    processSelectedKeys();
                } finally {
                    // 计算出处理就绪channel的IO所使用的时长
                    final long ioTime = System.nanoTime() - ioStartTime;
                    // 执行任务队列中的任务
                    runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                }
            }
            // 省略。。。
    }
}

selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());

// hasTasks  tailTasks 收尾任务队列
protected boolean hasTasks() {
    return super.hasTasks() || !tailTasks.isEmpty();
}

// super.hasTasks()   taskQueue 普通任务队列
protected boolean hasTasks() {
    assert inEventLoop();
    return !taskQueue.isEmpty();
}

// SelectStrategy.SELECT = -1
public int calculateStrategy(IntSupplier selectSupplier, boolean hasTasks) throws Exception {
    return hasTasks ? selectSupplier.get() : SelectStrategy.SELECT;
}

public int get() throws Exception {
    return selectNow();
}

// io.netty.channel.nio.NioEventLoop#selectNow
int selectNow() throws IOException {
    try {
        return selector.selectNow();
    } finally {
        // restore wakeup state if needed
        if (wakenUp.get()) {
            selector.wakeup();
        }
    }
}

switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
    case SelectStrategy.CONTINUE:    // -2 NioEventLoop不支持 
        continue;

    case SelectStrategy.BUSY_WAIT:  // -3 NioEventLoop不支持
   
    case SelectStrategy.SELECT:  // -1 能走到这里，说明当前任务队列中没有任务
        // 进行阻塞式选择
        select(wakenUp.getAndSet(false));
        if (wakenUp.get()) {
            selector.wakeup();
        }
    default:
}

private void select(boolean oldWakenUp) throws IOException {
    Selector selector = this.selector;
    try {
        // 计数器：用于记录空轮询导致CPU占用率飙升，select()提前结束的次数（其值大于1时）
        int selectCnt = 0;
        // 获取当前时间，也就是for循环第一次开始执行的时间点
        long currentTimeNanos = System.nanoTime();
        // delayNanos() 表示定时任务队列中第一个定时任务还有多久就到开始执行的时间了
        long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);

        for (;;) {
            // 处理小于0.5毫秒的任务
            long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
            if (timeoutMillis <= 0) {  // 该条件为true，表示具有立即需要执行的定时任务
                if (selectCnt == 0) {  // 只有第一次for循环才会执行下面的“非阻塞选择”
                    selector.selectNow();
                    selectCnt = 1;
                }
                break;
            }

            if (hasTasks() && wakenUp.compareAndSet(false, true)) {
                selector.selectNow();
                selectCnt = 1;
                break;
            }
            
            int selectedKeys = selector.select(timeoutMillis);
            selectCnt ++;
            // 若有就绪的channel了，则直接结束
            if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
                break;
            }
            // 若当前线程被中断
            if (Thread.interrupted()) {
                selectCnt = 1;
                break;
            }

            // 获取当前时间
            long time = System.nanoTime();
            // 下面的式子等价于：  time - currentTimeNanos >= timeoutMillis
            // 若下面的条件成立，则说明select()是在指定的阻塞时间过期后才跳出的，即正常结束的
            if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
                // timeoutMillis elapsed without anything selected.
                selectCnt = 1;
            } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
                       selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
                selector = selectRebuildSelector(selectCnt);  // 重构selector
                selectCnt = 1;
                break;
            }
            // 本轮for循环结束时间点，同时也是下一轮for循环的开始时间点
            currentTimeNanos = time;
        }  
    } catch (CancelledKeyException e) {
    }
}

final int ioRatio = this.ioRatio; // 默认值 50
if (ioRatio == 100) {
    try {
        processSelectedKeys();
    } finally {
        // Ensure we always run tasks.
        runAllTasks();
    }
} else {
    // 记录处理就绪channel的IO开始执行的时间点
    final long ioStartTime = System.nanoTime();
    try {
        // 处理就绪channel的IO
        processSelectedKeys();
    } finally {
        // 计算出处理就绪channel的IO所使用的时长
        final long ioTime = System.nanoTime() - ioStartTime;
        // 执行任务队列中的任务
        runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
    }
}

private void processSelectedKeys() {
    // 判断channel的selectedKeys是否是优化过的
    if (selectedKeys != null) {
        processSelectedKeysOptimized();  // 优化处理方式
    } else {
        processSelectedKeysPlain(selector.selectedKeys());  // 普通处理方式
    }
}

private void processSelectedKeysOptimized() {
    for (int i = 0; i < selectedKeys.size; ++i) {
        // 从数组中取出一个元素
        final SelectionKey k = selectedKeys.keys[i];
     // 移除已经取出的 SelectionKey，使 GC 可以处理到已经关闭的 channel
        selectedKeys.keys[i] = null;
        // 获取selectionKey的附件，该附件中可以存放任意数据，不过这里存放的是NIO原生channel
        final Object a = k.attachment();

        if (a instanceof AbstractNioChannel) {
            processSelectedKey(k, (AbstractNioChannel) a);  // 处理就绪事件
        } else {
            NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
            processSelectedKey(k, task); // 这里是测试代码。跟进去可以看到实现方法是测试类
        }
// 省略......

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
    // 处理selectionKey失效的情况
    if (!k.isValid()) {
        final EventLoop eventLoop;
        try {
            eventLoop = ch.eventLoop();
        } catch (Throwable ignored) {
            return;
        }
        if (eventLoop != this || eventLoop == null) {
            return;
        }
        unsafe.close(unsafe.voidPromise());
        return;
    }

    try {
        int readyOps = k.readyOps();
  // 判断当前 channnel 就绪的事件类型
        if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
            // 获取当前selectionKey的interestOps
            int ops = k.interestOps();
            // 先将SelectionKey.OP_CONNECT按位取或，再与ops进行按位与
            ops &= ~SelectionKey.OP_CONNECT;
            // 将修改过的ops再写入到selectionsKey中
            k.interestOps(ops);
            // 连接server
            unsafe.finishConnect();
        }

        // 处理写就绪的情况
        if ((readyOps & SelectionKey.OP_WRITE) != 0) {
            // 强制刷新（将user buffer中的数据写入到网关缓存）
            ch.unsafe().forceFlush();
        }
        // readyOps为0表示当前没有任何channel就绪
        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
            // 将网卡缓存中的数据写入到user buffer
            unsafe.read();
        }
    } catch (CancelledKeyException ignored) {
        unsafe.close(unsafe.voidPromise());
    }
}

runAllTasks(ioTime * (100 - ioRatio) / ioRatio);

protected boolean runAllTasks(long timeoutNanos) {
    // 从定时任务队列中取出所有当前马上就要到期的定时任务放入到任务队列
    fetchFromScheduledTaskQueue();
    // 从任务队列中取出一个任务
    Runnable task = pollTask();
    // 若该任务为空，则说明任务队列中已经没有任务了，此时就可以执行收尾任务了
    if (task == null) {
        // 执行收尾队列中的收尾任务
        afterRunningAllTasks();
        return false;
    }

    final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
    // 计数器
    long runTasks = 0;
    long lastExecutionTime;
    for (;;) {
        // 执行任务
        safeExecute(task);
        runTasks ++;
        // 每64个任务查看一次超时
        if ((runTasks & 0x3F) == 0) {
            lastExecutionTime = ScheduledFutureTask.nanoTime();
            if (lastExecutionTime >= deadline) {
                break;
            }
        }

        // 从任务队列中再取出一个任务
        task = pollTask();
        if (task == null) {
            lastExecutionTime = ScheduledFutureTask.nanoTime();
            break;
        }
    } // end-for

    // 处理收尾队列中的任务
    afterRunningAllTasks();
    this.lastExecutionTime = lastExecutionTime;
    return true;
}

private static void doBind0(
    final ChannelFuture regFuture, final Channel channel,
    final SocketAddress localAddress, final ChannelPromise promise) {
    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            if (regFuture.isSuccess()) {  // 只有当channel初始化注册成功后，才会进行绑定
                channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                promise.setFailure(regFuture.cause());
            }
        }
    });
}

public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
    return pipeline.bind(localAddress, promise);
}

public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
    return tail.bind(localAddress, promise);
}

public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
    。。。。
    final AbstractChannelHandlerContext next = findContextOutbound(MASK_BIND);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeBind(localAddress, promise);
    } else {
        safeExecute(executor, new Runnable() {
            @Override
            public void run() {
                next.invokeBind(localAddress, promise);
            }
        }, promise, null);
    }
    return promise;
}

private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {
    if (invokeHandler()) {
        try {
            ((ChannelOutboundHandler) handler()).bind(this, localAddress, promise);
        } catch (Throwable t) {
            notifyOutboundHandlerException(t, promise);
        }
    } else {
        bind(localAddress, promise);
    }
}

public void bind(
    ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
    unsafe.bind(localAddress, promise);
}

public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
    // 省略。。。

    // 获取当前channel是否被激活。注意，现在还没有被激活，所以其值为false
    boolean wasActive = isActive();
    try {
        // 绑定
        doBind(localAddress);
    } catch (Throwable t) {
        safeSetFailure(promise, t);
        closeIfClosed();
        return;
    }

    if (!wasActive && isActive()) {
        invokeLater(new Runnable() {
            @Override
            public void run() {
                pipeline.fireChannelActive();  // 触发重写的channelActivate方法的执行
            }
        });
    }
    safeSetSuccess(promise);
}

protected void doBind(SocketAddress localAddress) throws Exception {
    if (PlatformDependent.javaVersion() >= 7) {
        javaChannel().bind(localAddress, config.getBacklog());
    } else {
        javaChannel().socket().bind(localAddress, config.getBacklog());
    }
}

NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
try {
    Bootstrap bootstrap = new Bootstrap();
    bootstrap.group(eventLoopGroup)
        .channel(NioSocketChannel.class)
        .handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
                pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
                pipeline.addLast(new SomeSocketClientHandler());
            }
        });

    ChannelFuture future = bootstrap.connect("localhost", 8888).sync();
    future.channel().closeFuture().sync();
} finally {
    if(eventLoopGroup != null) {
        eventLoopGroup.shutdownGracefully();
    }
}

public ChannelFuture connect(String inetHost, int inetPort) {
    return connect(InetSocketAddress.createUnresolved(inetHost, inetPort));
}

public ChannelFuture connect(SocketAddress remoteAddress) {
    if (remoteAddress == null) {
        throw new NullPointerException("remoteAddress");
    }
    // 验证bootstrap的group、channelFactory与handler是否为空
    validate();
    // 解析并连接地址
    return doResolveAndConnect(remoteAddress, config.localAddress());
}

private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
    // 创建、初始化并注册channel
    final ChannelFuture regFuture = initAndRegister();
    final Channel channel = regFuture.channel();
    // 若channel注册完毕
    if (regFuture.isDone()) {
        if (!regFuture.isSuccess()) {
            return regFuture;
        }
        // 解析并连接server端地址
        return doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise());
    } else {
        final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
        regFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                Throwable cause = future.cause();
                if (cause != null) {
                    promise.setFailure(cause);
                } else {
                    promise.registered();
                    doResolveAndConnect0(channel, remoteAddress, localAddress, promise);
                }
            }
        });
        return promise;
    }
}

final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
        // 创建channel
        channel = channelFactory.newChannel();
        // 初始化channel
        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);
    }

    // 将channel注册到selector
    ChannelFuture regFuture = config().group().register(channel);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }
        return regFuture;
    }

public NioSocketChannel() {
    // DEFAULT_SELECTOR_PROVIDER：全局唯一的provider，通过它可以创建出selector与channel
    this(DEFAULT_SELECTOR_PROVIDER);
}

// 继续跟进 this
public NioSocketChannel(SelectorProvider provider) {
    this(newSocket(provider));
}

public NioSocketChannel(Channel parent, SocketChannel socket) {
    super(parent, socket);
    config = new NioSocketChannelConfig(this, socket.socket());
}

protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
    // 第三个参数： 指定关注事件为读事件 OP_READ
    super(parent, ch, SelectionKey.OP_READ);
}

void init(Channel channel) throws Exception {
    ChannelPipeline p = channel.pipeline();
    // 将bootstrap中创建的ChannelInitializer处理器添加到pipeline
    p.addLast(config.handler());

    // 将bootstrap中的options初始化到channel
    final Map<ChannelOption<?>, Object> options = options0();
    synchronized (options) {
        setChannelOptions(channel, options, logger);
    }

    // 将bootstrap中的attrs初始化到channel
    final Map<AttributeKey<?>, Object> attrs = attrs0();
    synchronized (attrs) {
        for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
            channel.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
        }
    }
}

private ChannelFuture doResolveAndConnect0(final Channel channel, SocketAddress remoteAddress,
                                           final SocketAddress localAddress, final ChannelPromise promise) {
    try {
        final EventLoop eventLoop = channel.eventLoop();
        // 创建一个地址解析器，其中包含一个地址格式匹配器
        final AddressResolver<SocketAddress> resolver = this.resolver.getResolver(eventLoop);

        // 若解析器不支持该地址 或 该地址已经解析过了，则直接对该地址进行连接，
        // 返回可修改的promise，即成功了就成功，失败了则promise中有失败信息
        if (!resolver.isSupported(remoteAddress) || resolver.isResolved(remoteAddress)) {
            doConnect(remoteAddress, localAddress, promise);
            return promise;
        }
        // 以异步方式解析server地址
        final Future<SocketAddress> resolveFuture = resolver.resolve(remoteAddress);

        if (resolveFuture.isDone()) {  // 处理解析完成的情况（成功或异常）
            final Throwable resolveFailureCause = resolveFuture.cause();

            if (resolveFailureCause != null) {  // 若异步解析中出现了问题，则直接关闭channel
                channel.close();
                promise.setFailure(resolveFailureCause);
            } else {  // 处理异步解析成功的情况
                // resolveFuture.getNow() 从异步对象中获取解析结果，即解析过的地址
                doConnect(resolveFuture.getNow(), localAddress, promise);
            }
            return promise;
        }

        resolveFuture.addListener(new FutureListener<SocketAddress>() {
            @Override
            public void operationComplete(Future<SocketAddress> future) throws Exception {
                if (future.cause() != null) {
                    channel.close();
                    promise.setFailure(future.cause());
                } else {
                    doConnect(future.getNow(), localAddress, promise);
                }
            }
        });
    } catch (Throwable cause) {
        promise.tryFailure(cause);
    }
    return promise;
}

private static void doConnect(
    final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise connectPromise) {

    final Channel channel = connectPromise.channel();
    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            if (localAddress == null) {
                channel.connect(remoteAddress, connectPromise);
            } else {
                channel.connect(remoteAddress, localAddress, connectPromise);
            }
            // 为promise添加一个异常监听器。连接过程发生异常，则关闭channel
            connectPromise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
        }
    });
}

public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
    return pipeline.connect(remoteAddress, localAddress, promise);
}

public final ChannelFuture connect(
    SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
    return tail.connect(remoteAddress, localAddress, promise);
}

public ChannelFuture connect(
    final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {

    if (remoteAddress == null) {
        throw new NullPointerException("remoteAddress");
    }
    if (isNotValidPromise(promise, false)) {
        // cancelled
        return promise;
    }

    // 查找要处理该请求的处理器节点
    final AbstractChannelHandlerContext next = findContextOutbound(MASK_CONNECT);
    // 获取处理器节点的executor
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeConnect(remoteAddress, localAddress, promise);
    } else {
        safeExecute(executor, new Runnable() {
            @Override
            public void run() {
                next.invokeConnect(remoteAddress, localAddress, promise);
            }
        }, promise, null);
    }
    return promise;
}

private void invokeConnect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
    if (invokeHandler()) {  // 判断该处理器节点中对应的处理器是否已经添加
        try {
            ((ChannelOutboundHandler) handler()).connect(this, remoteAddress, localAddress, promise);
        } catch (Throwable t) {
            notifyOutboundHandlerException(t, promise);
        }
    } else {
        connect(remoteAddress, localAddress, promise);
    }
}

public void connect(
    ChannelHandlerContext ctx,
    SocketAddress remoteAddress, SocketAddress localAddress,
    ChannelPromise promise) {
    // 连接
    unsafe.connect(remoteAddress, localAddress, promise);
}

public final void connect(
    final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
    if (!promise.setUncancellable() || !ensureOpen(promise)) {
        return;
    }
    try {
        if (connectPromise != null) {
            // Already a connect in process.
            throw new ConnectionPendingException();
        }

        boolean wasActive = isActive();
        if (doConnect(remoteAddress, localAddress)) {  // 连接            
            // 省略+......

protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
    if (localAddress != null) {
        doBind0(localAddress);  // 将localAddress绑定到channel
    }

    boolean success = false;
    try {
        // 连接server地址，若本次连接成功，则成功；若不成功，则当前channel的连接就绪
        boolean connected = SocketUtils.connect(javaChannel(), remoteAddress);
        if (!connected) {
            // 指定其关注的事件为  连接就绪
            selectionKey().interestOps(SelectionKey.OP_CONNECT);
        }
        success = true;
        return connected;
    } finally {
        if (!success) {
            doClose();
        }
    }
}

protected AbstractChannel(Channel parent) {
    this.parent = parent;
    // 为channel生成id，由五部分构成
    id = newId();
    // 生成一个底层操作对象unsafe
    unsafe = newUnsafe();
    // 创建与这个channel相绑定的channelPipeline
    pipeline = newChannelPipeline();
}

protected DefaultChannelPipeline newChannelPipeline() {
    return new DefaultChannelPipeline(this);
}

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

final boolean setAddComplete() {
    for (;;) {
        int oldState = handlerState; // 获取处理器状态
        if (oldState == REMOVE_COMPLETE) { // 处理器状态为移除状态
            return false;
        }
        // 通过CAS方式将处理器状态修改为 添加完毕
        if (HANDLER_STATE_UPDATER.compareAndSet(this, oldState, ADD_COMPLETE)) {
            return true;
        }
    }
}

AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor,
                              String name, Class<? extends ChannelHandler> handlerClass) {
    this.name = ObjectUtil.checkNotNull(name, "name");
    this.pipeline = pipeline;
    // 每个处理器节点都会绑定一个executor
    this.executor = executor;
    // 执行标记
    this.executionMask = mask(handlerClass);
    ordered = executor == null || executor instanceof OrderedEventExecutor;
}

static int mask(Class<? extends ChannelHandler> clazz) {
    // 从缓存中尝试着获取标记
    Map<Class<? extends ChannelHandler>, Integer> cache = MASKS.get();
    Integer mask = cache.get(clazz);
    if (mask == null) {
        // 创建一个标记
        mask = mask0(clazz);
        cache.put(clazz, mask);
    }
    return mask;
}

// 头节点既是inbound处理器，也是outbound处理器
final class HeadContext extends AbstractChannelHandlerContext
            implements ChannelOutboundHandler, ChannelInboundHandler {

        private final Unsafe unsafe;

        HeadContext(DefaultChannelPipeline pipeline) {
            super(pipeline, null, HEAD_NAME, HeadContext.class);
            unsafe = pipeline.channel().unsafe();
            setAddComplete();
        }
        // 。。。。

protected abstract void initChannel(C ch) throws Exception;

p.addLast(new ChannelInitializer<Channel>() {
    @Override
    public void initChannel(final Channel ch) throws Exception {
        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));
            }
        });
    }
});

// 当前处理器所封装的处理器节点被添加到pipeline后就会触发该方法的执行
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    if (ctx.channel().isRegistered()) {  // 若channel已经完成了注册
        if (initChannel(ctx)) {
            // 将当前处理器节点从initMap集合中删除
            removeState(ctx);
        }
    }
}

private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
    // 将当前处理器节点添加到initMap集合中
    if (initMap.add(ctx)) { // Guard against re-entrance.
        try {
            // 调用重写的initChannel()
            initChannel((C) ctx.channel());
        } catch (Throwable cause) {
            exceptionCaught(ctx, cause);
        } finally {
            // 立即将该处理器节点从pipeline上删除
            ChannelPipeline pipeline = ctx.pipeline();
            // 查找在pipeline中是否存在当前处理器
            if (pipeline.context(this) != null) {
                // 将当前处理器节点从pipeline中删除
                pipeline.remove(this);
            }
        }
        return true;
    }
    return false;
}

public final ChannelPipeline addLast(ChannelHandler... handlers) {
    // 第一个参数为group，其值默认为null
    return addLast(null, handlers);
}

@Override
public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
    if (handlers == null) {
        throw new NullPointerException("handlers");
    }
    // 遍历所有handlers，逐个添加到pipeline
    for (ChannelHandler h: handlers) {
        if (h == null) {
            break;
        }
        addLast(executor, null, h);// 这里第二个参数是处理器name
    }
    return this;
}

public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    final AbstractChannelHandlerContext newCtx;
    synchronized (this) {
        // 检测处理器是否被多次添加
        checkMultiplicity(handler);

        // 将处理器包装为一个节点 filterName() 获取到节点的名称
        newCtx = newContext(group, filterName(name, handler), handler);

        // 将新的节点添加到pipeline
        addLast0(newCtx);
        
        if (!registered) {
            newCtx.setAddPending();
            callHandlerCallbackLater(newCtx, true);
            return this;
        }

        // 获取新建节点绑定的eventLoop
        EventExecutor executor = newCtx.executor();
        // 若该eventLoop绑定的线程与当前线程不是同一个，则执行下面的代码
        if (!executor.inEventLoop()) {
            callHandlerAddedInEventLoop(newCtx, executor);
            return this;
        }
    }
    // 若该eventLoop绑定的线程与当前线程是同一个线程，
    // 则调用重写的handlerAdded()方法
    callHandlerAdded0(newCtx);
    return this;
}

private AbstractChannelHandlerContext newContext(EventExecutorGroup group, String name, ChannelHandler handler) {
    // childExecutor() 获取与当前处理器节点相绑定的eventLoop
    return new DefaultChannelHandlerContext(this, childExecutor(group), name, handler);
}

private EventExecutor childExecutor(EventExecutorGroup group) {
    // 若group为null，则与该节点绑定的eventLoop为null
    if (group == null) {
        return null;
    }
    
    Boolean pinEventExecutor = channel.config().getOption(ChannelOption.SINGLE_EVENTEXECUTOR_PER_GROUP);
    if (pinEventExecutor != null && !pinEventExecutor) {
        return group.next();
    }

    Map<EventExecutorGroup, EventExecutor> childExecutors = this.childExecutors;
    if (childExecutors == null) {
        childExecutors = this.childExecutors = new IdentityHashMap<EventExecutorGroup, EventExecutor>(4);
    }
    EventExecutor childExecutor = childExecutors.get(group);
    if (childExecutor == null) {
        childExecutor = group.next();
        childExecutors.put(group, childExecutor);
    }
    return childExecutor;
}

public EventExecutor executor() {
    if (executor == null) {
        return channel().eventLoop();  // 获取到channel所绑定的eventLoop
    } else {
        return executor;
    }
}

private EventExecutor childExecutor(EventExecutorGroup group) {
    if (group == null) {
        return null;
    }

    Boolean pinEventExecutor = channel.config().getOption(ChannelOption.SINGLE_EVENTEXECUTOR_PER_GROUP);
    if (pinEventExecutor != null && !pinEventExecutor) {
        // 轮询获取 eventLoop
        return group.next();
    }

    Map<EventExecutorGroup, EventExecutor> childExecutors = this.childExecutors;
    if (childExecutors == null) {
        childExecutors = this.childExecutors = new IdentityHashMap<EventExecutorGroup, EventExecutor>(4);
    }

    EventExecutor childExecutor = childExecutors.get(group);
    if (childExecutor == null) {
        childExecutor = group.next();
        childExecutors.put(group, childExecutor);
    }
    return childExecutor;
}

private void addLast0(AbstractChannelHandlerContext newCtx) {
    AbstractChannelHandlerContext prev = tail.prev;
    newCtx.prev = prev;
    newCtx.next = tail;
    prev.next = newCtx;
    tail.prev = newCtx;
}

private void callHandlerAddedInEventLoop(final AbstractChannelHandlerContext newCtx, EventExecutor executor) {
    newCtx.setAddPending(); // 将当前节点状态设置成处理中，等待 callHandlerAdded0 执行完成
    executor.execute(new Runnable() {
        @Override
        public void run() {
            callHandlerAdded0(newCtx);
        }
    });
}

private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
    try {
        ctx.callHandlerAdded();
    } catch (Throwable t) {
        boolean removed = false;
        try {
            remove0(ctx);
            ctx.callHandlerRemoved();
            removed = true;
        } catch (Throwable t2) {
            if (logger.isWarnEnabled()) {
                logger.warn("Failed to remove a handler: " + ctx.name(), t2);
            }
        }

        if (removed) {
            fireExceptionCaught(new ChannelPipelineException(
                ctx.handler().getClass().getName() +
                ".handlerAdded() has thrown an exception; removed.", t));
        } else {
            fireExceptionCaught(new ChannelPipelineException(
                ctx.handler().getClass().getName() +
                ".handlerAdded() has thrown an exception; also failed to remove.", t));
        }
    }
}

final void callHandlerAdded() throws Exception {
    if (setAddComplete()) { // CAS 方式将处理器状态修改为 添加完毕 ADD_COMPLETE
        handler().handlerAdded(this);
    }
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
    throws Exception {
    System.out.println("InboundHandler1 进入 " + msg);
    ctx.fireChannelRead(msg);
    System.out.println("InboundHandler1 退出 " + msg);
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
    throws Exception {
    System.out.println("InboundHandler2 进入 " + msg);
    ctx.fireChannelRead(msg);
    System.out.println("InboundHandler2 退出 " + msg);
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
    throws Exception {
    System.out.println("InboundHandler3 进入 " + msg);
    ctx.fireChannelRead(msg);
    System.out.println("InboundHandler3 退出 " + msg);
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
        throws Exception {
    System.out.println("OutboundHandler1 进入 " + msg);
    ctx.write(msg, promise);
    System.out.println("OutboundHandler1 退出 " + msg);
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
        throws Exception {
    System.out.println("OutboundHandler2 进入 " + msg);
    ctx.write(msg, promise);
    System.out.println("OutboundHandler2 退出 " + msg);
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
        throws Exception {
    System.out.println("OutboundHandler3 进入 " + msg);
    ctx.write(msg, promise);
    System.out.println("OutboundHandler3 退出 " + msg);
}

public void channelRead(ChannelHandlerContext ctx, Object msg) {
    ctx.fireChannelRead(msg);  // 准备调用其下一个节点的channelRead()
}

public ChannelHandlerContext fireChannelRead(final Object msg) {
    invokeChannelRead(findContextInbound(MASK_CHANNEL_READ), msg);
    return this;
}

private AbstractChannelHandlerContext findContextInbound(int mask) {
    AbstractChannelHandlerContext ctx = this;
    do {
        ctx = ctx.next;
    } while ((ctx.executionMask & mask) == 0);
    return ctx;
}

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
    final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRead(m);
    } else {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                next.invokeChannelRead(m);
            }
        });
    }
}

private void invokeChannelRead(Object msg) {
    if (invokeHandler()) {
        try {
            ((ChannelInboundHandler) handler()).channelRead(this, msg);
        } catch (Throwable t) {
            notifyHandlerException(t);
        }
    } else {
        fireChannelRead(msg);
    }
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
    throws Exception {
    System.out.println("WriteInboundHandler  进入 " + msg);
    ctx.channel().write("我是 Server。");
    System.out.println("InboundHWriteInboundHandler andler3 退出 " + msg);
}

@Override
public final ChannelFuture write(Object msg) {
    return tail.write(msg);
}

@Override
public ChannelFuture write(Object msg) {
    return write(msg, newPromise());
}

private void write(Object msg, boolean flush, ChannelPromise promise) {
        ObjectUtil.checkNotNull(msg, "msg");
        try {
            if (isNotValidPromise(promise, true)) {
                ReferenceCountUtil.release(msg);
                // cancelled
                return;
            }
        } catch (RuntimeException e) {
            ReferenceCountUtil.release(msg);
            throw e;
        }
        // 查找下一个节点(prev)
        final AbstractChannelHandlerContext next = findContextOutbound(flush ?
                (MASK_WRITE | MASK_FLUSH) : MASK_WRITE);
        final Object m = pipeline.touch(msg, next);
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            if (flush) {
                next.invokeWriteAndFlush(m, promise);
            } else {
                next.invokeWrite(m, promise);
            }
        } else {
            final AbstractWriteTask task;
            if (flush) {
                task = WriteAndFlushTask.newInstance(next, m, promise);
            }  else {
                task = WriteTask.newInstance(next, m, promise);
            }
            if (!safeExecute(executor, task, promise, m)) {
                task.cancel();
            }
        }
    }

private AbstractChannelHandlerContext findContextOutbound(int mask) {
    AbstractChannelHandlerContext ctx = this;
    do {
        ctx = ctx.prev;
    } while ((ctx.executionMask & mask) == 0);
    return ctx;
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
    unsafe.write(msg, promise);
}