优雅关闭,包括两部分,一个是 RPC 框架作为客户端,一个是 RPC 框架作为服务端。

作为服务端

作为服务端的时候,RPC 框架在关闭时,不应该直接暴力关闭。在 RPC 框架中

  1. com.alipay.sofa.rpc.context.RpcRuntimeContext

在静态初始化块中,添加了一个 ShutdownHook

  1. // 增加jvm关闭事件
  2. if (RpcConfigs.getOrDefaultValue(RpcOptions.JVM_SHUTDOWN_HOOK, true)) {
  3. Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
  4. @Override
  5. public void run() {
  6. if (LOGGER.isWarnEnabled()) {
  7. LOGGER.warn("SOFA RPC Framework catch JVM shutdown event, Run shutdown hook now.");
  8. }
  9. destroy(false);
  10. }
  11. }, "SOFA-RPC-ShutdownHook"));
  12. }

这个 ShutdownHook 的作用是当发布平台/用户执行 kill pid 的时候,会先执行 ShutdownHook 中的逻辑。在销毁操作中,RPC 框架会先执行向注册中心取消服务注册、关闭服务端口等动作。

  1. private static void destroy(boolean active) {
  2. RpcRunningState.setShuttingDown(true);
  3. for (Destroyable.DestroyHook destroyHook : DESTROY_HOOKS) {
  4. destroyHook.preDestroy();
  5. }
  6. List<ProviderConfig> providerConfigs = new ArrayList<ProviderConfig>();
  7. for (ProviderBootstrap bootstrap : EXPORTED_PROVIDER_CONFIGS) {
  8. providerConfigs.add(bootstrap.getProviderConfig());
  9. }
  10. // 先反注册服务端
  11. List<Registry> registries = RegistryFactory.getRegistries();
  12. if (CommonUtils.isNotEmpty(registries) && CommonUtils.isNotEmpty(providerConfigs)) {
  13. for (Registry registry : registries) {
  14. registry.batchUnRegister(providerConfigs);
  15. }
  16. }
  17. // 关闭启动的端口
  18. ServerFactory.destroyAll();
  19. // 关闭发布的服务
  20. for (ProviderBootstrap bootstrap : EXPORTED_PROVIDER_CONFIGS) {
  21. bootstrap.unExport();
  22. }
  23. // 关闭调用的服务
  24. for (ConsumerBootstrap bootstrap : REFERRED_CONSUMER_CONFIGS) {
  25. ConsumerConfig config = bootstrap.getConsumerConfig();
  26. if (!CommonUtils.isFalse(config.getParameter(RpcConstants.HIDDEN_KEY_DESTROY))) { // 除非不让主动unrefer
  27. bootstrap.unRefer();
  28. }
  29. }
  30. // 关闭注册中心
  31. RegistryFactory.destroyAll();
  32. // 关闭客户端的一些公共资源
  33. ClientTransportFactory.closeAll();
  34. // 卸载模块
  35. if (!RpcRunningState.isUnitTestMode()) {
  36. ModuleFactory.uninstallModules();
  37. }
  38. // 卸载钩子
  39. for (Destroyable.DestroyHook destroyHook : DESTROY_HOOKS) {
  40. destroyHook.postDestroy();
  41. }
  42. // 清理缓存
  43. RpcCacheManager.clearAll();
  44. RpcRunningState.setShuttingDown(false);
  45. if (LOGGER.isWarnEnabled()) {
  46. LOGGER.warn("SOFA RPC Framework has been release all resources {}...",
  47. active ? "actively " : "");
  48. }
  49. }

其中以 bolt 为例,关闭端口并不是一个立刻执行的动作

  1. @Override
  2. public void destroy() {
  3. if (!started) {
  4. return;
  5. }
  6. int stopTimeout = serverConfig.getStopTimeout();
  7. if (stopTimeout > 0) { // 需要等待结束时间
  8. AtomicInteger count = boltServerProcessor.processingCount;
  9. // 有正在执行的请求 或者 队列里有请求
  10. if (count.get() > 0 || bizThreadPool.getQueue().size() > 0) {
  11. long start = RpcRuntimeContext.now();
  12. if (LOGGER.isInfoEnabled()) {
  13. LOGGER.info("There are {} call in processing and {} call in queue, wait {} ms to end",
  14. count, bizThreadPool.getQueue().size(), stopTimeout);
  15. }
  16. while ((count.get() > 0 || bizThreadPool.getQueue().size() > 0)
  17. && RpcRuntimeContext.now() - start < stopTimeout) { // 等待返回结果
  18. try {
  19. Thread.sleep(10);
  20. } catch (InterruptedException ignore) {
  21. }
  22. }
  23. } // 关闭前检查已有请求?
  24. }
  25. // 关闭线程池
  26. bizThreadPool.shutdown();
  27. stop();
  28. }

而是会判断当前服务端上面的连接和队列的任务,先处理完队列中的任务,再缓慢关闭。

作为客户端

作为客户端。实际上就是 Cluster 的关闭,关闭调用的服务这一步,可以查看下

  1. com.alipay.sofa.rpc.client.AbstractCluster
  1. /**
  2. * 优雅关闭的钩子
  3. */
  4. protected class GracefulDestroyHook implements DestroyHook {
  5. @Override
  6. public void preDestroy() {
  7. // 准备关闭连接
  8. int count = countOfInvoke.get();
  9. final int timeout = consumerConfig.getDisconnectTimeout(); // 等待结果超时时间
  10. if (count > 0) { // 有正在调用的请求
  11. long start = RpcRuntimeContext.now();
  12. if (LOGGER.isWarnEnabled()) {
  13. LOGGER.warn("There are {} outstanding call in client, will close transports util return",
  14. count);
  15. }
  16. while (countOfInvoke.get() > 0 && RpcRuntimeContext.now() - start < timeout) { // 等待返回结果
  17. try {
  18. Thread.sleep(10);
  19. } catch (InterruptedException ignore) {
  20. }
  21. }
  22. }
  23. }
  24. @Override
  25. public void postDestroy() {
  26. }
  27. }

这里面也会逐步将正在调用的请求处理完成才会下线。

最佳实践

可以看到,优雅关闭是需要和发布平台联动的。如果强制 kill,那么任何优雅关闭的方案都不会生效。后续我们会考虑在 SOFABoot 层面提供一个统一的 API, 来给发布平台调用。而不是依赖 hook 的逻辑。