ES客户端数可配置化

Author: zengqiao

km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/es/ESOpClient.java

@@ -16,7 +16,6 @@
 import com.didiglobal.logi.log.LogFactory;
 import com.google.common.collect.Lists;
 import com.xiaojukeji.know.streaming.km.common.bean.po.BaseESPO;
-import com.xiaojukeji.know.streaming.km.common.constant.ESConstant;
 import com.xiaojukeji.know.streaming.km.common.utils.EnvUtil;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -37,14 +36,14 @@
 
 @Component
 public class ESOpClient {
-
     private static final ILog LOGGER   = LogFactory.getLog("ES_LOGGER");
 
     /**
      * es 地址
      */
     @Value("${es.client.address}")
     private String                          esAddress;
+
     /**
      * es 访问密码
      */
@@ -54,22 +53,32 @@ public class ESOpClient {
     /**
      * 客户端个数
      */
-    private static final int ES_CLIENT_COUNT    = 30;
-
-    private static final int MAX_RETRY_CNT      = 5;
+    @Value("${es.client.client-cnt:10}")
+    private Integer                         clientCnt;
 
-    private static final int ES_IO_THREAD_COUNT = 4;
+    /**
+     * 最大重试次数
+     */
+    @Value("${es.client.max-retry-cnt:5}")
+    private Integer                         maxRetryCnt;
 
+    /**
+     * IO线程数
+     */
+    @Value("${es.client.io-thread-cnt:2}")
+    private Integer                         ioThreadCnt;
 
     /**
      *  更新es数据的客户端连接队列
      */
-    private LinkedBlockingQueue<ESClient>   esClientPool  = new LinkedBlockingQueue<>( ES_CLIENT_COUNT );
+    private LinkedBlockingQueue<ESClient>   esClientPool;
 
     @PostConstruct
     public void init(){
-        for (int i = 0; i < ES_CLIENT_COUNT; ++i) {
-            ESClient esClient = buildEsClient(esAddress, esPass, "", "");
+        esClientPool = new LinkedBlockingQueue<>( clientCnt );
+
+        for (int i = 0; i < clientCnt; ++i) {
+            ESClient esClient = this.buildEsClient(esAddress, esPass, "", "");
             if (esClient != null) {
                 this.esClientPool.add(esClient);
                 LOGGER.info("class=ESOpClient||method=init||msg=add new es client {}", esAddress);
@@ -245,7 +254,7 @@ public boolean index(String indexName, String id, String source) {
             esIndexRequest.source(source);
             esIndexRequest.id(id);
 
-            for (int i = 0; i < MAX_RETRY_CNT; ++i) {
+            for (int i = 0; i < this.maxRetryCnt; ++i) {
                 response = esClient.index(esIndexRequest).actionGet(10, TimeUnit.SECONDS);
                 if (response == null) {
                     continue;
@@ -307,7 +316,7 @@ public boolean batchInsert(String indexName, List<? extends BaseESPO> pos) {
                 batchRequest.addNode(BatchType.INDEX, indexName, null, po.getKey(), JSON.toJSONString(po));
             }
 
-            for (int i = 0; i < MAX_RETRY_CNT; ++i) {
+            for (int i = 0; i < this.maxRetryCnt; ++i) {
                 response = esClient.batch(batchRequest).actionGet(2, TimeUnit.MINUTES);
                 if (response == null) {continue;}
 
@@ -428,8 +437,8 @@ private ESClient buildEsClient(String address,String password,String clusterName
             if(StringUtils.isNotBlank(password)){
                 esClient.setPassword(password);
             }
-            if(ES_IO_THREAD_COUNT > 0) {
-                esClient.setIoThreadCount( ES_IO_THREAD_COUNT );
+            if(this.ioThreadCnt > 0) {
+                esClient.setIoThreadCount( this.ioThreadCnt );
             }
 
             // 配置http超时
@@ -439,11 +448,13 @@ private ESClient buildEsClient(String address,String password,String clusterName
 
             return esClient;
         } catch (Exception e) {
-            esClient.close();
-
-            LOGGER.error("class=ESESOpClient||method=buildEsClient||errMsg={}||address={}", e.getMessage(), address,
-                    e);
+            try {
+                esClient.close();
+            } catch (Exception innerE) {
+                // ignore
+            }
 
+            LOGGER.error("class=ESESOpClient||method=buildEsClient||errMsg={}||address={}", e.getMessage(), address, e);
             return null;
         }
     }

km-rest/src/main/resources/application.yml

@@ -73,8 +73,13 @@ client-pool:
     borrow-timeout-unit-ms: 5000  # 租借超时时间，单位秒
 
 
-# es客户端服务地址
-es.client.address: 127.0.0.1:8060
+# ES客户端配置
+es:
+  client:
+    address: 127.0.0.1:8091,127.0.0.1:8061,127.0.0.1:8061
+    client-cnt: 10
+    io-thread-cnt: 2
+    max-retry-cnt: 5
 
 # 普罗米修斯指标导出相关配置
 management: