南贺神社

<!-- more -->
public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop
  public ChannelFuture register(final ChannelPromise promise) {
      ObjectUtil.checkNotNull(promise, "promise");
      promise.channel().unsafe().register(this, promise);
      return promise;
  }
}

protected abstract class AbstractUnsafe implements Unsafe {

      private volatile ChannelOutboundBuffer outboundBuffer = new ChannelOutboundBuffer(AbstractChannel.this);
      private RecvByteBufAllocator.Handle recvHandle;
      private boolean inFlush0;
      /** true if the channel has never been registered, false otherwise */
      private boolean neverRegistered = true;
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;
          }
           //上边的逻辑主要是一些非空判断之类的东西
          AbstractChannel.this.eventLoop = eventLoop;
    //主要的注册逻辑分支，if和else分支可以看到最后调用的都是register0，但是else里边加了一个外壳---线程
          if (eventLoop.inEventLoop()) {
              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);
              }
          }
      }
}

public interface EventExecutor extends EventExecutorGroup {
...略
  /**
   * Calls {@link #inEventLoop(Thread)} with {@link Thread#currentThread()} as argument
   * 将当前线程作为参数。
   */
  boolean inEventLoop();
  ...略
  }

public abstract class AbstractEventExecutor extends AbstractExecutorService implements EventExecutor {
  ...略
  public boolean inEventLoop() {
      return inEventLoop(Thread.currentThread());//参数为当前线程。
  }
  ...略
}

public boolean inEventLoop(Thread thread) {
    return thread == this.thread;
}

private void register0(ChannelPromise promise) {
    try {
        // check if the channel is still open as it could be closed in the mean time when the register
        // call was outside of the eventLoop
        if (!promise.setUncancellable() || !ensureOpen(promise)) {
            return;
        }
        boolean firstRegistration = neverRegistered;
        doRegister();//核心注册方法
        neverRegistered = false;
        registered = true;

        // Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
        // user may already fire events through the pipeline in the ChannelFutureListener.
        pipeline.invokeHandlerAddedIfNeeded();

        safeSetSuccess(promise);
        pipeline.fireChannelRegistered();
        // Only fire a channelActive if the channel has never been registered. This prevents firing
        // multiple channel actives if the channel is deregistered and re-registered.
        if (isActive()) {
            if (firstRegistration) {
                pipeline.fireChannelActive();
            } else if (config().isAutoRead()) {
                // This channel was registered before and autoRead() is set. This means we need to begin read
                // again so that we process inbound data.
                //
                // See https://github.com/netty/netty/issues/4805
                beginRead();
            }
        }
    } catch (Throwable t) {
        // Close the channel directly to avoid FD leak.
        closeForcibly();
        closeFuture.setClosed();
        safeSetFailure(promise, t);
    }
}

public abstract class AbstractNioChannel extends AbstractChannel {
...略
  protected void doRegister() throws Exception {
      boolean selected = false;
      for (;;) {
          try {
              selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
              //javaChannel()返回的是SelectableChannel(jdk的java.nio.channels.SelectableChannel)
              //javaChannel().register是将channel注册到Selector上去，所以eventLoop().unwrappedSelector()返回的是Selector
              //（jdk的java.nio.channels.Selector）
              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;
              }
          }
      }
  }
  ...略
}