Netty服务端启动流程

NIO是一个IO多路复用的一个概念，我们看到的epoll，kqueue是不同操作系统对nio不同的实现

EventLoop :事件循环，简单来说就是 某个地方产生事件（也可以是eventLoop自己），然后他自己又去消费事件，造成了一个事件循环，简单实现就是 产生的事件放入一个队列，然后一个线程不断的从队列里面去事件执行（这也就是为什么所有操作都异步，因为每个操作都可以变成一个可执行事件放入事件队列，扔进去一个ChannelFuture进行回调获取执行结果）

NioEventLoopGroup 这个东西是用来选择一个EventLoop的用于channel注册的

 public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup boss = new NioEventLoopGroup(1);
        NioEventLoopGroup work = new NioEventLoopGroup();

        ServerBootstrap serverBootstrap=new ServerBootstrap();
        ChannelFuture bind = serverBootstrap.group(boss, work).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel ch) throws Exception {
//                ch.pipeline().addLast();
            }
        }).bind(8080);

        bind.channel().closeFuture().sync();


    }

NioServerSocketChannel初始化

 private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
    public NioServerSocketChannel() {
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));
    }

   public NioServerSocketChannel(ServerSocketChannel channel) {
        super(null, channel, SelectionKey.OP_ACCEPT);
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }

protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent);
        this.ch = ch;
        this.readInterestOp = readInterestOp;
        try {
            //netty不支持nio是阻塞的
            ch.configureBlocking(false);
        } catch (IOException e) {
            try {
                ch.close();
            } catch (IOException e2) {
                logger.warn(
                            "Failed to close a partially initialized socket.", e2);
            }

            throw new ChannelException("Failed to enter non-blocking mode.", e);
        }
    }
   protected AbstractChannel(Channel parent) {
        this.parent = parent;
        id = newId();
        unsafe = newUnsafe(); //
        pipeline = newChannelPipeline();
    }

UnSafe

    protected AbstractNioUnsafe newUnsafe() {
        return new NioMessageUnsafe();
    }

    private final class NioMessageUnsafe extends AbstractNioUnsafe {

        private final List<Object> readBuf = new ArrayList<Object>();

        @Override
        public void read() {
            assert eventLoop().inEventLoop();
            final ChannelConfig config = config();
            final ChannelPipeline pipeline = pipeline();
            final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
            allocHandle.reset(config);

            boolean closed = false;
            Throwable exception = null;
            try {
                try {
                    do {
                        int localRead = doReadMessages(readBuf);
                        if (localRead == 0) {
                            break;
                        }
                        if (localRead < 0) {
                            closed = true;
                            break;
                        }

                        allocHandle.incMessagesRead(localRead);
                    } while (allocHandle.continueReading());
                } catch (Throwable t) {
                    exception = t;
                }

                int size = readBuf.size();
                for (int i = 0; i < size; i ++) {
                    readPending = false;
                    pipeline.fireChannelRead(readBuf.get(i));
                }
                readBuf.clear();
                allocHandle.readComplete();
                pipeline.fireChannelReadComplete();

                if (exception != null) {
                    closed = closeOnReadError(exception);

                    pipeline.fireExceptionCaught(exception);
                }

                if (closed) {
                    inputShutdown = true;
                    if (isOpen()) {
                        close(voidPromise());
                    }
                }
            } finally {
                // Check if there is a readPending which was not processed yet.
                // This could be for two reasons:
                // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
                // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
                //
                // See https://github.com/netty/netty/issues/2254
                if (!readPending && !config.isAutoRead()) {
                    removeReadOp();
                }
            }
        }
    }


 protected abstract class AbstractNioUnsafe extends AbstractUnsafe implements NioUnsafe {

        protected final void removeReadOp() {
            SelectionKey key = selectionKey();
            // Check first if the key is still valid as it may be canceled as part of the deregistration
            // from the EventLoop
            // See https://github.com/netty/netty/issues/2104
            if (!key.isValid()) {
                return;
            }
            int interestOps = key.interestOps();
            if ((interestOps & readInterestOp) != 0) {
                // only remove readInterestOp if needed
                key.interestOps(interestOps & ~readInterestOp);
            }
        }

        @Override
        public final SelectableChannel ch() {
            return javaChannel();
        }

        @Override
        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)) {
                    fulfillConnectPromise(promise, wasActive);
                } else {
                    connectPromise = promise;
                    requestedRemoteAddress = remoteAddress;

                    // Schedule connect timeout.
                    int connectTimeoutMillis = config().getConnectTimeoutMillis();
                    if (connectTimeoutMillis > 0) {
                        connectTimeoutFuture = eventLoop().schedule(new Runnable() {
                            @Override
                            public void run() {
                                ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
                                ConnectTimeoutException cause =
                                        new ConnectTimeoutException("connection timed out: " + remoteAddress);
                                if (connectPromise != null && connectPromise.tryFailure(cause)) {
                                    close(voidPromise());
                                }
                            }
                        }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
                    }

                    promise.addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture future) throws Exception {
                            if (future.isCancelled()) {
                                if (connectTimeoutFuture != null) {
                                    connectTimeoutFuture.cancel(false);
                                }
                                connectPromise = null;
                                close(voidPromise());
                            }
                        }
                    });
                }
            } catch (Throwable t) {
                promise.tryFailure(annotateConnectException(t, remoteAddress));
                closeIfClosed();
            }
        }

        private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
            if (promise == null) {
                // Closed via cancellation and the promise has been notified already.
                return;
            }

            // Get the state as trySuccess() may trigger an ChannelFutureListener that will close the Channel.
            // We still need to ensure we call fireChannelActive() in this case.
            boolean active = isActive();

            // trySuccess() will return false if a user cancelled the connection attempt.
            boolean promiseSet = promise.trySuccess();

            // Regardless if the connection attempt was cancelled, channelActive() event should be triggered,
            // because what happened is what happened.
            if (!wasActive && active) {
                pipeline().fireChannelActive();
            }

            // If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().
            if (!promiseSet) {
                close(voidPromise());
            }
        }

        private void fulfillConnectPromise(ChannelPromise promise, Throwable cause) {
            if (promise == null) {
                // Closed via cancellation and the promise has been notified already.
                return;
            }

            // Use tryFailure() instead of setFailure() to avoid the race against cancel().
            promise.tryFailure(cause);
            closeIfClosed();
        }

        @Override
        public final void finishConnect() {
            // Note this method is invoked by the event loop only if the connection attempt was
            // neither cancelled nor timed out.

            assert eventLoop().inEventLoop();

            try {
                boolean wasActive = isActive();
                doFinishConnect();
                fulfillConnectPromise(connectPromise, wasActive);
            } catch (Throwable t) {
                fulfillConnectPromise(connectPromise, annotateConnectException(t, requestedRemoteAddress));
            } finally {
                // Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0 is used
                // See https://github.com/netty/netty/issues/1770
                if (connectTimeoutFuture != null) {
                    connectTimeoutFuture.cancel(false);
                }
                connectPromise = null;
            }
        }

        @Override
        protected final void flush0() {
            // Flush immediately only when there's no pending flush.
            // If there's a pending flush operation, event loop will call forceFlush() later,
            // and thus there's no need to call it now.
            if (!isFlushPending()) {
                super.flush0();
            }
        }

        @Override
        public final void forceFlush() {
            // directly call super.flush0() to force a flush now
            super.flush0();
        }

        private boolean isFlushPending() {
            SelectionKey selectionKey = selectionKey();
            return selectionKey.isValid() && (selectionKey.interestOps() & SelectionKey.OP_WRITE) != 0;
        }
    }


AbstractUnsafe 太多了

Pipeline

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

pipeline的添加handler

  p.addLast();

    public final ChannelPipeline addLast(ChannelHandler... handlers) {
        return addLast(null, handlers);
    }

 public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
        ObjectUtil.checkNotNull(handlers, "handlers");

        for (ChannelHandler h: handlers) {
            if (h == null) {
                break;
            }
            //遍历每个拦截器，逐个添加
            addLast(executor, null, h);
        }

        return this;
    }

/*
 private static void checkMultiplicity(ChannelHandler handler) {
 检查handler
 
        if (handler instanceof ChannelHandlerAdapter) {
        //意思就是同一个ChannelHandlerAdapter类型的handler，如果没有@Sharable注解
        //就不能多次添加和移除
        
            ChannelHandlerAdapter h = (ChannelHandlerAdapter) handler;
            //看有没有带@Sharable注解，没有带并且已经添加过了直接报错
            if (!h.isSharable() && h.added) {
                throw new ChannelPipelineException(
                        h.getClass().getName() +
                        " is not a @Sharable handler, so can't be added or removed multiple times.");
            }
            h.added = true;
        }
    }


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

*/

 public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
        final AbstractChannelHandlerContext newCtx;
        synchronized (this) {
            //
            checkMultiplicity(handler);
			// 把handler包装成一个 ChannelHandlerContext
            newCtx = newContext(group, filterName(name, handler), handler);

            addLast0(newCtx);

            if (!registered) {
                //如果还没有注册，那么pipeline的等待队列放入次handler的包装HandlerContext对象
                //然后发射handle传播添加事件
                newCtx.setAddPending();
                callHandlerCallbackLater(newCtx, true);
                return this;
            }

           
            EventExecutor executor = newCtx.executor();
            if (!executor.inEventLoop()) {
           		     
                //如果ctx的执行线程不是ctx的事件循环线程，那么交给指定的事件循环线程处理，可能是IO线程，也可能是自定义线程
                callHandlerAddedInEventLoop(newCtx, executor);
                return this;
            }
        }
     //这里只有在channel已经注册并且 ctx处理线程就是IO线程，并且已经事件循环了，才会进来
        callHandlerAdded0(newCtx);
        return this;
    }


// 事件循环线程的启动
    private void callHandlerAddedInEventLoop(final AbstractChannelHandlerContext newCtx, EventExecutor executor) {
        newCtx.setAddPending();
        executor.execute(new Runnable() {
            @Override
            public void run() {
                callHandlerAdded0(newCtx);
            }
        });
    }


private void execute(Runnable task, boolean immediate) {
        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 && immediate) {
            wakeup(inEventLoop);
        }
    }

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

EventLoopGroup的初始化

NioEventLoopGroup(1)

    public NioEventLoopGroup(int nThreads) {
        this(nThreads, (Executor) null);
    }
    public NioEventLoopGroup(int nThreads, Executor executor) {
        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,
                                            EventExecutorChooserFactory chooserFactory, Object... args) {
        if (nThreads <= 0) {
            throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
        }

        if (executor == null) {
            executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
        }
		// 一个group就是用来选择 EventLoop的，所以用来一个数组来记录有多少个执行器
        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;
                        }
                    }
                }
            }
        }
	
     //根据children的长度选择 合适的 Channel注册选择器
        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);
    }



//初始化执行器

    protected EventLoop newChild(Executor executor, Object... args) throws Exception {
        EventLoopTaskQueueFactory queueFactory = args.length == 4 ? (EventLoopTaskQueueFactory) args[3] : null;
        return new NioEventLoop(this, executor, (SelectorProvider) args[0],
            ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2], queueFactory);
    }

EventLoopGroup注册

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

MultithreadEventExecutorGroup
    @Override
    public EventExecutor next() {
        return chooser.next();//其实就是在children里面 选一个，怎么选，就是 ++ 对长度children取模
    }


SingleThreadEventLoop
    public ChannelFuture register(Channel channel) {
        return register(new DefaultChannelPromise(channel, this));
    }
//Promise的概念与JS的promise概念差不多，都是用于异步编程，执行完毕进行回调

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

AbstractChannel#AbstractUnsafe
 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;
            }
			//此时Channel的对应的 事件循环线程指定
            AbstractChannel.this.eventLoop = eventLoop;
			
            if (eventLoop.inEventLoop()) {
                //如果当前线程是eventLoop事件循环线程，那就直接注册
                register0(promise);
            } else {
                // 如果事件循环线程为启动，那么，交给事件循环线程处理
                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);
                }
            }
        }

//此时被异步了，交给了io线程处理，而不是主线程
 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;

                //使用pipeline进行注册前等待中的handler的add事件
               
                pipeline.invokeHandlerAddedIfNeeded();
				//安全设置成功，其实就添加一个事件到evenLoop中
                safeSetSuccess(promise);
                
                //pipeline发射channel注册事件
                pipeline.fireChannelRegistered();

                if (isActive()) {
                    //是不是已经启动了
                    if (firstRegistration) {
                        //如果是第一次注册，那么 发射 Channel活跃可用事件
                        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.
                        beginRead();
                    }
                }
            } catch (Throwable t) {
                // Close the channel directly to avoid FD leak.
                closeForcibly();
                closeFuture.setClosed();
                safeSetFailure(promise, t);
            }
        }
AbstractNioChannel
//真正的注册，
 protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            try {
                //java原生channel 注册到 selector中，attach对像就是以后要用到数据
                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;
                }
            }
        }
    }

NioEventLoop初始化

  NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
               SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,
               EventLoopTaskQueueFactory queueFactory) {
      super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory),
              rejectedExecutionHandler);
      this.provider = ObjectUtil.checkNotNull(selectorProvider, "selectorProvider");
      this.selectStrategy = ObjectUtil.checkNotNull(strategy, "selectStrategy");
      final SelectorTuple selectorTuple = openSelector();//对selector做的一层包装
      this.selector = selectorTuple.selector;
      this.unwrappedSelector = selectorTuple.unwrappedSelector;//原生态selector
  }

protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,
                                  boolean addTaskWakesUp, Queue<Runnable> taskQueue, Queue<Runnable> tailTaskQueue,
                                  RejectedExecutionHandler rejectedExecutionHandler) {
      super(parent, executor, addTaskWakesUp, taskQueue, rejectedExecutionHandler);
      tailTasks = ObjectUtil.checkNotNull(tailTaskQueue, "tailTaskQueue");
  }


 protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
                                      boolean addTaskWakesUp, Queue<Runnable> taskQueue,
                                      RejectedExecutionHandler rejectedHandler) {
      super(parent);
      this.addTaskWakesUp = addTaskWakesUp;
      this.maxPendingTasks = DEFAULT_MAX_PENDING_EXECUTOR_TASKS;
     // 多层包装，其实最后就是一个线程在执行
      this.executor = ThreadExecutorMap.apply(executor, this);
      this.taskQueue = ObjectUtil.checkNotNull(taskQueue, "taskQueue");
      this.rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
  }

ServerBootstrap

serverBootstrap.group(boss, work).channel(NioServerSocketChannel.class).handler(serverHandler).childHandler(childHandler)

serverBootstrap.group =  boss
serverBootstrap.childGroup =  work
serverBootstrap.channelFactory =  创建NioServerSocketChannel的一个工厂
serverBootstrap.handler  =   serverHandler
serverBootstrap.childHandler = childHandler

init（）

这个方法主要是对Channel的初始化

@Override
    void init(Channel channel) throws Exception {
        //拿到操作集合
        final Map<ChannelOption<?>, Object> options = options();
        synchronized (options) {
            //配置channel的config
            channel.config().setOptions(options);
        }
		//拿到属性集合
        final Map<AttributeKey<?>, Object> attrs = attrs();
        synchronized (attrs) {
            for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
                @SuppressWarnings("unchecked")
                AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
                //设置channel属性
                channel.attr(key).set(e.getValue());
            }
        }
		//每一个channel在初始化是就会带着把它对应的pipeline初始化
        ChannelPipeline p = channel.pipeline();
        if (handler() != null) {
            //pipeline加上执行的处理器
            p.addLast(handler());
        }

        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(childOptions.size()));
        }
        synchronized (childAttrs) {
            currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
        }
        //上面是获取等下接待的客户端channel的一些初始操作
        
		// 又添加一个handler进行对ServerChannel的处理
        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(Channel ch) throws Exception {
                ch.pipeline().addLast(new ServerBootstrapAcceptor(
                        currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
            }
        });
    }

initAndRegister()

final ChannelFuture initAndRegister() {
      final Channel channel = channelFactory().newChannel();
      try {
          init(channel);
      } catch (Throwable t) {
          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);
      }
//注册操作
      ChannelFuture regFuture = group().register(channel);
      if (regFuture.cause() != null) {
          //注册失败了，那就关闭
          if (channel.isRegistered()) {
              channel.close();
          } else {
              channel.unsafe().closeForcibly();
          }
      }
      return regFuture;
  }

bind(8080)

public ChannelFuture bind(int inetPort) {
       return bind(new InetSocketAddress(inetPort));
   }

  public ChannelFuture bind(SocketAddress localAddress) {
       validate();
       if (localAddress == null) {
           throw new NullPointerException("localAddress");
       }
       return doBind(localAddress);
   }


private ChannelFuture doBind(final SocketAddress localAddress) {
    // 初始化并且注册
       final ChannelFuture regFuture = initAndRegister();
       final Channel channel = regFuture.channel();
       if (regFuture.cause() != null) {
           return regFuture;
       }

       if (regFuture.isDone()) {
           // At this point we know that the registration was complete and successful.
           //如果注册成功，那就开始真正的绑定
           ChannelPromise promise = channel.newPromise();
           doBind0(regFuture, channel, localAddress, promise);
           return promise;
       } else {
           // Registration future is almost always fulfilled already, but just in case it's not.
           //如果还没有注册成功
           
           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) {
                       // 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.setFailure(cause);
                   } else {
                       // Registration was successful, so set the correct executor to use.
                       // See https://github.com/netty/netty/issues/2586
                       promise.executor = channel.eventLoop();
                   }
                   doBind0(regFuture, channel, localAddress, promise);
               }
           });
           return promise;
       }
   }

doBind0

private static void doBind0(
          final ChannelFuture regFuture, final Channel channel,
          final SocketAddress localAddress, final ChannelPromise promise) {

      // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
      // the pipeline in its channelRegistered() implementation.
      channel.eventLoop().execute(new Runnable() {
          @Override
          public void run() {
              if (regFuture.isSuccess()) {
                  channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
              } else {
                  promise.setFailure(regFuture.cause());
              }
          }
      });
  }

Netty服务端接待流程

NioEventLoop的run

protected void run() {
       int selectCnt = 0;
       for (;;) {
           try {
               int strategy;
               try {
                   //获取一个策略
                   // 默认的，如果任务队列有事件处理的任务，那么先使用 selector.selectNow()，来看是不是有io事件，如果没有io事件，那就是0，任务队列没有任务，那么就是执行SELECT那一段
                   strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
                   switch (strategy) {
                   case SelectStrategy.CONTINUE:
                       continue;

                   case SelectStrategy.BUSY_WAIT:
                       // fall-through to SELECT since the busy-wait is not supported with NIO

                   case SelectStrategy.SELECT:
                           //这是没有任务的情况下执行的
                           //获取下一次计划执行任务时间，这里就是防止没有任务或者没有IO事件时，线程频繁执行，浪费cpu
                       long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
                       if (curDeadlineNanos == -1L) {
                           curDeadlineNanos = NONE; // nothing on the calendar
                       }
                       nextWakeupNanos.set(curDeadlineNanos);
                       try {
                           if (!hasTasks()) {
                               strategy = select(curDeadlineNanos);
                           }
                       } finally {
                           // This update is just to help block unnecessary selector wakeups
                           // so use of lazySet is ok (no race condition)
                           nextWakeupNanos.lazySet(AWAKE);
                       }
                       // fall through
                   default:
                   }
               } catch (IOException e) {
                   // If we receive an IOException here its because the Selector is messed up. Let's rebuild
                   // the selector and retry. https://github.com/netty/netty/issues/8566
                   rebuildSelector0();
                   selectCnt = 0;
                   handleLoopException(e);
                   continue;
               }
			// 这一步，一定有任务或者io事件
               selectCnt++;
               cancelledKeys = 0;
               needsToSelectAgain = false;
               //这个io比率代表，执行 完io操作后所用时间 乘 ioRatio 就是执行任务的限制时间
               final int ioRatio = this.ioRatio;
               boolean ranTasks;
               if (ioRatio == 100) {
                   try {
                       if (strategy > 0) {
                           processSelectedKeys();
                       }
                   } finally {
                       // Ensure we always run tasks.
                       ranTasks = runAllTasks();
                   }
               } else if (strategy > 0) {
                   
                   final long ioStartTime = System.nanoTime();
                   try {
                       processSelectedKeys();
                   } finally {
                       // Ensure we always run tasks.
                       final long ioTime = System.nanoTime() - ioStartTime;
                       ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                   }
               } else {
                   //虽然比率不是百分百，但是，没有感兴趣的io操作过来，于是执行任务
                   ranTasks = runAllTasks(0); // This will run the minimum number of tasks
               }

               if (ranTasks || strategy > 0) {
                   // 只有在io和任务都处理了一下的情况才会进来
                   if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS && logger.isDebugEnabled()) {
                       logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
                               selectCnt - 1, selector);
                   }
                   selectCnt = 0;
               } else if (unexpectedSelectorWakeup(selectCnt)) { // Unexpected wakeup (unusual case)
                   selectCnt = 0;
               }
           } catch (CancelledKeyException e) {
               // Harmless exception - log anyway
               if (logger.isDebugEnabled()) {
                   logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?",
                           selector, e);
               }
           } catch (Throwable t) {
               handleLoopException(t);
           }
           
           //正常关闭
           // Always handle shutdown even if the loop processing threw an exception.
           try {
               if (isShuttingDown()) {
                   closeAll();
                   if (confirmShutdown()) {
                       return;
                   }
               }
           } catch (Throwable t) {
               handleLoopException(t);
           }
       }
   }

processSelectedKeys();

    private void processSelectedKeys() {
        if (selectedKeys != null) {
            processSelectedKeysOptimized();
        } else {
            processSelectedKeysPlain(selector.selectedKeys());
        }
    }


//对处理SelectedKeys的优化
  private void processSelectedKeysOptimized() {
        for (int i = 0; i < selectedKeys.size; ++i) {
            final SelectionKey k = selectedKeys.keys[i];
            // null out entry in the array to allow to have it GC'ed once the Channel close
            // See https://github.com/netty/netty/issues/2363
            selectedKeys.keys[i] = null;

            final Object a = k.attachment();

            if (a instanceof AbstractNioChannel) {
                processSelectedKey(k, (AbstractNioChannel) a);
            } else {
                @SuppressWarnings("unchecked")
                NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
                processSelectedKey(k, task);
            }

            if (needsToSelectAgain) {
                // null out entries in the array to allow to have it GC'ed once the Channel close
                // See https://github.com/netty/netty/issues/2363
                selectedKeys.reset(i + 1);

                selectAgain();
                i = -1;
            }
        }
    }

processSelectedKey

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
       final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
       if (!k.isValid()) {
           final EventLoop eventLoop;
           try {
               eventLoop = ch.eventLoop();
           } catch (Throwable ignored) {
               // If the channel implementation throws an exception because there is no event loop, we ignore this
               // because we are only trying to determine if ch is registered to this event loop and thus has authority
               // to close ch.
               return;
           }
           // Only close ch if ch is still registered to this EventLoop. ch could have deregistered from the event loop
           // and thus the SelectionKey could be cancelled as part of the deregistration process, but the channel is
           // still healthy and should not be closed.
           if (eventLoop == this) {
               // close the channel if the key is not valid anymore
               unsafe.close(unsafe.voidPromise());
           }
           return;
       }
	// 上面对Channel关闭无效时的处理
    
    
       try {
           int readyOps = k.readyOps();
           // We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
           // the NIO JDK channel implementation may throw a NotYetConnectedException.
           if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
               // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
               // See https://github.com/netty/netty/issues/924
               int ops = k.interestOps();
               ops &= ~SelectionKey.OP_CONNECT;
               k.interestOps(ops);

               unsafe.finishConnect();
           }

           // Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
           if ((readyOps & SelectionKey.OP_WRITE) != 0) {
               // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
               ch.unsafe().forceFlush();
           }

           // 还要检查0的readOps，以解决可能导致自旋循环的JDK错误
           if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
               unsafe.read();
           }
       } catch (CancelledKeyException ignored) {
           unsafe.close(unsafe.voidPromise());
       }
   }

服务端接待客户端

@Override
     public void read() {
         assert eventLoop().inEventLoop();
         final ChannelConfig config = config();
         final ChannelPipeline pipeline = pipeline();
         final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
         allocHandle.reset(config);

         boolean closed = false;
         Throwable exception = null;
         try {
             try {
                 do {
                     int localRead = doReadMessages(readBuf);
                     if (localRead == 0) {
                         break;
                     }
                     if (localRead < 0) {
                         closed = true;
                         break;
                     }
			//获取接待的客户端Channel
                     allocHandle.incMessagesRead(localRead);
                 } while (allocHandle.continueReading());
             } catch (Throwable t) {
                 exception = t;
             }
	
             int size = readBuf.size();
             for (int i = 0; i < size; i ++) {
                 readPending = false;
                 // 发射Channel可读事件
                 pipeline.fireChannelRead(readBuf.get(i));
             }
             readBuf.clear();
             allocHandle.readComplete();
             //发射读取完成的事件
             pipeline.fireChannelReadComplete();

             if (exception != null) {
                 closed = closeOnReadError(exception);
		//发射异常事件
                 pipeline.fireExceptionCaught(exception);
             }

             if (closed) {
                 inputShutdown = true;
                 if (isOpen()) {
                     close(voidPromise());
                 }
             }
         } finally {
             // Check if there is a readPending which was not processed yet.
             // This could be for two reasons:
             // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
             // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
             if (!readPending && !config.isAutoRead()) {
                 removeReadOp();
             }
         }
     }

ServerBootstrapAcceptor

这是一个服务端的Hanlder

初始化

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;

           // Task which is scheduled to re-enable auto-read.
           // It's important to create this Runnable before we try to submit it as otherwise the URLClassLoader may
           // not be able to load the class because of the file limit it already reached.
           //
           // See https://github.com/netty/netty/issues/1328
           enableAutoReadTask = new Runnable() {
               @Override
               public void run() {
                   channel.config().setAutoRead(true);
               }
           };
       }

执行服务端接待事件

@SuppressWarnings("unchecked")
      public void channelRead(ChannelHandlerContext ctx, Object msg) {
          final Channel child = (Channel) msg;
	//加上指定的处理器
          child.pipeline().addLast(childHandler);
	//设置操作
          setChannelOptions(child, childOptions, logger);
          //设置属性
          setAttributes(child, childAttrs);

          try {
              //注册到eventloop中，过程与服务端相同
              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);
          }
      }

客户端流程

启动就是注册事件循环组，然后connect到指定服务器

客户端读取

 
NioByteUnsafe
@Override
        public final void read() {
            final ChannelConfig config = config();
            if (shouldBreakReadReady(config)) {
                clearReadPending();
                return;
            }
            final ChannelPipeline pipeline = pipeline();
            final ByteBufAllocator allocator = config.getAllocator();
            final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
            allocHandle.reset(config);

            ByteBuf byteBuf = null;
            boolean close = false;
            try {
                do {
                    byteBuf = allocHandle.allocate(allocator);
                    // 这里把io内容读取到
                    allocHandle.lastBytesRead(doReadBytes(byteBuf));
                    if (allocHandle.lastBytesRead() <= 0) {
                        // nothing was read. release the buffer.
                        //没有读到任何东西，回收或者释放
                        byteBuf.release();
                        byteBuf = null;
                        close = allocHandle.lastBytesRead() < 0;
                        if (close) {
                            // There is nothing left to read as we received an EOF.
                            readPending = false;
                        }
                        break;
                    }

                    allocHandle.incMessagesRead(1);
                    readPending = false;
                    //发射读取事件
                    pipeline.fireChannelRead(byteBuf);
                    byteBuf = null;
                } while (allocHandle.continueReading());

                allocHandle.readComplete();
                //发射读取完成事件
                pipeline.fireChannelReadComplete();

                if (close) {
                    //触发用户事件
                    closeOnRead(pipeline);
                }
            } catch (Throwable t) {
                //发射异常
                handleReadException(pipeline, byteBuf, t, close, allocHandle);
            } finally {
                // Check if there is a readPending which was not processed yet.
                // This could be for two reasons:
                // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
                // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
                //
                // See https://github.com/netty/netty/issues/2254
                if (!readPending && !config.isAutoRead()) {
                    removeReadOp();
                }
            }
        }

写消息

         if ((readyOps & SelectionKey.OP_WRITE) != 0) {
               // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
               ch.unsafe().forceFlush();
           }
      public final void forceFlush() {
           // directly call super.flush0() to force a flush now
           super.flush0();
       }


protected void flush0() {
           if (inFlush0) {
               // Avoid re-entrance
               return;
           }

           final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
           if (outboundBuffer == null || outboundBuffer.isEmpty()) {
               return;
           }

           inFlush0 = true;

           // Mark all pending write requests as failure if the channel is inactive.
           if (!isActive()) {
               try {
                   if (isOpen()) {
                       outboundBuffer.failFlushed(new NotYetConnectedException(), true);
                   } else {
                       // Do not trigger channelWritabilityChanged because the channel is closed already.
                       outboundBuffer.failFlushed(newClosedChannelException(initialCloseCause), false);
                   }
               } finally {
                   inFlush0 = false;
               }
               return;
           }

           try {
               doWrite(outboundBuffer);
           } catch (Throwable t) {
               if (t instanceof IOException && config().isAutoClose()) {
                   initialCloseCause = t;
                   close(voidPromise(), t, newClosedChannelException(t), false);
               } else {
                   try {
                       shutdownOutput(voidPromise(), t);
                   } catch (Throwable t2) {
                       initialCloseCause = t;
                       close(voidPromise(), t2, newClosedChannelException(t), false);
                   }
               }
           } finally {
               inFlush0 = false;
           }
       }

客户端连接操作

   if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
               // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
               // See https://github.com/netty/netty/issues/924
               int ops = k.interestOps();
               ops &= ~SelectionKey.OP_CONNECT;
               k.interestOps(ops);

               unsafe.finishConnect();
           }

@Override
       public final void finishConnect() {
           // Note this method is invoked by the event loop only if the connection attempt was
           // neither cancelled nor timed out.

           assert eventLoop().inEventLoop();

           try {
               boolean wasActive = isActive();
               doFinishConnect();
               fulfillConnectPromise(connectPromise, wasActive);
           } catch (Throwable t) {
               fulfillConnectPromise(connectPromise, annotateConnectException(t, requestedRemoteAddress));
           } finally {
               // Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0 is used
               // See https://github.com/netty/netty/issues/1770
               if (connectTimeoutFuture != null) {
                   connectTimeoutFuture.cancel(false);
               }
               connectPromise = null;
           }
       }


private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
           if (promise == null) {
               // Closed via cancellation and the promise has been notified already.
               return;
           }

           // Get the state as trySuccess() may trigger an ChannelFutureListener that will close the Channel.
           // We still need to ensure we call fireChannelActive() in this case.
           boolean active = isActive();

           // trySuccess() will return false if a user cancelled the connection attempt.
           boolean promiseSet = promise.trySuccess();

           // Regardless if the connection attempt was cancelled, channelActive() event should be triggered,
           // because what happened is what happened.
           if (!wasActive && active) {
               pipeline().fireChannelActive();
           }

           // If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().
           if (!promiseSet) {
               close(voidPromise());
           }
       }

pipeline和ChannelHandlerContext的事件传播

static final int MASK_ONLY_INBOUND =  MASK_CHANNEL_REGISTERED |
           MASK_CHANNEL_UNREGISTERED | MASK_CHANNEL_ACTIVE | MASK_CHANNEL_INACTIVE | MASK_CHANNEL_READ |
           MASK_CHANNEL_READ_COMPLETE | MASK_USER_EVENT_TRIGGERED | MASK_CHANNEL_WRITABILITY_CHANGED;
   private static final int MASK_ALL_INBOUND = MASK_EXCEPTION_CAUGHT | MASK_ONLY_INBOUND;
   static final int MASK_ONLY_OUTBOUND =  MASK_BIND | MASK_CONNECT | MASK_DISCONNECT |
           MASK_CLOSE | MASK_DEREGISTER | MASK_READ | MASK_WRITE | MASK_FLUSH;
   private static final int MASK_ALL_OUTBOUND = MASK_EXCEPTION_CAUGHT | MASK_ONLY_OUTBOUND;

入站事件

从pipeline的head开始，直到tail，注意入站事件 netty默认处理器不处理

注册事件传播

DefaultChannelPipeline

    public final ChannelPipeline fireChannelRegistered() {
        AbstractChannelHandlerContext.invokeChannelRegistered(head);
        return this;
    }

AbstractChannelHandlerContext
   static void invokeChannelRegistered(final AbstractChannelHandlerContext next) {
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeChannelRegistered();
        } else {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    next.invokeChannelRegistered();
                }
            });
        }
    }


private void invokeChannelRegistered() {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelRegistered(this);
            } catch (Throwable t) {
                invokeExceptionCaught(t);
            }
        } else {
            fireChannelRegistered();
        }
    }


//head 开始
        public void channelRegistered(ChannelHandlerContext ctx) {
            invokeHandlerAddedIfNeeded();
            ctx.fireChannelRegistered();
        }
//然后交给 ctx 进行对事件传播
   @Override
    public ChannelHandlerContext fireChannelRegistered() {
        invokeChannelRegistered(findContextInbound(MASK_CHANNEL_REGISTERED));
        return this;
    }
//ctx去找下一个ctx
  private AbstractChannelHandlerContext findContextInbound(int mask) {
      //这里的mask代表了入站和出站的handler的表示
        AbstractChannelHandlerContext ctx = this;
        EventExecutor currentExecutor = executor();
        do {
            ctx = ctx.next;
        } while (skipContext(ctx, currentExecutor, mask, MASK_ONLY_INBOUND));
        return ctx;
    }

 static void invokeChannelRegistered(final AbstractChannelHandlerContext next) {
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeChannelRegistered();
        } else {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    next.invokeChannelRegistered();
                }
            });
        }
    }

 private void invokeChannelRegistered() {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelRegistered(this);
            } catch (Throwable t) {
                invokeExceptionCaught(t);
            }
        } else {
            fireChannelRegistered();
        }
    }

最后一个入站 处理器 TailContext
     public void channelRegistered(ChannelHandlerContext ctx) { }

Channel活跃状态事件传播

就是可以与客户端或者服务端进行连接了

  
DefaultChannelPipeline

public final ChannelPipeline fireChannelActive() {
        AbstractChannelHandlerContext.invokeChannelActive(head);
        return this;
    }
AbstractChannelHandlerContext
  private void invokeChannelActive() {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelActive(this);
            } catch (Throwable t) {
                invokeExceptionCaught(t);
            }
        } else {
            fireChannelActive();
        }
    }

HeadContext
      @Override
        public void channelActive(ChannelHandlerContext ctx) {
            ctx.fireChannelActive();

            readIfIsAutoRead();
        }

Channel可读事件传播

DefaultChannelPipeline
   @Override
    public final ChannelPipeline fireChannelRead(Object msg) {
        AbstractChannelHandlerContext.invokeChannelRead(head, msg);
        return this;
    }
AbstractChannelHandlerContext
 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);
                }
            });
        }
    }
AbstractChannelHandlerContext
    private void invokeChannelRead(Object msg) {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelRead(this, msg);
            } catch (Throwable t) {
                invokeExceptionCaught(t);
            }
        } else {
            fireChannelRead(msg);
        }
    }

head
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            ctx.fireChannelRead(msg);
        }

tail
      @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            onUnhandledInboundMessage(ctx, msg);
        }
        protected void onUnhandledInboundMessage(ChannelHandlerContext ctx, Object msg) {
        onUnhandledInboundMessage(msg);
        if (logger.isDebugEnabled()) {
            logger.debug("Discarded message pipeline : {}. Channel : {}.",
                         ctx.pipeline().names(), ctx.channel());
        }
    }

    protected void onUnhandledInboundMessage(Object msg) {
        try {
            logger.debug(
                    "Discarded inbound message {} that reached at the tail of the pipeline. " +
                            "Please check your pipeline configuration.", msg);
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

读取完成事件传播

ChannelPipeline
    public final ChannelPipeline fireChannelReadComplete() {
        AbstractChannelHandlerContext.invokeChannelReadComplete(head);
        return this;
    }
AbstractChannelHandlerContext

   static void invokeChannelReadComplete(final AbstractChannelHandlerContext next) {
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeChannelReadComplete();
        } else {
            Tasks tasks = next.invokeTasks;
            if (tasks == null) {
                next.invokeTasks = tasks = new Tasks(next);
            }
            executor.execute(tasks.invokeChannelReadCompleteTask);
        }
    }

    private void invokeChannelReadComplete() {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelReadComplete(this);
            } catch (Throwable t) {
                invokeExceptionCaught(t);
            }
        } else {
            fireChannelReadComplete();
        }
    }

head
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) {
            ctx.fireChannelReadComplete();

            readIfIsAutoRead();
        }

出站事件

从pipeline的tail开始，直到head，出站处理器一定要记得发射对应事件，除非已经处理过了，netty默认会处理

绑定事件传播

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

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

注意tail本身没有出站的能力，所以直接开始找下一个出站处理器
public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
        ObjectUtil.checkNotNull(localAddress, "localAddress");
        if (isNotValidPromise(promise, false)) {
            // cancelled
            return 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, false);
        }
        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);
        }
    }

出站的最后一个就是
HeadContext
       public void bind(
                ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
            unsafe.bind(localAddress, promise);
        }

AbstractUnsafe
  public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
            assertEventLoop();

            if (!promise.setUncancellable() || !ensureOpen(promise)) {
                return;
            }

            // See: https://github.com/netty/netty/issues/576
            if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&
                localAddress instanceof InetSocketAddress &&
                !((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() &&
                !PlatformDependent.isWindows() && !PlatformDependent.maybeSuperUser()) {
                // Warn a user about the fact that a non-root user can't receive a
                // broadcast packet on *nix if the socket is bound on non-wildcard address.
                logger.warn(
                        "A non-root user can't receive a broadcast packet if the socket " +
                        "is not bound to a wildcard address; binding to a non-wildcard " +
                        "address (" + localAddress + ") anyway as requested.");
            }

            boolean wasActive = isActive();
            try {
                doBind(localAddress);
            } catch (Throwable t) {
                safeSetFailure(promise, t);
                closeIfClosed();
                return;
            }

            if (!wasActive && isActive()) {
                //绑定成功后会发射 channel活跃事件
                invokeLater(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.fireChannelActive();
                    }
                });
            }

            safeSetSuccess(promise);
        }

//java原生态绑定
    protected void doBind(SocketAddress localAddress) throws Exception {
        if (PlatformDependent.javaVersion() >= 7) {
            javaChannel().bind(localAddress, config.getBacklog());
        } else {
            javaChannel().socket().bind(localAddress, config.getBacklog());
        }
    }

对读取感兴趣事件传播

AbstractChannel
    public Channel read() {
        pipeline.read();
        return this;
    }
DefaultChannelPipeline
   public final ChannelPipeline read() {
        tail.read();
        return this;
    }
AbstractChannelHandlerContext
    @Override
    public ChannelHandlerContext read() {
        final AbstractChannelHandlerContext next = findContextOutbound(MASK_READ);
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeRead();
        } else {
            Tasks tasks = next.invokeTasks;
            if (tasks == null) {
                next.invokeTasks = tasks = new Tasks(next);
            }
            executor.execute(tasks.invokeReadTask);
        }

        return this;
    }

   private void invokeRead() {
        if (invokeHandler()) {
            try {
                ((ChannelOutboundHandler) handler()).read(this);
            } catch (Throwable t) {
                invokeExceptionCaught(t);
            }
        } else {
            read();
        }
    }


最后一个HeadContext
        public void read(ChannelHandlerContext ctx) {
            unsafe.beginRead();
        }

AbstractUnsafe
  @Override
        public final void beginRead() {
            assertEventLoop();

            if (!isActive()) {
                return;
            }

            try {
                doBeginRead();
            } catch (final Exception e) {
                invokeLater(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.fireExceptionCaught(e);
                    }
                });
                close(voidPromise());
            }
        }


    protected void doBeginRead() throws Exception {
        if (inputShutdown) {
            return;
        }
        super.doBeginRead();
    }
AbstractNioChannel
  protected void doBeginRead() throws Exception {
        // Channel.read() or ChannelHandlerContext.read() was called
        final SelectionKey selectionKey = this.selectionKey;
        if (!selectionKey.isValid()) {
            return;
        }

        readPending = true;

        final int interestOps = selectionKey.interestOps();
        if ((interestOps & readInterestOp) == 0) {
            selectionKey.interestOps(interestOps | readInterestOp);
        }
    }

赏

谢谢你请我吃糖果

支付宝

微信

• 本文作者： 忘忧症
• 本文链接： https://NepenthesZGW.github.io/2020/06/20/framework/Netty/Netty源码流程/
• 版权声明： 本博客所有文章除特别声明外，均采用 MIT 许可协议。转载请注明出处！