Java 技术栈中间件优雅停机方案设计与实现全景图

//延迟5秒暴露服务<dubbo:service delay="5000" /> 

# kill -l 1) SIGHUP       2) SIGINT       3) SIGQUIT      4) SIGILL       5) SIGTRAP 6) SIGABRT      7) SIGBUS       8) SIGFPE       9) SIGKILL     10) SIGUSR111) SIGSEGV     12) SIGUSR2     13) SIGPIPE     14) SIGALRM     15) SIGTERM16) SIGSTKFLT   17) SIGCHLD     18) SIGCONT     19) SIGSTOP     20) SIGTSTP21) SIGTTIN     22) SIGTTOU     23) SIGURG      24) SIGXCPU     25) SIGXFSZ26) SIGVTALRM   27) SIGPROF     28) SIGWINCH    29) SIGIO       30) SIGPWR31) SIGSYS      34) SIGRTMIN    35) SIGRTMIN+1  36) SIGRTMIN+2  37) SIGRTMIN+338) SIGRTMIN+4  39) SIGRTMIN+5  40) SIGRTMIN+6  41) SIGRTMIN+7  42) SIGRTMIN+843) SIGRTMIN+9  44) SIGRTMIN+10 45) SIGRTMIN+11 46) SIGRTMIN+12 47) SIGRTMIN+1348) SIGRTMIN+14 49) SIGRTMIN+15 50) SIGRTMAX-14 51) SIGRTMAX-13 52) SIGRTMAX-1253) SIGRTMAX-11 54) SIGRTMAX-10 55) SIGRTMAX-9  56) SIGRTMAX-8  57) SIGRTMAX-758) SIGRTMAX-6  59) SIGRTMAX-5  60) SIGRTMAX-4  61) SIGRTMAX-3  62) SIGRTMAX-263) SIGRTMAX-1  64) SIGRTMAX

Signal     Value     Action   Comment──────────────────────────────────────────────────────────────────────SIGHUP        1       Term    Hangup detected on controlling terminal                              or death of controlling processSIGINT        2       Term    Interrupt from keyboardSIGQUIT       3       Core    Quit from keyboardSIGILL        4       Core    Illegal Instruction

SIGABRT       6       Core    Abort signal from abort(3)SIGFPE        8       Core    Floating point exceptionSIGKILL       9       Term    Kill signalSIGSEGV      11       Core    Invalid memory referenceSIGPIPE      13       Term    Broken pipe: write to pipe with no                              readersSIGALRM      14       Term    Timer signal from alarm(2)SIGTERM      15       Term    Termination signalSIGUSR1   30,10,16    Term    User-defined signal 1SIGUSR2   31,12,17    Term    User-defined signal 2……

int sigaction(int signum, const struct sigaction *act,                     struct sigaction *oldact);

struct sigaction {  __sighandler_t sa_handler;  unsigned long sa_flags;        .......  sigset_t sa_mask; };

#include <signal.h>
static void sig_handler(int signum) {
    if (signum == SIGTERM) {
        .....执行优雅关闭逻辑....
    }
}
int main (Void) {
    struct sigaction sa_usr; //定义sigaction结构体    sa_usr.sa_flags = 0;    sa_usr.sa_handler = sig_handler;   //设置信号处理函数
    sigaction(SIGTERM, &sa_usr, NULL);//进程捕获信号，注册信号处理函数                ,,,,,,,,,,,,}

        Runtime.getRuntime().addShutdownHook(new Thread(){            @Override            public void run() {                .....执行优雅关闭步骤.....            }        });

public static void main(String[] args) {        try {
           ......进程启动main函数.......
        } catch (RuntimeException e) {            logger.error(e.getMessage(), e);            // JVM 进程主动关闭触发调用 shutdownHook            System.exit(1);        }}

   public void halt(int status) {        SecurityManager sm = System.getSecurityManager();        if (sm != null) {            sm.checkExit(status);        }        Shutdown.halt(status);    }

int sigaction(int signum, const struct sigaction *act,                     struct sigaction *oldact);

        Runtime.getRuntime().addShutdownHook(new Thread(){            @Override            public void run() {                .....执行优雅关闭步骤.....            }        });

public final class System {
    private static void initializeSystemClass() {
           .......省略.......
            // Setup Java signal handlers for HUP, TERM, and INT (where available).           Terminator.setup();
           .......省略.......
    }
}

class Terminator {    //信号处理函数    private static SignalHandler handler = null;
    static void setup() {        if (handler != null) return;        SignalHandler sh = new SignalHandler() {            public void handle(Signal sig) {                Shutdown.exit(sig.getNumber() + 0200);            }        };        handler = sh;
        try {            Signal.handle(new Signal("HUP"), sh);        } catch (IllegalArgumentException e) {        }        try {            Signal.handle(new Signal("INT"), sh);        } catch (IllegalArgumentException e) {        }        try {            Signal.handle(new Signal("TERM"), sh);        } catch (IllegalArgumentException e) {        }    }
}

int sigaction(int signum, const struct sigaction *act,                     struct sigaction *oldact);

    SignalHandler sh = new SignalHandler() {            public void handle(Signal sig) {                Shutdown.exit(sig.getNumber() + 0200);            }        };

class Shutdown {
    static void exit(int status) {
          ........省略.........
          synchronized (Shutdown.class) {              // 开始 JVM 关闭流程，执行 ShutdownHooks              sequence();              // 强制关闭 JVM              halt(status);          }
    }
    private static void sequence() {        synchronized (lock) {            if (state != HOOKS) return;        }        //触发 ShutdownHooks        runHooks();        boolean rfoe;        synchronized (lock) {            state = FINALIZERS;            rfoe = runFinalizersOnExit;        }        //如果 runFinalizersOnExit = true        //开始运行所有未被调用过的 Finalizers        if (rfoe) runAllFinalizers();    }}

    // The system shutdown hooks are registered with a predefined slot.    // The list of shutdown hooks is as follows:    // (0) Console restore hook    // (1) Application hooks    // (2) DeleteOnExit hook    private static final int MAX_SYSTEM_HOOKS = 10;    private static final Runnable[] hooks = new Runnable[MAX_SYSTEM_HOOKS];
    /* Run all registered shutdown hooks     */    private static void runHooks() {        for (int i=0; i < MAX_SYSTEM_HOOKS; i++) {            try {                Runnable hook;                synchronized (lock) {                    // acquire the lock to make sure the hook registered during                    // shutdown is visible here.                    currentRunningHook = i;                    hook = hooks[i];                }                if (hook != null) hook.run();            } catch(Throwable t) {                if (t instanceof ThreadDeath) {                    ThreadDeath td = (ThreadDeath)t;                    throw td;                }            }        }    }

public class Runtime {
    public void addShutdownHook(Thread hook) {        SecurityManager sm = System.getSecurityManager();        if (sm != null) {            sm.checkPermission(new RuntimePermission("shutdownHooks"));        }        //注意 这里注册的是 Application 类型的 hooks        ApplicationShutdownHooks.add(hook);    }
}

class ApplicationShutdownHooks {    // 存放用户自定义的 Application 类型的 hooks    private static IdentityHashMap<Thread, Thread> hooks;
    static synchronized void add(Thread hook) {        if(hooks == null)            throw new IllegalStateException("Shutdown in progress");
        if (hook.isAlive())            throw new IllegalArgumentException("Hook already running");
        if (hooks.containsKey(hook))            throw new IllegalArgumentException("Hook previously registered");
        hooks.put(hook, hook);    }
    static void runHooks() {        Collection<Thread> threads;        synchronized(ApplicationShutdownHooks.class) {            threads = hooks.keySet();            hooks = null;        }        // 顺序启动 shutdownhooks        for (Thread hook : threads) {            hook.start();        }        // 并发调用 shutdownhooks ，等待所有 hooks 运行完毕退出        for (Thread hook : threads) {            try {                hook.join();            } catch (InterruptedException x) { }        }    }}

class ApplicationShutdownHooks {    /* The set of registered hooks */    private static IdentityHashMap<Thread, Thread> hooks;
    static {        try {            Shutdown.add(1 /* shutdown hook invocation order */,                false /* not registered if shutdown in progress */,                new Runnable() {                    public void run() {                        runHooks();                    }                }            );            hooks = new IdentityHashMap<>();        } catch (IllegalStateException e) {            // application shutdown hooks cannot be added if            // shutdown is in progress.            hooks = null;        }    }}

class Shutdown {
    private static final int MAX_SYSTEM_HOOKS = 10;    private static final Runnable[] hooks = new Runnable[MAX_SYSTEM_HOOKS];
    static void add(int slot, boolean registerShutdownInProgress, Runnable hook) {        synchronized (lock) {            if (hooks[slot] != null)                throw new InternalError("Shutdown hook at slot " + slot + " already registered");
            if (!registerShutdownInProgress) {                if (state > RUNNING)                    throw new IllegalStateException("Shutdown in progress");            } else {                if (state > HOOKS || (state == HOOKS && slot <= currentRunningHook))                    throw new IllegalStateException("Shutdown in progress");            }
            hooks[slot] = hook;        }    }}

@Componentpublic class ShutdownListener implements ApplicationListener<ContextClosedEvent> {       @Override       public void onApplicationEvent(ContextClosedEvent event) {                  ........优雅关闭逻辑.....       }}

@Componentpublic class DestroyBeanPostProcessor implements DestructionAwareBeanPostProcessor {
    @Override    public void postProcessBeforeDestruction(Object bean, String beanName) throws BeansException {
             ........Spring容器中的Bean开始销毁前回调.......    }}

@Componentpublic class Shutdown {    @PreDestroy    public void preDestroy() {        ......释放资源.......    }}

@Componentpublic class Shutdown implements DisposableBean{
    @Override    public void destroy() throws Exception {         ......释放资源......    }
}

<bean id="Shutdown" class="com.test.netty.Shutdown"  destroy-method="doDestroy"/>

public class Shutdown {
    public void doDestroy() {        .....自定义销毁方法....    }}

public abstract class AbstractApplicationContext extends DefaultResourceLoader    implements ConfigurableApplicationContext, DisposableBean {
  @Override  public void registerShutdownHook() {    if (this.shutdownHook == null) {      // No shutdown hook registered yet.      this.shutdownHook = new Thread() {        @Override        public void run() {          synchronized (startupShutdownMonitor) {            doClose();          }        }      };      Runtime.getRuntime().addShutdownHook(this.shutdownHook);    }  }}

public class SpringShutdownHook {
    public static void main(String[] args) throws IOException {        GenericApplicationContext context = new GenericApplicationContext();                      ........        // 注册 Shutdown Hook        context.registerShutdownHook();                      ........    }}

public class SpringApplication {
  public ConfigurableApplicationContext run(String... args) {
                  ...............省略.................
                  ConfigurableApplicationContext context = null;                  context = createApplicationContext();                  refreshContext(context);
                  ...............省略.................  }
  private void refreshContext(ConfigurableApplicationContext context) {    refresh(context);    if (this.registerShutdownHook) {      try {        context.registerShutdownHook();      }      catch (AccessControlException ex) {        // Not allowed in some environments.      }    }  }
}

  protected void doClose() {    // 更新上下文状态    if (this.active.get() && this.closed.compareAndSet(false, true)) {      if (logger.isInfoEnabled()) {        logger.info("Closing " + this);      }            // 取消 JMX 托管      LiveBeansView.unregisterApplicationContext(this);
      try {        // 发布 ContextClosedEvent 事件        publishEvent(new ContextClosedEvent(this));      }      catch (Throwable ex) {        logger.warn("Exception thrown from ApplicationListener handling ContextClosedEvent", ex);      }
      // 回调 Lifecycle beans,相关 stop 方法      if (this.lifecycleProcessor != null) {        try {          this.lifecycleProcessor.onClose();        }        catch (Throwable ex) {          logger.warn("Exception thrown from LifecycleProcessor on context close", ex);        }      }
      // 销毁 bean，触发前面介绍的几种回调      destroyBeans();
      // Close the state of this context itself.      closeBeanFactory();
      // Let subclasses do some final clean-up if they wish...      onClose();
      // Switch to inactive.      this.active.set(false);    }  }

public class DubboBootstrapApplicationListener extends OneTimeExecutionApplicationContextEventListener        implements Ordered {
    @Override    public void onApplicationContextEvent(ApplicationContextEvent event) {        // 这里是 Spring 的同步事件，publishEvent 和处理 Event 是在同一个线程中        if (event instanceof ContextRefreshedEvent) {            onContextRefreshedEvent((ContextRefreshedEvent) event);        } else if (event instanceof ContextClosedEvent) {            onContextClosedEvent((ContextClosedEvent) event);        }    }
    private void onContextClosedEvent(ContextClosedEvent event) {        // spring 在 shutdownhook 中会先触发 ContextClosedEvent ，然后在销毁 spring beans        // 所以这里 dubbo 开始优雅关闭时，依赖的 spring beans 并未销毁        dubboBootstrap.stop();    }
}

public class ServiceClassPostProcessor implements BeanDefinitionRegistryPostProcessor, EnvironmentAware,        ResourceLoaderAware, BeanClassLoaderAware {
    @Override    public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {
        // @since 2.7.5 注册spring启动 关闭事件的listener        //在事件回调中中调用启动类 DubboBootStrap的start  stop来启动 关闭dubbo应用        registerBeans(registry, DubboBootstrapApplicationListener.class);                        ........省略.......    }}

public class DubboBootstrap extends GenericEventListener {
    public DubboBootstrap stop() throws IllegalStateException {        destroy();        return this;    }
}

    public void destroy() {        if (destroyLock.tryLock()) {            try {                DubboShutdownHook.destroyAll();
                if (started.compareAndSet(true, false)                        && destroyed.compareAndSet(false, true)) {
                    //取消注册                    unregisterServiceInstance();                    //取消元数据服务                    unexportMetadataService();                    //停止暴露服务                    unexportServices();                    //取消订阅服务                    unreferServices();                    //注销注册中心                    destroyRegistries();                    //关闭服务                    DubboShutdownHook.destroyProtocols();                    //销毁注册中心客户端实例                    destroyServiceDiscoveries();                    //清除应用配置类以及相关应用模型                    clear();                    //关闭线程池                    shutdown();                    //释放资源                    release();                }            } finally {                destroyLock.unlock();            }        }    }

public class SpringExtensionFactory implements ExtensionFactory {    private static final Logger logger = LoggerFactory.getLogger(SpringExtensionFactory.class);
    private static final Set<ApplicationContext> CONTEXTS = new ConcurrentHashSet<ApplicationContext>();
    public static void addApplicationContext(ApplicationContext context) {        CONTEXTS.add(context);        if (context instanceof ConfigurableApplicationContext) {            //在spring启动成功之后设置shutdownHook（兼容非SpringBoot环境）            ((ConfigurableApplicationContext) context).registerShutdownHook();        }    }
}

public class ServiceBean<T> extends ServiceConfig<T> implements InitializingBean, DisposableBean,        ApplicationContextAware, BeanNameAware, ApplicationEventPublisherAware {
   @Override    public void setApplicationContext(ApplicationContext applicationContext) {        this.applicationContext = applicationContext;        SpringExtensionFactory.addApplicationContext(applicationContext);    }
}

public class ReferenceBean<T> extends ReferenceConfig<T> implements FactoryBean,        ApplicationContextAware, InitializingBean, DisposableBean {
    @Override    public void setApplicationContext(ApplicationContext applicationContext) {        this.applicationContext = applicationContext;        SpringExtensionFactory.addApplicationContext(applicationContext);    }
}

public class DubboBootstrap extends GenericEventListener {
    private DubboBootstrap() {        configManager = ApplicationModel.getConfigManager();        environment = ApplicationModel.getEnvironment();
        DubboShutdownHook.getDubboShutdownHook().register();        ShutdownHookCallbacks.INSTANCE.addCallback(new ShutdownHookCallback() {            @Override            public void callback() throws Throwable {                DubboBootstrap.this.destroy();            }        });    }
}

public class DubboShutdownHook extends Thread {
   public void register() {        if (registered.compareAndSet(false, true)) {            DubboShutdownHook dubboShutdownHook = getDubboShutdownHook();            Runtime.getRuntime().addShutdownHook(dubboShutdownHook);            dispatch(new DubboShutdownHookRegisteredEvent(dubboShutdownHook));        }    }
    @Override    public void run() {        if (logger.isInfoEnabled()) {            logger.info("Run shutdown hook now.");        }
        callback();        doDestroy();    }
   private void callback() {        callbacks.callback();    }
}

public class DubboBootstrap extends GenericEventListener {
    public void destroy() {        if (destroyLock.tryLock()) {            try {                DubboShutdownHook.destroyAll();
                if (started.compareAndSet(true, false)                        && destroyed.compareAndSet(false, true)) {
                    ........取消注册......                                      ........取消元数据服务........                                      ........停止暴露服务........                                     ........取消订阅服务........                                     ........注销注册中心........                                     ........关闭服务........                                      ........销毁注册中心客户端实例........                                     ........清除应用配置类以及相关应用模型........                                    ........关闭线程池........                                     ........释放资源........                                 }            } finally {                destroyLock.unlock();            }        }    }
}

public class DubboBootstrap extends GenericEventListener {
    private final Lock destroyLock = new ReentrantLock();
    public void destroy() {        if (destroyLock.tryLock()) {            try {                DubboShutdownHook.destroyAll();
                if (started.compareAndSet(true, false)                        && destroyed.compareAndSet(false, true)) {                                            .......dubbo应用的优雅关闭.......                                 }            } finally {                destroyLock.unlock();            }        }    }
}

public class SpringExtensionFactory implements ExtensionFactory {    private static final Logger logger = LoggerFactory.getLogger(SpringExtensionFactory.class);
    private static final Set<ApplicationContext> CONTEXTS = new ConcurrentHashSet<ApplicationContext>();
    public static void addApplicationContext(ApplicationContext context) {        CONTEXTS.add(context);        if (context instanceof ConfigurableApplicationContext) {            // 注册 Spring 的 ShutdownHook            ((ConfigurableApplicationContext) context).registerShutdownHook();            // 在 Spring 环境下将 Dubbo 的 ShutdownHook 取消掉            DubboShutdownHook.getDubboShutdownHook().unregister();        }    }}

public class DubboBootstrap {
    private DubboBootstrap() {        configManager = ApplicationModel.getConfigManager();        environment = ApplicationModel.getEnvironment();
        DubboShutdownHook.getDubboShutdownHook().register();        ShutdownHookCallbacks.INSTANCE.addCallback(DubboBootstrap.this::destroy);    }
}

public class DubboBootstrap extends GenericEventListener {
    private final Lock destroyLock = new ReentrantLock();
    public void destroy() {        if (destroyLock.tryLock()) {            try {                DubboShutdownHook.destroyAll();
                if (started.compareAndSet(true, false)                        && destroyed.compareAndSet(false, true)) {                                            .......dubbo应用的优雅关闭.......                    //关闭服务                    DubboShutdownHook.destroyProtocols();
                        .......dubbo应用的优雅关闭.......
                }            } finally {                destroyLock.unlock();            }        }    }
}

public class DubboProtocol extends AbstractProtocol {
   @Override    public void destroy() {        for (String key : new ArrayList<>(serverMap.keySet())) {            ProtocolServer protocolServer = serverMap.remove(key);            RemotingServer server = protocolServer.getRemotingServer();            server.close(ConfigurationUtils.getServerShutdownTimeout());             ...........省略........        }
         ...........省略........}

public class NettyServer extends AbstractServer implements RemotingServer {
    @Override    protected void doClose() throws Throwable {        ..........关闭底层Channel......        try {            if (bootstrap != null) {                // 关闭 Netty 的主从 Reactor 线程组                bossGroup.shutdownGracefully();                workerGroup.shutdownGracefully();            }        } catch (Throwable e) {            logger.warn(e.getMessage(), e);        }        .........清理缓存Channel数据.......    }
}

public abstract class AbstractEventExecutorGroup implements EventExecutorGroup {
    static final long DEFAULT_SHUTDOWN_QUIET_PERIOD = 2;    static final long DEFAULT_SHUTDOWN_TIMEOUT = 15;
   @Override    public Future<?> shutdownGracefully() {        return shutdownGracefully(DEFAULT_SHUTDOWN_QUIET_PERIOD, DEFAULT_SHUTDOWN_TIMEOUT, TimeUnit.SECONDS);    }
}

public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {
    // Reactor线程组中的Reactor集合    private final EventExecutor[] children;
    // 关闭future    private final Promise<?> terminationFuture = new DefaultPromise(GlobalEventExecutor.INSTANCE);
    @Override    public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {        for (EventExecutor l: children) {            l.shutdownGracefully(quietPeriod, timeout, unit);        }        return terminationFuture();    }
    @Override    public Future<?> terminationFuture() {        return terminationFuture;    }
}

    //记录关闭的Reactor个数，当Reactor全部关闭后，ReactorGroup才可以认为关闭成功    private final AtomicInteger terminatedChildren = new AtomicInteger();    //ReactorGroup的关闭future    private final Promise<?> terminationFuture = new DefaultPromise(GlobalEventExecutor.INSTANCE);
    protected MultithreadEventExecutorGroup(int nThreads, Executor executor,                                            EventExecutorChooserFactory chooserFactory, Object... args) {
        ........挨个创建Reactor............
        final FutureListener<Object> terminationListener = new FutureListener<Object>() {            @Override            public void operationComplete(Future<Object> future) throws Exception {                if (terminatedChildren.incrementAndGet() == children.length) {                    //当所有Reactor关闭后 ReactorGroup才认为是关闭成功                    terminationFuture.setSuccess(null);                }            }        };
        for (EventExecutor e: children) {            //向每个Reactor注册terminationListener            e.terminationFuture().addListener(terminationListener);        }    }

public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
    //Reactor的状态  初始为未启动状态    private volatile int state = ST_NOT_STARTED;     //Reactor的初始状态，未启动    private static final int ST_NOT_STARTED = 1;    //Reactor启动后的状态    private static final int ST_STARTED = 2;    //准备正在进行优雅关闭，此时用户仍然可以提交任务，Reactor仍可以执行任务    private static final int ST_SHUTTING_DOWN = 3;    //Reactor停止状态，表示优雅关闭结束，此时用户不能在提交任务，Reactor最后一次执行剩余的任务    private static final int ST_SHUTDOWN = 4;    //Reactor中的任务已被全部执行完毕，且不在接受新的任务，真正的终止状态    private static final int ST_TERMINATED = 5;
    //优雅关闭的静默期    private volatile long gracefulShutdownQuietPeriod;    //优雅关闭超时时间    private volatile long gracefulShutdownTimeout;
    //Reactor的关闭Future    private final Promise<?> terminationFuture = new DefaultPromise<Void>(GlobalEventExecutor.INSTANCE);
    @Override    public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
        ......省略参数校验.......
        //此时Reactor的状态为ST_STARTED        if (isShuttingDown()) {            return terminationFuture();        }
        boolean inEventLoop = inEventLoop();        boolean wakeup;        int oldState;        for (;;) {            if (isShuttingDown()) {                return terminationFuture();            }            int newState;            //需要唤醒Reactor去执行关闭流程            wakeup = true;            oldState = state;            if (inEventLoop) {                newState = ST_SHUTTING_DOWN;            } else {                switch (oldState) {                    case ST_NOT_STARTED:                    case ST_STARTED:                        newState = ST_SHUTTING_DOWN;                        break;                    default:                        //Reactor正在关闭或者已经关闭                        newState = oldState;                        wakeup = false;                }            }            if (STATE_UPDATER.compareAndSet(this, oldState, newState)) {                break;            }        }        //优雅关闭静默期，在该时间内，用户还是可以向Reactor提交任务并且执行，只要有任务在Reactor中，就不能进行关闭        //每隔100ms检测是否有任务提交进来，如果在静默期内没有新的任务提交，那么才会进行关闭 保证关闭行为的优雅        gracefulShutdownQuietPeriod = unit.toNanos(quietPeriod);        //优雅关闭的最大超时时间，优雅关闭行为不能超过该时间，如果超过的话 不管当前是否还有任务 都要进行关闭        //保证关闭行为的可控        gracefulShutdownTimeout = unit.toNanos(timeout);
        //这里需要保证Reactor线程是在运行状态，如果已经停止，那么就不在进行后续关闭行为，直接返回terminationFuture        if (ensureThreadStarted(oldState)) {            return terminationFuture;        }
        //将正在监听IO事件的Reactor从Selector上唤醒，表示要关闭了，开始执行关闭流程        if (wakeup) {            //确保Reactor线程在执行完任务之后 不会在selector上停留            taskQueue.offer(WAKEUP_TASK);            if (!addTaskWakesUp) {                //如果此时Reactor正在Selector上阻塞，则可以确保Reactor被及时唤醒                wakeup(inEventLoop);            }        }
        return terminationFuture();    }
    @Override    public Future<?> terminationFuture() {        return terminationFuture;    }
}

    @Override    public boolean isShuttingDown() {        return state >= ST_SHUTTING_DOWN;    }

            switch (oldState) {                    case ST_NOT_STARTED:                    case ST_STARTED:                        newState = ST_SHUTTING_DOWN;                        break;                    default:                        //Reactor正在关闭或者已经关闭                        newState = oldState;                        //当前Reactor已经处于关闭流程中，则无需在唤醒Reactor了                        wakeup = false;                }

          if (STATE_UPDATER.compareAndSet(this, oldState, newState)) {                break;            }

        //这里需要保证Reactor线程是在运行状态，如果已经停止，        //那么就不在进行后续关闭行为，直接返回terminationFuture        if (ensureThreadStarted(oldState)) {            return terminationFuture;        }

    private boolean ensureThreadStarted(int oldState) {        if (oldState == ST_NOT_STARTED) {            try {                doStartThread();            } catch (Throwable cause) {                STATE_UPDATER.set(this, ST_TERMINATED);                terminationFuture.tryFailure(cause);
                if (!(cause instanceof Exception)) {                    // Also rethrow as it may be an OOME for example                    PlatformDependent.throwException(cause);                }                return true;            }        }        return false;    }

     //将正在监听IO事件的Reactor从Selector上唤醒，表示要关闭了，开始执行关闭流程        if (wakeup) {            //确保Reactor线程在执行完任务之后 不会在selector上停留            taskQueue.offer(WAKEUP_TASK);            if (!addTaskWakesUp) {                //如果此时Reactor正在Selector上阻塞，则可以确保Reactor被及时唤醒                wakeup(inEventLoop);            }        }

public final class NioEventLoop extends SingleThreadEventLoop {    @Override    protected void wakeup(boolean inEventLoop) {        if (!inEventLoop && nextWakeupNanos.getAndSet(AWAKE) != AWAKE) {            selector.wakeup();        }    }}

public final class NioEventLoop extends SingleThreadEventLoop {
    @Override    protected void run() {        for (;;) {            try {                  .......1.监听Channel上的IO事件.......                  .......2.处理Channel上的IO事件.......                  .......3.执行异步任务..........            } finally {                try {                    if (isShuttingDown()) {                        //关闭Reactor上注册的所有Channel,停止处理IO事件，触发unActive以及unRegister事件                        closeAll();                        //注销掉所有Channel停止处理IO事件之后，剩下的就需要执行Reactor中剩余的异步任务了                        if (confirmShutdown()) {                            return;                        }                    }                } catch (Error e) {                    throw (Error) e;                } catch (Throwable t) {                    handleLoopException(t);                }            }        }    }
}

    private void closeAll() {        //这里的目的是清理selector中的一些无效key        selectAgain();        //获取Selector上注册的所有Channel        Set<SelectionKey> keys = selector.keys();        Collection<AbstractNioChannel> channels = new ArrayList<AbstractNioChannel>(keys.size());        for (SelectionKey k: keys) {            //获取NioSocketChannel            Object a = k.attachment();            if (a instanceof AbstractNioChannel) {                channels.add((AbstractNioChannel) a);            } else {                .........省略......            }        }
        for (AbstractNioChannel ch: channels) {            //关闭Reactor上注册的所有Channel，并在pipeline中触发unActive事件和unRegister事件            ch.unsafe().close(ch.unsafe().voidPromise());        }    }

    protected boolean confirmShutdown() {        if (!isShuttingDown()) {            return false;        }
        if (!inEventLoop()) {            throw new IllegalStateException("must be invoked from an event loop");        }
        //取消掉所有的定时任务        cancelScheduledTasks();
        if (gracefulShutdownStartTime == 0) {            //获取优雅关闭开始时间，相对时间            gracefulShutdownStartTime = ScheduledFutureTask.nanoTime();        }
        //这里判断只要有task任务需要执行就不能关闭        if (runAllTasks() || runShutdownHooks()) {            if (isShutdown()) {                // Executor shut down - no new tasks anymore.                return true;            }
            /**             * gracefulShutdownQuietPeriod表示在这段时间内，用户还是可以继续提交异步任务的，Reactor在这段时间内             * 是会保证这些任务被执行到的。             *             * gracefulShutdownQuietPeriod = 0 表示 没有这段静默时期，当前Reactor中的任务执行完毕后，无需等待静默期，执行关闭             * */            if (gracefulShutdownQuietPeriod == 0) {                return true;            }            //避免Reactor在Selector上阻塞，因为此时已经不会再去处理IO事件了，专心处理关闭流程            taskQueue.offer(WAKEUP_TASK);            return false;        }
        //此时Reactor中已经没有任务可执行了，是时候考虑关闭的事情了        final long nanoTime = ScheduledFutureTask.nanoTime();
        //当Reactor中所有的任务执行完毕后，判断是否超过gracefulShutdownTimeout        //如果超过了 则直接关闭        if (isShutdown() || nanoTime - gracefulShutdownStartTime > gracefulShutdownTimeout) {            return true;        }
        //即使现在没有任务也还是不能进行关闭，需要等待一个静默期，在静默期内如果没有新的任务提交，才会进行关闭        //如果在静默期内还有任务继续提交，那么静默期将会重新开始计算，进入一轮新的静默期检测        if (nanoTime - lastExecutionTime <= gracefulShutdownQuietPeriod) {            taskQueue.offer(WAKEUP_TASK);            try {                //gracefulShutdownQuietPeriod内每隔100ms检测一下 是否有任务需要执行                Thread.sleep(100);            } catch (InterruptedException e) {                // Ignore            }
            return false;        }
        // 在整个gracefulShutdownQuietPeriod期间内没有任务需要执行或者静默期结束 则无需等待gracefulShutdownTimeout超时，直接关闭        return true;    }

        NioEventLoop reactor = (NioEventLoop) ctx.channel().eventLoop();        reactor.addShutdownHook(new Runnable() {            @Override            public void run() {                .....关闭逻辑....            }        });

public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
   //可以向Reactor添加shutdownHook，当Reactor关闭的时候会被调用   private final Set<Runnable> shutdownHooks = new LinkedHashSet<Runnable>();
   private boolean runShutdownHooks() {        boolean ran = false;        while (!shutdownHooks.isEmpty()) {            List<Runnable> copy = new ArrayList<Runnable>(shutdownHooks);            shutdownHooks.clear();            for (Runnable task: copy) {                try {                    //Reactor线程挨个顺序同步执行                    task.run();                } catch (Throwable t) {                    logger.warn("Shutdown hook raised an exception.", t);                } finally {                    ran = true;                }            }        }
        if (ran) {            lastExecutionTime = ScheduledFutureTask.nanoTime();        }
        return ran;    }
}

nanoTime - gracefulShutdownStartTime > gracefulShutdownTimeout

public final class NioEventLoop extends SingleThreadEventLoop {
    @Override    protected void run() {        for (;;) {            try {                  .......1.监听Channel上的IO事件.......                  .......2.处理Channel上的IO事件.......                  .......3.执行异步任务..........            } finally {                try {                    if (isShuttingDown()) {                        //关闭Reactor上注册的所有Channel,停止处理IO事件，触发unActive以及unRegister事件                        closeAll();                        //注销掉所有Channel停止处理IO事件之后，剩下的就需要执行Reactor中剩余的异步任务了                        if (confirmShutdown()) {                            return;                        }                    }                } catch (Error e) {                    throw (Error) e;                } catch (Throwable t) {                    handleLoopException(t);                }            }        }    }
}

public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
    private void doStartThread() {        assert thread == null;        executor.execute(new Runnable() {            @Override            public void run() {
                ..........省略.........
                try {                    //Reactor线程开始轮询处理IO事件，执行异步任务                    SingleThreadEventExecutor.this.run();                    //后面的逻辑为用户调用shutdownGracefully关闭Reactor退出循环 走到这里                    success = true;                } catch (Throwable t) {                    logger.warn("Unexpected exception from an event executor: ", t);                } finally {                    //走到这里表示在静默期内已经没有用户在向Reactor提交任务了，或者达到优雅关闭超时时间，开始对Reactor进行关闭                    //如果当前Reactor不是关闭状态则将Reactor的状态设置为ST_SHUTTING_DOWN                    for (;;) {                        int oldState = state;                        if (oldState >= ST_SHUTTING_DOWN || STATE_UPDATER.compareAndSet(                                SingleThreadEventExecutor.this, oldState, ST_SHUTTING_DOWN)) {                            break;                        }                    }
                    try {                        for (;;) {                            //此时Reactor线程虽然已经退出，而此时Reactor的状态为shuttingdown，但任务队列还在                            //用户在此时依然可以提交任务，这里是确保用户在最后的这一刻提交的任务可以得到执行。                            if (confirmShutdown()) {                                break;                            }                        }
                        for (;;) {                            // 当Reactor的状态被更新为SHUTDOWN后，用户提交的任务将会被拒绝                            int oldState = state;                            if (oldState >= ST_SHUTDOWN || STATE_UPDATER.compareAndSet(                                    SingleThreadEventExecutor.this, oldState, ST_SHUTDOWN)) {                                break;                            }                        }
                        // 这里Reactor的状态已经变为SHUTDOWN了，不会在接受用户提交的新任务了                        // 但为了防止用户在状态变为SHUTDOWN之前，也就是Reactor在SHUTTINGDOWN的时候 提交了任务                        // 所以此时Reactor中可能还会有任务，需要将剩余的任务执行完毕                        confirmShutdown();                    } finally {                        try {                            //SHUTDOWN状态下，在将全部的剩余任务执行完毕后，则将Selector关闭                            cleanup();                        } finally {                            // 清理Reactor线程中的threadLocal缓存，并通知相应future。                            FastThreadLocal.removeAll();
                            //ST_TERMINATED状态为Reactor真正的终止状态                            STATE_UPDATER.set(SingleThreadEventExecutor.this, ST_TERMINATED);                                                        //使得awaitTermination方法返回                            threadLock.countDown();
                            //统计一下当前reactor任务队列中还有多少未执行的任务，打出日志                            int numUserTasks = drainTasks();                            if (numUserTasks > 0 && logger.isWarnEnabled()) {                                logger.warn("An event executor terminated with " +                                        "non-empty task queue (" + numUserTasks + ')');                            }
                            /**                             * 通知Reactor的terminationFuture成功，在创建Reactor的时候会向其terminationFuture添加Listener                             * 在listener中增加terminatedChildren个数，当所有Reactor关闭后 ReactorGroup关闭成功                             * */                            terminationFuture.setSuccess(null);                        }                    }                }            }        });    }}

public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {  @Override  public boolean isShutdown() {        return state >= ST_SHUTDOWN;  }
  private void execute(Runnable task, boolean immediate) {        boolean inEventLoop = inEventLoop();        addTask(task);        if (!inEventLoop) {            startThread();            //当Reactor的状态为ST_SHUTDOWN时，拒绝用户提交的异步任务，但是在优雅关闭ST_SHUTTING_DOWN状态时还是可以接受用户提交的任务的            if (isShutdown()) {                boolean reject = false;                try {                    if (removeTask(task)) {                        reject = true;                    }                } catch (UnsupportedOperationException e) {                }                if (reject) {                    reject();                }            }        }
        .........省略........    }}

    @Override    protected void cleanup() {        try {            selector.close();        } catch (IOException e) {            logger.warn("Failed to close a selector.", e);        }    }

FastThreadLocal.removeAll();

 STATE_UPDATER.set(SingleThreadEventExecutor.this, ST_TERMINATED);

public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
    private final CountDownLatch threadLock = new CountDownLatch(1);
    @Override    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {                 ........省略.......
        //等待Reactor关闭        threadLock.await(timeout, unit);        return isTerminated();    }
    @Override    public boolean isTerminated() {        return state == ST_TERMINATED;    }}

public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {
    //Reactor线程组中的Reactor集合    private final EventExecutor[] children;    //记录关闭的Reactor个数，当Reactor全部关闭后，才可以认为关闭成功    private final AtomicInteger terminatedChildren = new AtomicInteger();    //ReactorGroup关闭future    private final Promise<?> terminationFuture = new DefaultPromise(GlobalEventExecutor.INSTANCE);
    protected MultithreadEventExecutorGroup(int nThreads, Executor executor,                                            EventExecutorChooserFactory chooserFactory, Object... args) {              ........挨个创建Reactor........
        final FutureListener<Object> terminationListener = new FutureListener<Object>() {            @Override            public void operationComplete(Future<Object> future) throws Exception {                if (terminatedChildren.incrementAndGet() == children.length) {                    //当所有Reactor关闭后 才认为是关闭成功                    terminationFuture.setSuccess(null);                }            }        };
        for (EventExecutor e: children) {            e.terminationFuture().addListener(terminationListener);        }
        ........省略........    }
}