Spring Boot 基于 HTTP 请求实现心跳

╰半橙微兮° 2023-10-05 19:02 25阅读 0赞

### 写在最前 ###

最近项目中有个要求是写个心跳的功能，判断某个服务是否还可用。先用 socket 写了个心跳，但是不太符合要求（因为 socket 和应用不在一个端口上），想了解的可以去 socket 实现心跳看下简单的 demo。

### 解决方案 ###

> 大概逻辑：A 是用来派发任务的，B 是用来接收任务并执行的，如果 A 给 B 发请求，没收到 B 的请求则判断 B 下线。

专门在某个服务 B 中提供个接口用来给被请求：

@RestController
    public class HeartbeatController {
        
        private static final Logger log = LoggerFactory.getLogger(HeartbeatController.class);
        @RequestMapping(value = "/heartbeat", method = RequestMethod.GET)
        public String heartbeat() throws InterruptedException {
        
            // Thread.sleep(30000);
            log.info("B 收到 A 的请求");
            // 返回结果
            return "B 收到请求";
        }
    }

在 A 服务中添加定时任务：

/**
     * 心跳定时任务
     */
    @Component
    public class HeartbeatSchedule {
        
        private static final Logger log = LoggerFactory.getLogger(HeartbeatSchedule.class);
        ExecutorService executorService = Executors.newFixedThreadPool(20);
    
    
        /**
         * 向各服务发 GET 请求，服务启动后 10 秒执行，之后每单个任务「执行完」后 5 秒执行一次
         *
         * @throws InterruptedException
         */
        @Scheduled(initialDelay = 10000, fixedDelay = 5000)
        public void sendGetRequest() throws InterruptedException {
        
    
            for (JobType jobType : JobType.values()) {
        
                JobTypeDetail jobTypeDetail = JobUtil.getJobTypeDetail(jobType.code());
                heartBeatMultithread(jobTypeDetail);
            }
        }
    
        /**
         * 心跳方法多线程版
         */
        public void heartBeatMultithread(JobTypeDetail jobTypeDetail) throws InterruptedException {
        
    
            ConcurrentHashMap<String, Set<String>> allProgress = jobTypeDetail.getAllProgress();
            ConcurrentHashMap<String, Set<String>> aliveProgress = jobTypeDetail.getProgress();
            boolean flag;
            for (String progressUri : allProgress.keySet()) {
        
    
                while (true) {
        
                    executorService.execute(new Runnable() {
        
                        @Override
                        public void run() {
        
                            String heartbeatResponse = HttpUtil.getHeartbeat(progressUri + "/heartbeat");
                            if (heartbeatResponse == null) {
        
                                log.info("{}当前失联", progressUri);
                                if (aliveProgress.containsKey(progressUri)) {
        
                                    aliveProgress.remove(progressUri);
                                    log.info("之前在线的{}当前失联", progressUri);
                                }
                            } else {
        
                                if (!aliveProgress.containsKey(progressUri)) {
        
                                    aliveProgress.put(progressUri, new HashSet<>());
                                    log.info("{}当前存活，已被放入 aliveProgress", progressUri);
                                }
                                log.info("{}当前存活，之前已保存在 aliveProgress 中", progressUri);
                            }
                        }
                    });
                    break;
                }
    
            }
        }
    
        /**
         * 心跳方法单线程版
         */
        public void heartBeatSingleThread(JobTypeDetail jobTypeDetail) throws InterruptedException {
        
    
            ConcurrentHashMap<String, Set<String>> allProgress = jobTypeDetail.getAllProgress();
            ConcurrentHashMap<String, Set<String>> aliveProgress = jobTypeDetail.getProgress();
    
            boolean flag;
            for (String progressUri : allProgress.keySet()) {
        
                flag = true;
    
                while (flag) {
        
                    String heartbeatResponse = HttpUtil.getHeartbeat(progressUri + "/heartbeat");
                    if (heartbeatResponse == null) {
        
                        flag = false;
                        log.info("{}当前失联", progressUri);
                        if (aliveProgress.contains(progressUri)) {
        
                            aliveProgress.remove(progressUri);
                            log.info("之前在线的{}当前失联", progressUri);
                        }
                    } else {
        
                        if (!aliveProgress.containsKey(progressUri)) {
        
                            aliveProgress.put(progressUri, new HashSet<>());
                            log.info("{}当前存活，已被放入 aliveProgress", progressUri);
                        }
                        log.info("{}当前存活，之前已保存在 aliveProgress 中", progressUri);
                    }
                    break;
                }
            }
        }
    }

用到的 `HttpUtil#getHeartbeat()` 参考 [HttpClient 发送 HTTP 请求工具类][HttpClient _ HTTP]

### 写在最后 ###

整体的逻辑就是通过执行定时任务实现心跳中的频率，A 是否收到 B 的回复（返回值）来判断 B 的状态，核心代码：

// 响应结果
    String heartbeatResponse = HttpUtil.getHeartbeat(progressUri + "/heartbeat");
    // 没收到响应
    if (heartbeatResponse == null) {
        
        log.info("{}当前失联", progressUri);
        if (aliveProgress.containsKey(progressUri)) {
        
            aliveProgress.remove(progressUri);
            log.info("之前在线的{}当前失联", progressUri);
        }
    } else {
        
        if (!aliveProgress.containsKey(progressUri)) {
        
            aliveProgress.put(progressUri, new HashSet<>());
            log.info("{}当前存活，已被放入 aliveProgress", progressUri);
        }
        log.info("{}当前存活，之前已保存在 aliveProgress 中", progressUri);
    }

以上代码并不是用在项目中的最终版，有一些地方需要注意下：

1.  用到了多线程，创建的线程池方式不对，读者需要根据实际情况创建；
2.  `HttpUtil#getHeartbeat()` 中设置的超时时间需要根据实际情况设定；

其他的实体类信息空了再完善到博客里。

[HttpClient _ HTTP]: https://mindartisan.blog.csdn.net/article/details/113276631