[Optimized ]Optimized the heartbeat of the cluster (#3119)

DataLinkDC · Jan 31, 2024 · 8fe8baf · 8fe8baf
1 parent 541abf7
commit 8fe8baf
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 6 deletions.
diff --git a/dinky-admin/src/main/java/org/dinky/controller/ClusterInstanceController.java b/dinky-admin/src/main/java/org/dinky/controller/ClusterInstanceController.java
@@ -195,12 +195,8 @@ public Result<List<ClusterInstance>> listSessionEnable() {
     @Log(title = "Cluster Instance Heartbeat", businessType = BusinessType.UPDATE)
     @ApiOperation("Cluster Instance Heartbeat")
     @SaCheckPermission(value = {PermissionConstants.REGISTRATION_CLUSTER_INSTANCE_HEARTBEATS})
-    public Result<Void> heartbeat() {
-        List<ClusterInstance> clusterInstances = clusterInstanceService.list();
-        for (ClusterInstance clusterInstance : clusterInstances) {
-            clusterInstanceService.registersCluster(clusterInstance);
-        }
-        return Result.succeed(Status.CLUSTER_INSTANCE_HEARTBEAT_SUCCESS);
+    public Result<Long> heartbeat() {
+        return Result.succeed(clusterInstanceService.heartbeat(), Status.CLUSTER_INSTANCE_HEARTBEAT_SUCCESS);
     }
 
     /**

diff --git a/dinky-admin/src/main/java/org/dinky/service/ClusterInstanceService.java b/dinky-admin/src/main/java/org/dinky/service/ClusterInstanceService.java
@@ -138,4 +138,10 @@ public interface ClusterInstanceService extends ISuperService<ClusterInstance> {
      * @return {@link Boolean} true: has relationship, false: no relationship
      */
     boolean hasRelationShip(Integer id);
+
+    /**
+     * heartbeat
+     * @return {@link Long}
+     */
+    Long heartbeat();
 }
diff --git a/dinky-admin/src/main/java/org/dinky/service/impl/ClusterInstanceServiceImpl.java b/dinky-admin/src/main/java/org/dinky/service/impl/ClusterInstanceServiceImpl.java
@@ -48,6 +48,9 @@
 import java.time.LocalDateTime;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Lazy;
@@ -57,6 +60,7 @@
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 
+import cn.hutool.core.thread.ThreadUtil;
 import cn.hutool.core.util.StrUtil;
 import lombok.RequiredArgsConstructor;
 
@@ -275,6 +279,17 @@ public boolean hasRelationShip(Integer id) {
                 .isEmpty();
     }
 
+    @Override
+    public Long heartbeat() {
+        List<ClusterInstance> clusterInstances = this.list();
+        ExecutorService executor = ThreadUtil.newExecutor(Math.min(clusterInstances.size(), 10));
+        List<CompletableFuture<Integer>> futures = clusterInstances.stream()
+                .map(c -> CompletableFuture.supplyAsync(
+                        () -> this.registersCluster(c).getStatus(), executor))
+                .collect(Collectors.toList());
+        return futures.stream().map(CompletableFuture::join).filter(x -> x == 1).count();
+    }
+
     private boolean checkHealth(ClusterInstance clusterInstance) {
         FlinkClusterInfo info = checkHeartBeat(clusterInstance.getHosts(), clusterInstance.getJobManagerHost());
         if (!info.isEffective()) {