1. import lombok.experimental.UtilityClass;
    2. import lombok.extern.slf4j.Slf4j;
    4. import java.util.Map;
    5. import java.util.concurrent.ConcurrentHashMap;
    6. import java.util.concurrent.ScheduledThreadPoolExecutor;
    7. import java.util.concurrent.TimeUnit;
    8. import java.util.function.Supplier;
    10. /**
    11. * 心跳管理器
    12. * <p>
    13. *     该类通常与类{@link SwitchBarrier}一起使用，心跳管理器管理业务的运行状态，而类{@link SwitchBarrier}管理对远端业务的访问
    14. * </p>
    15. * @author shizi
    16. * @since 2020-11-24 15:16:35
    17. */
    18. @Slf4j
    19. @UtilityClass
    20. public class HeartBeanManager {
    22.     /**
    23.     * 心跳守护线程池
    24.     */
    25.     private final ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), r -> {
    26.         Thread thread = new Thread(r, "heart-daemon");
    27.         thread.setDaemon(true);
    28.         return thread;
    29.     });
    30.     /**
    31.     * 心跳异常时候的阈值打印
    32.     */
    33.     private final Integer HEARD_BEAN_PRINT_THRESHOLD = 6;
    34.     /**
    35.     * 打印日志频率阈值
    36.     */
    37.     private Integer printLogThresholdNum = 0;
    38.     /**
    39.     * 业务的心跳任务
    40.     */
    41.     private final Map<String, Pair<Supplier<Boolean>, Boolean>> heartTaskMap = new ConcurrentHashMap<>();
    43.     static {
    44.         scheduler.scheduleWithFixedDelay(HeartBeanManager::heartBeat, 5, 5, TimeUnit.SECONDS);
    45.     }
    47.     private static void heartBeat() {
    48.         heartTaskMap.forEach((k, v) -> {
    49.             try {
    50.                 v.getKey().get();
    51.                 serverRestore(k, v.getValue());
    52.                 v.setValue(true);
    53.             } catch (Throwable e) {
    54.                 serverUnAvailable(k);
    55.                 v.setValue(false);
    56.             }
    57.         });
    58.     }
    60.     /**
    61.     * 添加业务的心跳判断
    62.     *
    63.     * @param bizName         业务名
    64.     * @param healthCheck   业务的心跳检测url
    65.     */
    66.     public void addHeartWatch(String bizName, Supplier<Boolean> healthCheck) {
    67.         heartTaskMap.putIfAbsent(bizName, new Pair<>(healthCheck, true));
    68.     }
    70.     public void addHeartWatchAndRun(String bizName, Supplier<Boolean> healthCheck) {
    71.         heartTaskMap.putIfAbsent(bizName, new Pair<>(healthCheck, true));
    72.         handleRun(bizName);
    73.     }
    75.     /**
    76.     * 手动运行
    77.     * @param bizName 业务名
    78.     */
    79.     public void handleRun(String bizName) {
    80.         if (!heartTaskMap.containsKey(bizName)) {
    81.             return;
    82.         }
    83.         Pair<Supplier<Boolean>, Boolean> healthPair = heartTaskMap.get(bizName);
    84.         try {
    85.             healthPair.getKey().get();
    86.             serverRestore(bizName, healthPair.getValue());
    87.             healthPair.setValue(true);
    88.         } catch (Throwable e) {
    89.             serverUnAvailable(bizName);
    90.             healthPair.setValue(false);
    91.         }
    92.     }
    94.     /**
    95.     * 业务是否可用
    96.     *
    97.     * @param bizName 业务名
    98.     * @return true：业务可用，false：业务不可用
    99.     */
    100.     public boolean isHealth(String bizName) {
    101.         return SwitchBarrier.canCross(bizName);
    102.     }
    104.     private void serverRestore(String bizName, Boolean serverAvailable) {
    105.         if (!serverAvailable) {
    106.             log.info("服务【{}】心跳恢复", bizName);
    107.         }
    108.         printLogThresholdNum = 0;
    109.         SwitchBarrier.allowCross(bizName);
    110.     }
    112.     private void serverUnAvailable(String bizName) {
    113.         if (printLogThresholdNum <= 0) {
    114.             log.error("服务【{}】心跳异常，url:{}", bizName, heartTaskMap.get(bizName).getKey());
    115.             printLogThresholdNum = HEARD_BEAN_PRINT_THRESHOLD;
    116.         } else {
    117.             printLogThresholdNum--;
    118.         }
    119.         SwitchBarrier.forbidCross(bizName);
    120.     }
    121. }

    用法

    1. // 启动健康检查任务
    2. HeartBeanManager.addHeartWatchAndRun(CLOUD_UPLOAD_JOB, () -> {
    3.     String remoteUrl = cloudRemoteAddress();
    4.     if (StringUtils.isEmpty(remoteUrl)) {
    5.         log.info("异常：地址为空");
    6.         return false;
    7.     }
    9.     NeoMap resultMap = HttpHelper.get(NeoMap.class, remoteUrl + CLOUD_HEALTH_CHECK);
    10.     if (resultMap.containsKey("ok")) {
    11.         return resultMap.getBoolean("ok");
    12.     }
    13.     return false;
    14. });