Обмен технологиями

한어Русский языкEnglishFrançaisIndonesianSanskrit日本語DeutschPortuguêsΕλληνικάespañolItalianoSuomalainenLatina

Когда машина отключается от сети, она сталкивается со многими проблемами: как отключить ее от центра регистрации? Как очистить и освободить ресурсы? Клиент также использует локальный кеш при извлечении списка сервисов. Как вовремя обновить локальный кеш?

Для корректного завершения работы серверной машины требуется использование ShutdownHook, что эквивалентно добавлению перехватчика выключения. Этот перехват представляет собой поток, который вызывается при завершении работы JVM (то есть, когда программа завершает работу) для очистки ресурсов и очистки. закрыть изящно.

   public void clearAll() {
        log.info("addShutdownHook for clearAll");
        // 添加了一个关闭钩子，这个钩子是一个线程，它在JVM关闭时（即程序结束时）被调用，清理资源，优雅下机
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                InetSocketAddress inetSocketAddress = new InetSocketAddress(InetAddress.getLocalHost().getHostAddress(), NettyRpcServer.PORT);
                CuratorUtils.clearRegistry(CuratorUtils.getZkClient(), inetSocketAddress);
            } catch (UnknownHostException ignored) {
            }
            // 操作完整、优雅，便于释放连接资源，便于自定义清理逻辑
            ThreadPoolFactoryUtil.shutDownAllThreadPool();
        }));
    }

В потоке-хуке нужно написать логику удаления узлов из центра регистрации, как показано ниже:

  // RPC Server端 本机所注册服务的缓存
  private static final Set<String> REGISTERED_PATH_SET = ConcurrentHashMap.newKeySet();
 
  public static void clearRegistry(CuratorFramework zkClient, InetSocketAddress inetSocketAddress) {
        REGISTERED_PATH_SET.stream().parallel().forEach(p -> {
            try {
                // 是本机在ZK注册的节点
                if (p.endsWith(inetSocketAddress.toString())) {
                    // 根据路径名删除节点
                    zkClient.delete().forPath(p);
                }
            } catch (Exception e) {
                log.error("clear registry for path [{}] fail", p);
            }
        });
        log.info("All registered services on the server are cleared:[{}]", REGISTERED_PATH_SET.toString());
    }

После удаления узла в центре регистрации ЗК необходимо освободить ресурсы пула потоков:

  public static void shutDownAllThreadPool() {
        log.info("call shutDownAllThreadPool method");
        THREAD_POOLS.entrySet().parallelStream().forEach(entry -> {
            ExecutorService executorService = entry.getValue();
            // 停止接收新的任务，但已提交的任务会继续执行
            executorService.shutdown();
            log.info("shut down thread pool [{}] [{}]", entry.getKey(), executorService.isTerminated());
            try {
                // 等待线程池中的任务在指定的时间内完成。如果在指定时间内线程池未能终止，会抛出 InterruptedException
                executorService.awaitTermination(10, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                log.error("Thread pool never terminated");
                // 指定时间内线程池未能终止，立即停止所有正在执行的任务
                executorService.shutdownNow();
            }
        });
    }

Настройка логики закрытия пула потоков позволяет более элегантно реализовать освобождение ресурсов пула потоков. Вы можете прекратить прием новых задач и продолжить выполнение отправленных задач в течение определенного времени ожидания. Если время ожидания превышено, пул потоков принудительно завершается.