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 3d284a0 [ISSUE #3026] [type:new feature] part2: dubbo support shared
thread pool (#3053)
3d284a0 is described below
commit 3d284a0ceba5694c929d0beba7a9002c7cf3ee38
Author: dragon-zhang <[email protected]>
AuthorDate: Fri Mar 18 10:03:00 2022 +0800
[ISSUE #3026] [type:new feature] part2: dubbo support shared thread pool
(#3053)
* [ISSUE #3026] [type:new feature] dubbo support shared thread pool
* fix bug and code style
* fix code style
* fix code style
* fix code style
* fix code style
* fix message
---
script/2.4.2-upgrade-2.4.3-mysql.sql | 1 +
script/2.4.2-upgrade-2.4.3-pg.sql | 1 +
.../src/main/resources/sql-script/h2/schema.sql | 1 +
.../src/main/resources/sql-script/mysql/schema.sql | 1 +
.../src/main/resources/sql-script/pg/schema.sql | 1 +
.../src/main/resources/application.yml | 8 +
.../common/concurrent/MemoryLimitCalculator.java | 67 ++++++++
.../MemoryLimitedLinkedBlockingQueue.java | 3 +
.../common/concurrent/MemoryLimitedTaskQueue.java | 91 +++++++++++
.../shenyu/common/concurrent/MemoryLimiter.java | 2 +
.../concurrent/ShenyuThreadPoolExecutor.java | 62 ++++++++
.../apache/shenyu/common/config/ShenyuConfig.java | 176 ++++++++++++++++++++-
.../alibaba/dubbo/threadpool/SharedThreadPool.java | 41 +++++
.../com.alibaba.dubbo.common.threadpool.ThreadPool | 18 +++
.../apache/dubbo/threadpool/SharedThreadPool.java | 41 +++++
.../org.apache.dubbo.common.threadpool.ThreadPool | 18 +++
shenyu-web/pom.xml | 5 +
.../ShenyuThreadPoolConfiguration.java | 59 +++++++
18 files changed, 595 insertions(+), 1 deletion(-)
diff --git a/script/2.4.2-upgrade-2.4.3-mysql.sql
b/script/2.4.2-upgrade-2.4.3-mysql.sql
index f15e187..4d2bfd7 100644
--- a/script/2.4.2-upgrade-2.4.3-mysql.sql
+++ b/script/2.4.2-upgrade-2.4.3-mysql.sql
@@ -31,6 +31,7 @@ INSERT IGNORE INTO plugin_handle (`plugin_id`, `field`,
`label`, `data_type`, `t
INSERT IGNORE INTO plugin_handle (`plugin_id`, `field`, `label`, `data_type`,
`type`, `sort`, `ext_obj`) VALUES ('6', 'queues', 'queues', '1', '3', '0',
'{"required":"0","defaultValue":"0","placeholder":"queues","rule":""}');
-- insert dict for dubbo plugin
+INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'shared', 'shared', '', '4', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'fixed', 'fixed', '', '3', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'eager', 'eager', '', '2', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'cached', 'cached', '', '0', '1');
diff --git a/script/2.4.2-upgrade-2.4.3-pg.sql
b/script/2.4.2-upgrade-2.4.3-pg.sql
index d2fc997..92751cd 100644
--- a/script/2.4.2-upgrade-2.4.3-pg.sql
+++ b/script/2.4.2-upgrade-2.4.3-pg.sql
@@ -31,6 +31,7 @@ INSERT IGNORE INTO plugin_handle (`plugin_id`, `field`,
`label`, `data_type`, `t
INSERT IGNORE INTO plugin_handle (`plugin_id`, `field`, `label`, `data_type`,
`type`, `sort`, `ext_obj`) VALUES ('6', 'queues', 'queues', '1', '3', '0',
'{"required":"0","defaultValue":"0","placeholder":"queues","rule":""}');
-- insert dict for dubbo plugin
+INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'shared', 'shared', '', '4', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'fixed', 'fixed', '', '3', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'eager', 'eager', '', '2', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'cached', 'cached', '', '0', '1');
diff --git a/shenyu-admin/src/main/resources/sql-script/h2/schema.sql
b/shenyu-admin/src/main/resources/sql-script/h2/schema.sql
index 6696a90..385dd28 100644
--- a/shenyu-admin/src/main/resources/sql-script/h2/schema.sql
+++ b/shenyu-admin/src/main/resources/sql-script/h2/schema.sql
@@ -312,6 +312,7 @@ INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`,
`dict_name`, `dict_value`,
/*insert dict for dubbo plugin*/
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('gray', 'GRAY_STATUS',
'close', 'false', 'close', '1', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('gray', 'GRAY_STATUS', 'open',
'true', 'open', '0', '1');
+INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'shared', 'shared', '', '4', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'fixed', 'fixed', '', '3', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'eager', 'eager', '', '2', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'cached', 'cached', '', '0', '1');
diff --git a/shenyu-admin/src/main/resources/sql-script/mysql/schema.sql
b/shenyu-admin/src/main/resources/sql-script/mysql/schema.sql
index 7264c45..c097f07 100644
--- a/shenyu-admin/src/main/resources/sql-script/mysql/schema.sql
+++ b/shenyu-admin/src/main/resources/sql-script/mysql/schema.sql
@@ -317,6 +317,7 @@ INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`,
`dict_name`, `dict_value`,
/*insert dict for dubbo plugin*/
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('gray', 'GRAY_STATUS',
'close', 'false', 'close', '1', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('gray', 'GRAY_STATUS', 'open',
'true', 'open', '0', '1');
+INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'shared', 'shared', '', '4', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'fixed', 'fixed', '', '3', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'eager', 'eager', '', '2', '1');
INSERT IGNORE INTO shenyu_dict (`type`, `dict_code`, `dict_name`,
`dict_value`, `desc`, `sort`, `enabled`) VALUES ('threadpool', 'THREADPOOL',
'cached', 'cached', '', '0', '1');
diff --git a/shenyu-admin/src/main/resources/sql-script/pg/schema.sql
b/shenyu-admin/src/main/resources/sql-script/pg/schema.sql
index 78ca24c..4c30bf2 100644
--- a/shenyu-admin/src/main/resources/sql-script/pg/schema.sql
+++ b/shenyu-admin/src/main/resources/sql-script/pg/schema.sql
@@ -1006,6 +1006,7 @@ ELSE
/*insert dict for dubbo plugin*/
PERFORM public.dblink_exec('init_conn', 'INSERT INTO shenyu_dict ( type
, dict_code , dict_name , dict_value , "desc" , sort , enabled ) VALUES
(''' || 'gray' || ''', ''' || 'GRAY_STATUS' || ''', ''' || 'close' || ''', '''
|| 'false' || ''', ''' || 'close' || ''', ''' || '1' || ''', ''' || '1' ||
''');');
PERFORM public.dblink_exec('init_conn', 'INSERT INTO shenyu_dict ( type
, dict_code , dict_name , dict_value , "desc" , sort , enabled ) VALUES
(''' || 'gray' || ''', ''' || 'GRAY_STATUS' || ''', ''' || 'open' || ''', '''
|| 'true' || ''', ''' || 'open' || ''', ''' || '0' || ''', ''' || '1' ||
''');');
+ PERFORM public.dblink_exec('init_conn', 'INSERT INTO shenyu_dict ( type
, dict_code , dict_name , dict_value , "desc" , sort , enabled ) VALUES
(''' || 'threadpool' || ''', ''' || 'THREADPOOL' || ''', ''' || 'shared' ||
''', ''' || 'shared' || ''', ''' || '' || ''', ''' || '4' || ''', ''' || '1' ||
''');');
PERFORM public.dblink_exec('init_conn', 'INSERT INTO shenyu_dict ( type
, dict_code , dict_name , dict_value , "desc" , sort , enabled ) VALUES
(''' || 'threadpool' || ''', ''' || 'THREADPOOL' || ''', ''' || 'fixed' || ''',
''' || 'fixed' || ''', ''' || '' || ''', ''' || '3' || ''', ''' || '1' ||
''');');
PERFORM public.dblink_exec('init_conn', 'INSERT INTO shenyu_dict ( type
, dict_code , dict_name , dict_value , "desc" , sort , enabled ) VALUES
(''' || 'threadpool' || ''', ''' || 'THREADPOOL' || ''', ''' || 'eager' || ''',
''' || 'eager' || ''', ''' || '' || ''', ''' || '2' || ''', ''' || '1' ||
''');');
PERFORM public.dblink_exec('init_conn', 'INSERT INTO shenyu_dict ( type
, dict_code , dict_name , dict_value , "desc" , sort , enabled ) VALUES
(''' || 'threadpool' || ''', ''' || 'THREADPOOL' || ''', ''' || 'cached' ||
''', ''' || 'cached' || ''', ''' || '' || ''', ''' || '0' || ''', ''' || '1' ||
''');');
diff --git a/shenyu-bootstrap/src/main/resources/application.yml
b/shenyu-bootstrap/src/main/resources/application.yml
index 8a7b801..5b72726 100644
--- a/shenyu-bootstrap/src/main/resources/application.yml
+++ b/shenyu-bootstrap/src/main/resources/application.yml
@@ -198,6 +198,14 @@ shenyu:
local:
enabled: true
sha512Key:
"BA3253876AED6BC22D4A6FF53D8406C6AD864195ED144AB5C87621B6C233B548BAEAE6956DF346EC8C17F5EA10F35EE3CBC514797ED7DDD3145464E2A0BAB413"
+# sharedPool:
+# enable: true
+# prefix: "shenyu-shared"
+# corePoolSize: 200
+# maximumPoolSize: 2000
+# keepAliveTime: 60000
+# # 1GB
+# maxWorkQueueMemory: 1073741824
logging:
level:
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitCalculator.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitCalculator.java
new file mode 100644
index 0000000..5afc8ab
--- /dev/null
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitCalculator.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.common.concurrent;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryMXBean;
+import java.lang.management.MemoryUsage;
+
+/**
+ * {@link javax.management.MXBean} technology is used to calculate the memory
+ * limit by using the percentage of the current maximum available memory,
+ * which can be used with {@link
org.apache.shenyu.common.concurrent.MemoryLimiter}.
+ *
+ * @see org.apache.shenyu.common.concurrent.MemoryLimiter
+ */
+public class MemoryLimitCalculator {
+
+ private static final MemoryMXBean MX_BEAN =
ManagementFactory.getMemoryMXBean();
+
+ /**
+ * Get the maximum available memory of the current JVM.
+ *
+ * @return maximum available memory
+ */
+ public static long maxAvailable() {
+ final MemoryUsage usage = MX_BEAN.getHeapMemoryUsage();
+ return usage.getCommitted();
+ }
+
+ /**
+ * Take the current JVM's maximum available memory
+ * as a percentage of the result as the limit.
+ *
+ * @param percentage percentage
+ * @return available memory
+ */
+ public static long calculate(final float percentage) {
+ if (percentage <= 0 || percentage > 1) {
+ throw new IllegalArgumentException();
+ }
+ return (long) (maxAvailable() * percentage);
+ }
+
+ /**
+ * By default, it takes 80% of the maximum available memory of the current
JVM.
+ *
+ * @return available memory
+ */
+ public static long defaultLimit() {
+ return (long) (maxAvailable() * 0.8);
+ }
+}
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueue.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueue.java
index dc64d99..ea6ec11 100644
---
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueue.java
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedLinkedBlockingQueue.java
@@ -24,6 +24,9 @@ import java.util.concurrent.TimeUnit;
/**
* Can completely solve the OOM problem caused by {@link
java.util.concurrent.LinkedBlockingQueue}.
+ *
+ * @see org.apache.shenyu.common.concurrent.MemoryLimiter
+ * @see org.apache.shenyu.common.concurrent.MemoryLimitCalculator
*/
public class MemoryLimitedLinkedBlockingQueue<E> extends
LinkedBlockingQueue<E> {
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedTaskQueue.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedTaskQueue.java
new file mode 100644
index 0000000..c6a1ec5
--- /dev/null
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimitedTaskQueue.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.common.concurrent;
+
+import java.lang.instrument.Instrumentation;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * MemoryLimitedTaskQueue in the {@link
org.apache.shenyu.common.concurrent.ShenyuThreadPoolExecutor}.
+ * It offer a task if the executor's submittedTaskCount less than
currentPoolThreadSize
+ * or the currentPoolThreadSize more than executor's maximumPoolSize.
+ * That can make the executor create new worker
+ * when the task num is bigger than corePoolSize but less than maximumPoolSize.
+ */
+public class MemoryLimitedTaskQueue<R extends Runnable> extends
MemoryLimitedLinkedBlockingQueue<Runnable> {
+
+ private static final long serialVersionUID = -2635853580887179627L;
+
+ private ShenyuThreadPoolExecutor executor;
+
+ public MemoryLimitedTaskQueue(final Instrumentation inst) {
+ super(inst);
+ }
+
+ public MemoryLimitedTaskQueue(final long memoryLimit, final
Instrumentation inst) {
+ super(memoryLimit, inst);
+ }
+
+ /**
+ * set the executor.
+ *
+ * @param executor executor
+ */
+ public void setExecutor(final ShenyuThreadPoolExecutor executor) {
+ this.executor = executor;
+ }
+
+ @Override
+ public boolean offer(final Runnable runnable) {
+ if (executor == null) {
+ throw new RejectedExecutionException("The task queue does not have
executor!");
+ }
+
+ int currentPoolThreadSize = executor.getPoolSize();
+ // have free worker. put task into queue to let the worker deal with
task.
+ if (executor.getActiveCount() < currentPoolThreadSize) {
+ return super.offer(runnable);
+ }
+
+ // return false to let executor create new worker.
+ if (currentPoolThreadSize < executor.getMaximumPoolSize()) {
+ return false;
+ }
+
+ // currentPoolThreadSize >= max
+ return super.offer(runnable);
+ }
+
+ /**
+ * retry offer task.
+ *
+ * @param o task
+ * @param timeout timeout
+ * @param unit timeout unit
+ * @return offer success or not
+ * @throws java.util.concurrent.RejectedExecutionException if executor is
terminated.
+ * @throws java.lang.InterruptedException if the current
thread is interrupted.
+ */
+ public boolean retryOffer(final Runnable o, final long timeout, final
TimeUnit unit) throws InterruptedException {
+ if (executor.isShutdown()) {
+ throw new RejectedExecutionException("Executor is shutdown!");
+ }
+ return super.offer(o, timeout, unit);
+ }
+}
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimiter.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimiter.java
index 2d3e62f..335728d 100644
---
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimiter.java
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/MemoryLimiter.java
@@ -25,6 +25,8 @@ import java.util.concurrent.locks.ReentrantLock;
/**
* memory limiter.
+ *
+ * @see org.apache.shenyu.common.concurrent.MemoryLimitCalculator
*/
public class MemoryLimiter {
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/ShenyuThreadPoolExecutor.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/ShenyuThreadPoolExecutor.java
new file mode 100644
index 0000000..1a9f93b
--- /dev/null
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/concurrent/ShenyuThreadPoolExecutor.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.common.concurrent;
+
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * ShenyuThreadPoolExecutor.
+ */
+public class ShenyuThreadPoolExecutor extends ThreadPoolExecutor {
+
+ public ShenyuThreadPoolExecutor(final int corePoolSize,
+ final int maximumPoolSize,
+ final long keepAliveTime,
+ final TimeUnit unit,
+ final MemoryLimitedTaskQueue<Runnable>
workQueue,
+ final ThreadFactory threadFactory,
+ final RejectedExecutionHandler handler) {
+ super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, handler);
+ }
+
+ @Override
+ public void execute(final Runnable command) {
+ if (command == null) {
+ throw new NullPointerException();
+ }
+
+ try {
+ super.execute(command);
+ } catch (RejectedExecutionException e) {
+ // retry to offer the task into queue.
+ @SuppressWarnings("all")
+ final MemoryLimitedTaskQueue queue = (MemoryLimitedTaskQueue)
super.getQueue();
+ try {
+ if (!queue.retryOffer(command, 0, TimeUnit.MILLISECONDS)) {
+ throw new RejectedExecutionException("Queue capacity is
full.", e);
+ }
+ } catch (InterruptedException t) {
+ throw new RejectedExecutionException(t);
+ }
+ }
+ }
+}
diff --git
a/shenyu-common/src/main/java/org/apache/shenyu/common/config/ShenyuConfig.java
b/shenyu-common/src/main/java/org/apache/shenyu/common/config/ShenyuConfig.java
index 1c1786e..8ffce22 100644
---
a/shenyu-common/src/main/java/org/apache/shenyu/common/config/ShenyuConfig.java
+++
b/shenyu-common/src/main/java/org/apache/shenyu/common/config/ShenyuConfig.java
@@ -17,6 +17,7 @@
package org.apache.shenyu.common.config;
+import org.apache.shenyu.common.concurrent.MemoryLimitCalculator;
import org.springframework.util.StringUtils;
import java.util.ArrayList;
@@ -55,7 +56,27 @@ public class ShenyuConfig {
private Local local = new Local();
private WebsocketConfig websocket = new WebsocketConfig();
-
+
+ private SharedPool sharedPool = new SharedPool();
+
+ /**
+ * Gets the shared thread pool config.
+ *
+ * @return the shared thread pool config
+ */
+ public SharedPool getSharedPool() {
+ return sharedPool;
+ }
+
+ /**
+ * Sets the shared thread pool config.
+ *
+ * @param sharedPool the shared thread pool config
+ */
+ public void setSharedPool(final SharedPool sharedPool) {
+ this.sharedPool = sharedPool;
+ }
+
/**
* Gets the local config.
*
@@ -1170,4 +1191,157 @@ public class ShenyuConfig {
this.maxFramePayloadSize = maxFramePayloadSize;
}
}
+
+ /**
+ * The type Shared Thread Pool.
+ */
+ public static class SharedPool {
+
+ /**
+ * Whether to enable shared thread pool, defaults to false.
+ * Note: it is planned to be enabled by default when all RPC/HTTP
+ * plugins support this shared thread pool.
+ */
+ private Boolean enable = Boolean.FALSE;
+
+ /**
+ * The the thread name prefix, defaults to shenyu-shared.
+ */
+ private String prefix = "shenyu-shared";
+
+ /**
+ * the number of threads to keep in the thread pool.
+ */
+ private Integer corePoolSize = 200;
+
+ /**
+ * the maximum number of threads to allow in the thread pool.
+ */
+ private Integer maximumPoolSize = Integer.MAX_VALUE;
+
+ /**
+ * when the number of threads is greater than the core,
+ * this is the maximum time that excess idle threads
+ * will wait for new tasks before terminating.
+ * Note: the unit of time is {@link
java.util.concurrent.TimeUnit#MILLISECONDS}
+ */
+ private Long keepAliveTime = 60000L;
+
+ /**
+ * Maximum memory allowed to be used by a blocking queue, yes, unlike
other
+ * implementations of {@link java.util.concurrent.BlockingQueue}
+ * (which all control memory based on the length of the blocking
queue),
+ * {@link
org.apache.shenyu.common.concurrent.MemoryLimitedLinkedBlockingQueue}
+ * controls memory directly by calculating the memory size used by the
blocking queue.
+ */
+ private Long maxWorkQueueMemory = MemoryLimitCalculator.defaultLimit();
+
+ /**
+ * Whether to enable shared thread pool.
+ *
+ * @return whether to enable
+ */
+ public Boolean getEnable() {
+ return enable;
+ }
+
+ /**
+ * Set enable.
+ *
+ * @param enable the enable
+ */
+ public void setEnable(final Boolean enable) {
+ this.enable = enable;
+ }
+
+ /**
+ * Get shared thread pool name prefix.
+ *
+ * @return the shared thread pool name prefix
+ */
+ public String getPrefix() {
+ return prefix;
+ }
+
+ /**
+ * Set prefix.
+ *
+ * @param prefix the prefix
+ */
+ public void setPrefix(final String prefix) {
+ this.prefix = prefix;
+ }
+
+ /**
+ * Get shared thread pool core size.
+ *
+ * @return the shared thread pool core size
+ */
+ public Integer getCorePoolSize() {
+ return corePoolSize;
+ }
+
+ /**
+ * Set core pool size.
+ *
+ * @param corePoolSize the core pool size
+ */
+ public void setCorePoolSize(final Integer corePoolSize) {
+ this.corePoolSize = corePoolSize;
+ }
+
+ /**
+ * Get shared thread pool maximum size.
+ *
+ * @return the shared thread pool name prefix
+ */
+ public Integer getMaximumPoolSize() {
+ return maximumPoolSize;
+ }
+
+ /**
+ * Set max pool size.
+ *
+ * @param maximumPoolSize the max pool size
+ */
+ public void setMaximumPoolSize(final Integer maximumPoolSize) {
+ this.maximumPoolSize = maximumPoolSize;
+ }
+
+ /**
+ * Get shared thread pool keep alive time.
+ *
+ * @return the shared thread pool keep alive time
+ */
+ public Long getKeepAliveTime() {
+ return keepAliveTime;
+ }
+
+ /**
+ * Set keep alive time.
+ *
+ * @param keepAliveTime the keep alive time
+ */
+ public void setKeepAliveTime(final Long keepAliveTime) {
+ this.keepAliveTime = keepAliveTime;
+ }
+
+ /**
+ * Get shared thread pool max work queue memory.
+ *
+ * @return the shared thread pool max work queue memory
+ */
+ public Long getMaxWorkQueueMemory() {
+ return maxWorkQueueMemory;
+ }
+
+ /**
+ * Set max work queue memory.
+ *
+ * @param maxWorkQueueMemory the max work queue memory
+ */
+ public void setMaxWorkQueueMemory(final Long maxWorkQueueMemory) {
+ this.maxWorkQueueMemory = maxWorkQueueMemory;
+ }
+ }
}
diff --git
a/shenyu-plugin/shenyu-plugin-dubbo/shenyu-plugin-alibaba-dubbo/src/main/java/org/apache/shenyu/plugin/alibaba/dubbo/threadpool/SharedThreadPool.java
b/shenyu-plugin/shenyu-plugin-dubbo/shenyu-plugin-alibaba-dubbo/src/main/java/org/apache/shenyu/plugin/alibaba/dubbo/threadpool/SharedThreadPool.java
new file mode 100644
index 0000000..ea39084
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-dubbo/shenyu-plugin-alibaba-dubbo/src/main/java/org/apache/shenyu/plugin/alibaba/dubbo/threadpool/SharedThreadPool.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.plugin.alibaba.dubbo.threadpool;
+
+import com.alibaba.dubbo.common.URL;
+import com.alibaba.dubbo.common.threadpool.ThreadPool;
+import org.apache.shenyu.common.concurrent.ShenyuThreadPoolExecutor;
+import org.apache.shenyu.common.exception.ShenyuException;
+import org.apache.shenyu.plugin.api.utils.SpringBeanUtils;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
+
+import java.util.concurrent.Executor;
+
+/**
+ * The type Shared Thread Pool.
+ */
+public class SharedThreadPool implements ThreadPool {
+ @Override
+ public Executor getExecutor(final URL url) {
+ try {
+ return
SpringBeanUtils.getInstance().getBean(ShenyuThreadPoolExecutor.class);
+ } catch (NoSuchBeanDefinitionException t) {
+ throw new ShenyuException("shared thread pool is not enable,
config ${shenyu.sharedPool.enable} in your xml/yml !", t);
+ }
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-dubbo/shenyu-plugin-alibaba-dubbo/src/main/resources/META-INF/dubbo/com.alibaba.dubbo.common.threadpool.ThreadPool
b/shenyu-plugin/shenyu-plugin-dubbo/shenyu-plugin-alibaba-dubbo/src/main/resources/META-INF/dubbo/com.alibaba.dubbo.common.threadpool.ThreadPool
new file mode 100644
index 0000000..b6d888e
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-dubbo/shenyu-plugin-alibaba-dubbo/src/main/resources/META-INF/dubbo/com.alibaba.dubbo.common.threadpool.ThreadPool
@@ -0,0 +1,18 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+shared=org.apache.shenyu.plugin.alibaba.dubbo.threadpool.SharedThreadPool
\ No newline at end of file
diff --git
a/shenyu-plugin/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/main/java/org/apache/shenyu/plugin/apache/dubbo/threadpool/SharedThreadPool.java
b/shenyu-plugin/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/main/java/org/apache/shenyu/plugin/apache/dubbo/threadpool/SharedThreadPool.java
new file mode 100644
index 0000000..c48e255
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/main/java/org/apache/shenyu/plugin/apache/dubbo/threadpool/SharedThreadPool.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.plugin.apache.dubbo.threadpool;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.threadpool.ThreadPool;
+import org.apache.shenyu.common.concurrent.ShenyuThreadPoolExecutor;
+import org.apache.shenyu.common.exception.ShenyuException;
+import org.apache.shenyu.plugin.api.utils.SpringBeanUtils;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
+
+import java.util.concurrent.Executor;
+
+/**
+ * The type Shared Thread Pool.
+ */
+public class SharedThreadPool implements ThreadPool {
+ @Override
+ public Executor getExecutor(final URL url) {
+ try {
+ return
SpringBeanUtils.getInstance().getBean(ShenyuThreadPoolExecutor.class);
+ } catch (NoSuchBeanDefinitionException t) {
+ throw new ShenyuException("shared thread pool is not enable,
config ${shenyu.sharedPool.enable} in your xml/yml !", t);
+ }
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/main/resources/META-INF/dubbo/org.apache.dubbo.common.threadpool.ThreadPool
b/shenyu-plugin/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/main/resources/META-INF/dubbo/org.apache.dubbo.common.threadpool.ThreadPool
new file mode 100644
index 0000000..86b6224
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-dubbo/shenyu-plugin-apache-dubbo/src/main/resources/META-INF/dubbo/org.apache.dubbo.common.threadpool.ThreadPool
@@ -0,0 +1,18 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+shared=org.apache.shenyu.plugin.apache.dubbo.threadpool.SharedThreadPool
\ No newline at end of file
diff --git a/shenyu-web/pom.xml b/shenyu-web/pom.xml
index 8f558a0..0048dc1 100644
--- a/shenyu-web/pom.xml
+++ b/shenyu-web/pom.xml
@@ -31,6 +31,11 @@
<artifactId>shenyu-plugin-base</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>net.bytebuddy</groupId>
+ <artifactId>byte-buddy-agent</artifactId>
+ <version>${bytebuddy.version}</version>
+ </dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
diff --git
a/shenyu-web/src/main/java/org/apache/shenyu/web/configuration/ShenyuThreadPoolConfiguration.java
b/shenyu-web/src/main/java/org/apache/shenyu/web/configuration/ShenyuThreadPoolConfiguration.java
new file mode 100644
index 0000000..7eea6f9
--- /dev/null
+++
b/shenyu-web/src/main/java/org/apache/shenyu/web/configuration/ShenyuThreadPoolConfiguration.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.web.configuration;
+
+import net.bytebuddy.agent.ByteBuddyAgent;
+import org.apache.shenyu.common.concurrent.MemoryLimitedTaskQueue;
+import org.apache.shenyu.common.concurrent.ShenyuThreadFactory;
+import org.apache.shenyu.common.concurrent.ShenyuThreadPoolExecutor;
+import org.apache.shenyu.common.config.ShenyuConfig;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.lang.instrument.Instrumentation;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * The type shenyu thread pool configuration.
+ */
+@Configuration
+public class ShenyuThreadPoolConfiguration {
+
+ /**
+ * crate shenyu shared thread pool executor.
+ *
+ * @param shenyuConfig the shenyu config
+ * @return the shenyu thread pool executor
+ */
+ @Bean
+ @ConditionalOnProperty(name = "shenyu.sharedPool.enable", havingValue =
"true")
+ public ShenyuThreadPoolExecutor shenyuThreadPoolExecutor(final
ShenyuConfig shenyuConfig) {
+ final Instrumentation instrumentation = ByteBuddyAgent.install();
+ final ShenyuConfig.SharedPool sharedPool =
shenyuConfig.getSharedPool();
+ final Integer corePoolSize = sharedPool.getCorePoolSize();
+ final Integer maximumPoolSize = sharedPool.getMaximumPoolSize();
+ final Long keepAliveTime = sharedPool.getKeepAliveTime();
+ final Long maxWorkQueueMemory = sharedPool.getMaxWorkQueueMemory();
+ return new ShenyuThreadPoolExecutor(corePoolSize, maximumPoolSize,
keepAliveTime,
+ TimeUnit.MILLISECONDS, new
MemoryLimitedTaskQueue<>(maxWorkQueueMemory, instrumentation),
+ ShenyuThreadFactory.create(sharedPool.getPrefix(), true),
+ new ThreadPoolExecutor.AbortPolicy());
+ }
+}
