[incubator-shenyu] branch master updated: [type:refactor] change logging-elasticsearch-plugin thread pool (#3593)

[xi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

xiaoyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-shenyu.git


The following commit(s) were added to refs/heads/master by this push:
     new a6aa6e714 [type:refactor] change logging-elasticsearch-plugin thread pool (#3593)
a6aa6e714 is described below

commit a6aa6e714c892bfa42579a1ba894998d4f3fd916
Author: qinghai777 <80...@users.noreply.github.com>
AuthorDate: Wed Jun 22 17:47:15 2022 +0800

    [type:refactor] change logging-elasticsearch-plugin thread pool (#3593)
    
    * change es thread pool
    
    * change es thread pool again
---
 .../logging/elasticsearch/AbstractLogCollector.java     | 17 ++++++++++++++---
 1 file changed, 14 insertions(+), 3 deletions(-)

diff --git a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-elasticsearch/src/main/java/org/apache/shenyu/plugin/logging/elasticsearch/AbstractLogCollector.java b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-elasticsearch/src/main/java/org/apache/shenyu/plugin/logging/elasticsearch/AbstractLogCollector.java
index 4aa98db44..04005cdc9 100644
--- a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-elasticsearch/src/main/java/org/apache/shenyu/plugin/logging/elasticsearch/AbstractLogCollector.java
+++ b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-elasticsearch/src/main/java/org/apache/shenyu/plugin/logging/elasticsearch/AbstractLogCollector.java
@@ -17,6 +17,11 @@
 
 package org.apache.shenyu.plugin.logging.elasticsearch;
 
+import org.apache.shenyu.common.concurrent.MemorySafeTaskQueue;
+import org.apache.shenyu.common.concurrent.ShenyuThreadFactory;
+import org.apache.shenyu.common.concurrent.ShenyuThreadPoolExecutor;
+import org.apache.shenyu.common.config.ShenyuConfig;
+import org.apache.shenyu.common.constant.Constants;
 import org.apache.shenyu.common.utils.ThreadUtils;
 import org.apache.shenyu.plugin.logging.elasticsearch.entity.ShenyuRequestLog;
 import org.apache.shenyu.plugin.logging.elasticsearch.utils.LogCollectConfigUtils;
@@ -28,9 +33,8 @@ import java.util.List;
 import java.util.Objects;
 
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -49,11 +53,18 @@ public abstract class AbstractLogCollector implements LogCollector {
 
     private final AtomicBoolean started = new AtomicBoolean(true);
 
+    private final ShenyuConfig config = new ShenyuConfig();
+
     @Override
     public void start() {
         bufferSize = LogCollectConfigUtils.getGlobalLogConfig().getBufferQueueSize();
         bufferQueue = new LinkedBlockingDeque<>(bufferSize);
-        ExecutorService threadExecutor = Executors.newSingleThreadExecutor();
+        final ShenyuConfig.SharedPool sharedPool = config.getSharedPool();
+        ShenyuThreadPoolExecutor threadExecutor = new ShenyuThreadPoolExecutor(sharedPool.getCorePoolSize(),
+                sharedPool.getMaximumPoolSize(), sharedPool.getKeepAliveTime(), TimeUnit.MILLISECONDS,
+                new MemorySafeTaskQueue<>(Constants.THE_256_MB),
+                ShenyuThreadFactory.create(config.getSharedPool().getPrefix(), true),
+                new ThreadPoolExecutor.AbortPolicy());
         started.set(true);
         threadExecutor.execute(this::consume);
     }