[ADD] add flink node

isacc · isacc · commit ea9018c88e95 · 2020-04-02T22:56:35.000+08:00
diff --git a/pom.xml b/pom.xml
@@ -143,6 +143,11 @@
             <artifactId>mapstruct</artifactId>
             <version>${org.mapstruct.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>transmittable-thread-local</artifactId>
+            <version>2.11.0</version>
+        </dependency>
 
     </dependencies>
 
diff --git a/src/main/java/com/github/codingdebugallday/client/api/dto/ClusterDTO.java b/src/main/java/com/github/codingdebugallday/client/api/dto/ClusterDTO.java
@@ -29,7 +29,7 @@ public class ClusterDTO implements Serializable {
 
     private static final long serialVersionUID = 854464206375410197L;
 
-    public static final String FIELD_CLUSTER_ID = "clusterId";
+    public static final String FIELD_CLUSTER_ID = "cluster_id";
 
     private Long clusterId;
 
diff --git a/src/main/java/com/github/codingdebugallday/client/api/dto/NodeDTO.java b/src/main/java/com/github/codingdebugallday/client/api/dto/NodeDTO.java
@@ -37,6 +37,7 @@ public class NodeDTO implements Serializable {
 
     /**
      * {
+     *     "host": "xxx"
      *     "username": "xxx",
      *     "password": "xxx"
      * }
diff --git a/src/main/java/com/github/codingdebugallday/client/api/dto/NodeSettingInfo.java b/src/main/java/com/github/codingdebugallday/client/api/dto/NodeSettingInfo.java
@@ -19,6 +19,7 @@
 @JsonInclude(JsonInclude.Include.NON_NULL)
 public class NodeSettingInfo {
 
+    private String host;
     private String username;
     private String password;
 }
diff --git a/src/main/java/com/github/codingdebugallday/client/domain/entity/Node.java b/src/main/java/com/github/codingdebugallday/client/domain/entity/Node.java
@@ -30,7 +30,9 @@ public class Node implements Serializable {
 
     private static final long serialVersionUID = 854464206375410197L;
 
-    public static final String FIELD_NODE_ID = "nodeId";
+    public static final String FIELD_NODE_ID = "node_id";
+    public static final String FIELD_CLUSTER_CODE = "cluster_code";
+    public static final String FIELD_TENANT_ID = "tenant_id";
 
     @TableId(type = IdType.AUTO)
     private Long nodeId;
diff --git a/src/main/java/com/github/codingdebugallday/client/domain/repository/ClusterRepository.java b/src/main/java/com/github/codingdebugallday/client/domain/repository/ClusterRepository.java
@@ -44,7 +44,8 @@ public interface ClusterRepository {
      * 根据clusterCode获取唯一条
      *
      * @param clusterCode clusterCode
+     * @param tenantId    租户id
      * @return org.abigballofmud.flink.client.api.dto.ClusterDTO
      */
-    ClusterDTO selectOne(String clusterCode);
+    ClusterDTO selectOne(String clusterCode, Long tenantId);
 }
diff --git a/src/main/java/com/github/codingdebugallday/client/domain/repository/NodeRepository.java b/src/main/java/com/github/codingdebugallday/client/domain/repository/NodeRepository.java
@@ -0,0 +1,25 @@
+package com.github.codingdebugallday.client.domain.repository;
+
+import java.util.List;
+
+import com.github.codingdebugallday.client.api.dto.NodeDTO;
+
+/**
+ * <p>
+ * description
+ * </p>
+ *
+ * @author isacc 2020/03/28 2:20
+ * @since 1.0
+ */
+public interface NodeRepository {
+
+    /**
+     * 根据clusterCode获取flink节点集合
+     *
+     * @param clusterCode clusterCode
+     * @param tenantId    租户id
+     * @return org.abigballofmud.flink.client.api.dto.ClusterDTO
+     */
+    List<NodeDTO> selectByClusterCode(String clusterCode, Long tenantId);
+}
diff --git a/src/main/java/com/github/codingdebugallday/client/infra/autoconfigure/FlinkApiAutoConfiguration.java b/src/main/java/com/github/codingdebugallday/client/infra/autoconfigure/FlinkApiAutoConfiguration.java
@@ -1,8 +1,9 @@
 package com.github.codingdebugallday.client.infra.autoconfigure;
 
 import com.github.codingdebugallday.client.domain.repository.ClusterRepository;
-import com.github.codingdebugallday.client.infra.exceptions.RestTemplateErrorHandler;
+import com.github.codingdebugallday.client.domain.repository.NodeRepository;
 import com.github.codingdebugallday.client.infra.context.FlinkApiContext;
+import com.github.codingdebugallday.client.infra.exceptions.RestTemplateErrorHandler;
 import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
@@ -27,8 +28,10 @@
 public class FlinkApiAutoConfiguration {
 
     @Bean
-    public FlinkApiContext flinkApiContext(RestTemplate restTemplate, ClusterRepository clusterRepository) {
-        return new FlinkApiContext(restTemplate, clusterRepository);
+    public FlinkApiContext flinkApiContext(RestTemplate restTemplate,
+                                           ClusterRepository clusterRepository,
+                                           NodeRepository nodeRepository) {
+        return new FlinkApiContext(restTemplate, clusterRepository, nodeRepository);
     }
 
     @Bean("flinkRestTemplate")
diff --git a/src/main/java/com/github/codingdebugallday/client/infra/context/FlinkApiContext.java b/src/main/java/com/github/codingdebugallday/client/infra/context/FlinkApiContext.java
@@ -4,8 +4,10 @@
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 
-import com.github.codingdebugallday.client.domain.repository.ClusterRepository;
+import com.github.codingdebugallday.client.api.dto.ClusterDTO;
 import com.github.codingdebugallday.client.app.service.FlinkApi;
+import com.github.codingdebugallday.client.domain.repository.ClusterRepository;
+import com.github.codingdebugallday.client.domain.repository.NodeRepository;
 import org.springframework.web.client.RestTemplate;
 
 /**
@@ -23,17 +25,22 @@ public class FlinkApiContext {
     private final RestTemplate flinkRestTemplate;
 
     private final ClusterRepository clusterRepository;
+    private final NodeRepository nodeRepository;
 
     public FlinkApiContext(RestTemplate flinkRestTemplate,
-                           ClusterRepository clusterRepository) {
+                           ClusterRepository clusterRepository,
+                           NodeRepository nodeRepository) {
         this.flinkRestTemplate = flinkRestTemplate;
         this.clusterRepository = clusterRepository;
+        this.nodeRepository = nodeRepository;
     }
 
-    public FlinkApi get(String clusterCode) {
+    public FlinkApi get(String clusterCode, Long tenantId) {
         if (Objects.isNull(flinkApiMap.get(clusterCode))) {
             FlinkApi flinkApi = new FlinkApi(flinkRestTemplate);
-            flinkApi.getApiClient().setClusterDTO(clusterRepository.selectOne(clusterCode));
+            ClusterDTO clusterDTO = clusterRepository.selectOne(clusterCode, tenantId);
+            clusterDTO.setNodeDTOList(nodeRepository.selectByClusterCode(clusterCode, tenantId));
+            flinkApi.getApiClient().setClusterDTO(clusterDTO);
             flinkApiMap.put(clusterCode, flinkApi);
             return flinkApi;
         }
diff --git a/src/main/java/com/github/codingdebugallday/client/infra/repository/impl/ClusterRepositoryImpl.java b/src/main/java/com/github/codingdebugallday/client/infra/repository/impl/ClusterRepositoryImpl.java
@@ -55,9 +55,10 @@ public ClusterDTO detail(Long tenantId, Long clusterId) {
     }
 
     @Override
-    public ClusterDTO selectOne(String clusterCode) {
+    public ClusterDTO selectOne(String clusterCode,Long tenantId) {
         QueryWrapper<Cluster> queryWrapper = new QueryWrapper<>();
         queryWrapper.eq(Cluster.FIELD_CLUSTER_CODE, clusterCode);
+        queryWrapper.eq(Cluster.FIELD_TENANT_ID, tenantId);
         Cluster cluster = clusterMapper.selectOne(queryWrapper);
         return ClusterConvertMapper.INSTANCE.entityToDTO(cluster);
     }
diff --git a/src/main/java/com/github/codingdebugallday/client/infra/repository/impl/NodeRepositoryImpl.java b/src/main/java/com/github/codingdebugallday/client/infra/repository/impl/NodeRepositoryImpl.java
@@ -0,0 +1,41 @@
+package com.github.codingdebugallday.client.infra.repository.impl;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.github.codingdebugallday.client.api.dto.NodeDTO;
+import com.github.codingdebugallday.client.domain.entity.Node;
+import com.github.codingdebugallday.client.domain.repository.NodeRepository;
+import com.github.codingdebugallday.client.infra.converter.NodeConvertMapper;
+import com.github.codingdebugallday.client.infra.mapper.NodeMapper;
+import org.springframework.stereotype.Component;
+
+/**
+ * <p>
+ * description
+ * </p>
+ *
+ * @author isacc 2020/03/28 2:24
+ * @since 1.0
+ */
+@Component
+public class NodeRepositoryImpl implements NodeRepository {
+
+    private final NodeMapper nodeMapper;
+
+    public NodeRepositoryImpl(NodeMapper nodeMapper) {
+        this.nodeMapper = nodeMapper;
+    }
+
+    @Override
+    public List<NodeDTO> selectByClusterCode(String clusterCode, Long tenantId) {
+        QueryWrapper<Node> queryWrapper = new QueryWrapper<>();
+        queryWrapper.eq(Node.FIELD_CLUSTER_CODE, clusterCode);
+        queryWrapper.eq(Node.FIELD_TENANT_ID, tenantId);
+        return nodeMapper.selectList(queryWrapper).stream()
+                .map(NodeConvertMapper.INSTANCE::entityToDTO)
+                .collect(Collectors.toList());
+    }
+
+}
diff --git a/src/main/java/com/github/codingdebugallday/client/infra/utils/ThreadPoolUtil.java b/src/main/java/com/github/codingdebugallday/client/infra/utils/ThreadPoolUtil.java
@@ -0,0 +1,59 @@
+package com.github.codingdebugallday.client.infra.utils;
+
+import java.util.Objects;
+import java.util.concurrent.*;
+
+import com.alibaba.ttl.threadpool.TtlExecutors;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+/**
+ * <p>
+ * 线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。 说明：Executors各个方法的弊端：
+ * 1）newFixedThreadPool和newSingleThreadExecutor:
+ *   主要问题是堆积的请求处理队列可能会耗费非常大的内存，甚至OOM。
+ * 2）newCachedThreadPool和newScheduledThreadPool:
+ *   主要问题是线程数最大数是Integer.MAX_VALUE，可能会创建数量非常多的线程，甚至OOM。
+ * </p>
+ *
+ * @author abigballofmud 2019/11/21 17:37
+ * @since 1.0
+ */
+public class ThreadPoolUtil {
+
+    private static volatile ExecutorService executorService;
+
+    private ThreadPoolUtil() {
+        throw new IllegalStateException("Utility class");
+    }
+
+    /**
+     * guava的方式
+     */
+    public static ExecutorService getExecutorService() {
+        if (Objects.isNull(executorService)) {
+            synchronized (ThreadPoolUtil.class) {
+                /*
+                 * 使用谷歌的guava框架
+                 * ThreadPoolExecutor参数解释
+                 *   1.corePoolSize 常驻核心线程池大小
+                 *   2.maximumPoolSize 线程池最大线程数 必须>=1
+                 *   3.keepAliveTime 多余的空闲线程存活时间
+                 *   4.TimeUnit 时间单位
+                 *   5.BlockingQueue 任务队列
+                 *   6.ThreadFactory 线程工厂用于创建线程
+                 *   7.RejectedExecutionHandler 队列满了并且工作线程等于最大线程 线程拒绝策略
+                 */
+                if (Objects.isNull(executorService)) {
+                    ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("flink-pool-%d").build();
+                    executorService = new ThreadPoolExecutor(8,
+                            8,
+                            1L,
+                            TimeUnit.MINUTES,
+                            new LinkedBlockingQueue<>(4), namedThreadFactory, new ThreadPoolExecutor.CallerRunsPolicy());
+                }
+            }
+        }
+        return TtlExecutors.getTtlExecutorService(executorService);
+    }
+
+}

Original file line number	Diff line number	Diff line change
`@@ -19,6 +19,7 @@`
`19`	`19`	`@JsonInclude(JsonInclude.Include.NON_NULL)`
`20`	`20`	`public class NodeSettingInfo {`
`21`	`21`
	`22`	`+ private String host;`
`22`	`23`	`private String username;`
`23`	`24`	`private String password;`
`24`	`25`	`}`
Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,8 @@ public interface ClusterRepository {`
`44`	`44`	`* 根据clusterCode获取唯一条`
`45`	`45`	`*`
`46`	`46`	`* @param clusterCode clusterCode`
	`47`	`+ * @param tenantId 租户id`
`47`	`48`	`* @return org.abigballofmud.flink.client.api.dto.ClusterDTO`
`48`	`49`	`*/`
`49`		`- ClusterDTO selectOne(String clusterCode);`
	`50`	`+ ClusterDTO selectOne(String clusterCode, Long tenantId);`
`50`	`51`	`}`
Original file line number	Diff line number	Diff line change
`@@ -55,9 +55,10 @@ public ClusterDTO detail(Long tenantId, Long clusterId) {`
`55`	`55`	`}`
`56`	`56`
`57`	`57`	`@Override`
`58`		`- public ClusterDTO selectOne(String clusterCode) {`
	`58`	`+ public ClusterDTO selectOne(String clusterCode,Long tenantId) {`
`59`	`59`	`QueryWrapper<Cluster> queryWrapper = new QueryWrapper<>();`
`60`	`60`	`queryWrapper.eq(Cluster.FIELD_CLUSTER_CODE, clusterCode);`
	`61`	`+ queryWrapper.eq(Cluster.FIELD_TENANT_ID, tenantId);`
`61`	`62`	`Cluster cluster = clusterMapper.selectOne(queryWrapper);`
`62`	`63`	`return ClusterConvertMapper.INSTANCE.entityToDTO(cluster);`
`63`	`64`	`}`