This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new 9abb7f39 [ISSUE #1318] [Java] Support opt-in virtual threads (#1319)
9abb7f39 is described below
commit 9abb7f39ce9f74b41873a4f5253ab82af052d7a7
Author: qianye <[email protected]>
AuthorDate: Thu Aug 6 20:57:28 2026 +0800
[ISSUE #1318] [Java] Support opt-in virtual threads (#1319)
---
.github/workflows/java_build.yml | 7 +-
.../rocketmq/client/apis/ClientConfiguration.java | 11 +-
.../client/apis/ClientConfigurationBuilder.java | 22 +++-
.../apache/rocketmq/client/java/impl/Client.java | 9 ++
.../rocketmq/client/java/impl/ClientImpl.java | 37 +++---
.../client/java/impl/ClientManagerImpl.java | 15 +--
.../client/java/impl/consumer/ConsumeService.java | 6 +-
.../java/impl/consumer/FifoConsumeService.java | 4 +-
.../java/impl/consumer/LiteFifoConsumeService.java | 4 +-
.../impl/consumer/LiteStandardConsumeService.java | 4 +-
.../java/impl/consumer/PushConsumerImpl.java | 20 +--
.../java/impl/consumer/StandardConsumeService.java | 4 +-
.../client/java/misc/ExecutorServices.java | 135 +++++++++++++++++++++
.../client/apis/ClientConfigurationTest.java} | 28 +++--
.../client/java/misc/ExecutorServicesTest.java | 99 +++++++++++++++
15 files changed, 351 insertions(+), 54 deletions(-)
diff --git a/.github/workflows/java_build.yml b/.github/workflows/java_build.yml
index 7a77aa75..21851d4c 100644
--- a/.github/workflows/java_build.yml
+++ b/.github/workflows/java_build.yml
@@ -9,7 +9,7 @@ jobs:
fail-fast: false
matrix:
os: [ ubuntu-22.04, macos-latest, windows-2022 ]
- jdk: [11, 17]
+ jdk: ["11", "17", "21"]
steps:
- name: Checkout
uses: actions/checkout@v3
@@ -22,8 +22,13 @@ jobs:
distribution: "adopt"
cache: maven
- name: Build with Maven
+ if: matrix.jdk != '21'
working-directory: ./java
run: mvn -B package --file pom.xml
+ - name: Test virtual threads with JDK 21
+ if: matrix.jdk == '21'
+ working-directory: ./java
+ run: mvn -B -pl client -am '-Dspotbugs.skip=true'
'-Dtest=ClientConfigurationTest,ExecutorServicesTest' '-DfailIfNoTests=false'
'-Dsurefire.failIfNoSpecifiedTests=false' test
opentelemetry-instrumentation-compatibility:
runs-on: ubuntu-latest
diff --git
a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfiguration.java
b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfiguration.java
index fa8c45ea..15cec038 100644
---
a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfiguration.java
+++
b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfiguration.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.client.apis;
+import com.google.common.annotations.Beta;
import java.time.Duration;
import java.util.Collections;
import java.util.LinkedHashMap;
@@ -31,6 +32,7 @@ public class ClientConfiguration {
private final SessionCredentialsProvider sessionCredentialsProvider;
private final Duration requestTimeout;
private final boolean sslEnabled;
+ private final boolean virtualThreadsEnabled;
private final String namespace;
private final int maxStartupAttempts;
private final Map<String, String> clientProperties;
@@ -40,12 +42,14 @@ public class ClientConfiguration {
* logging warnings already, so we avoid repeating args check here.
*/
ClientConfiguration(String endpoints, SessionCredentialsProvider
sessionCredentialsProvider,
- Duration requestTimeout, boolean sslEnabled, String namespace, int
maxStartupAttempts,
+ Duration requestTimeout, boolean sslEnabled, boolean
virtualThreadsEnabled, String namespace,
+ int maxStartupAttempts,
Map<String, String> clientProperties) {
this.endpoints = endpoints;
this.sessionCredentialsProvider = sessionCredentialsProvider;
this.requestTimeout = requestTimeout;
this.sslEnabled = sslEnabled;
+ this.virtualThreadsEnabled = virtualThreadsEnabled;
this.namespace = namespace;
this.maxStartupAttempts = maxStartupAttempts;
this.clientProperties = Collections.unmodifiableMap(new
LinkedHashMap<>(clientProperties));
@@ -71,6 +75,11 @@ public class ClientConfiguration {
return sslEnabled;
}
+ @Beta
+ public boolean isVirtualThreadsEnabled() {
+ return virtualThreadsEnabled;
+ }
+
public String getNamespace() {
return namespace;
}
diff --git
a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfigurationBuilder.java
b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfigurationBuilder.java
index 4688e500..0a23b0df 100644
---
a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfigurationBuilder.java
+++
b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfigurationBuilder.java
@@ -20,6 +20,7 @@ package org.apache.rocketmq.client.apis;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
+import com.google.common.annotations.Beta;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.LinkedHashMap;
@@ -43,6 +44,7 @@ public class ClientConfigurationBuilder {
private SessionCredentialsProvider sessionCredentialsProvider = null;
private Duration requestTimeout = Duration.ofSeconds(3);
private boolean sslEnabled = true;
+ private boolean virtualThreadsEnabled = false;
private String namespace = "";
private int maxStartupAttempts = 3;
private final Map<String, String> clientProperties = new LinkedHashMap<>();
@@ -97,6 +99,22 @@ public class ClientConfigurationBuilder {
return this;
}
+ /**
+ * Enable or disable virtual threads for client task execution.
+ *
+ * <p>Virtual threads are disabled by default. When enabled on JDK 21 or
later, the client uses virtual threads for
+ * asynchronous RPCs, callbacks, telemetry commands and message
consumption. On earlier JDK versions, the client
+ * falls back to its platform-thread executors.
+ *
+ * @param virtualThreadsEnabled whether virtual threads should be enabled.
+ * @return The {@link ClientConfigurationBuilder} instance, to allow for
method chaining.
+ */
+ @Beta
+ public ClientConfigurationBuilder enableVirtualThreads(boolean
virtualThreadsEnabled) {
+ this.virtualThreadsEnabled = virtualThreadsEnabled;
+ return this;
+ }
+
/**
* Configure namespace for client
* @param namespace namespace
@@ -183,8 +201,8 @@ public class ClientConfigurationBuilder {
checkNotNull(requestTimeout, "requestTimeout should not be null");
// Keep build() defensive for maps supplied through
setClientProperties or future builder paths.
validateClientProperties(clientProperties);
- return new ClientConfiguration(endpoints, sessionCredentialsProvider,
requestTimeout, sslEnabled, namespace,
- maxStartupAttempts, clientProperties);
+ return new ClientConfiguration(endpoints, sessionCredentialsProvider,
requestTimeout, sslEnabled,
+ virtualThreadsEnabled, namespace, maxStartupAttempts,
clientProperties);
}
private static void validateClientProperties(Map<String, String>
properties) {
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/Client.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/Client.java
index 83510113..deb5d40b 100644
--- a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/Client.java
+++ b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/Client.java
@@ -55,6 +55,15 @@ public interface Client {
*/
boolean isSslEnabled();
+ /**
+ * Check whether virtual threads are enabled for client task execution.
+ *
+ * @return a boolean value indicating whether virtual threads are enabled
or not.
+ */
+ default boolean isVirtualThreadsEnabled() {
+ return false;
+ }
+
/**
* Reconnect telemetry to the specified endpoints.
*
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
index b523af73..e487fcc0 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
@@ -114,7 +114,7 @@ public abstract class ClientImpl extends
AbstractIdleService implements Client,
/**
* Telemetry command executor, which aims to execute commands from the
remote.
*/
- protected final ThreadPoolExecutor telemetryCommandExecutor;
+ protected final ExecutorService telemetryCommandExecutor;
protected final ClientId clientId;
private final ClientManager clientManager;
@@ -151,13 +151,14 @@ public abstract class ClientImpl extends
AbstractIdleService implements Client,
this.clientManager = new ClientManagerImpl(this);
final long clientIdIndex = clientId.getIndex();
- this.clientCallbackExecutor = new ThreadPoolExecutor(
- Runtime.getRuntime().availableProcessors(),
- Runtime.getRuntime().availableProcessors(),
- 60,
- TimeUnit.SECONDS,
- new LinkedBlockingQueue<>(),
- new ThreadFactoryImpl("ClientCallbackWorker", clientIdIndex));
+ this.clientCallbackExecutor = ExecutorServices.newExecutorService(
+ clientConfiguration.isVirtualThreadsEnabled(), () -> new
ThreadPoolExecutor(
+ Runtime.getRuntime().availableProcessors(),
+ Runtime.getRuntime().availableProcessors(),
+ 60,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(),
+ new ThreadFactoryImpl("ClientCallbackWorker", clientIdIndex)));
this.clientMeterManager = new ClientMeterManager(clientId,
clientConfiguration);
@@ -165,13 +166,14 @@ public abstract class ClientImpl extends
AbstractIdleService implements Client,
new CompositedMessageInterceptor(Collections.singletonList(new
MessageMeterInterceptor(this,
clientMeterManager)));
- this.telemetryCommandExecutor = new ThreadPoolExecutor(
- 1,
- 1,
- 60,
- TimeUnit.SECONDS,
- new LinkedBlockingQueue<>(),
- new ThreadFactoryImpl("CommandExecutor", clientIdIndex));
+ this.telemetryCommandExecutor = ExecutorServices.newExecutorService(
+ clientConfiguration.isVirtualThreadsEnabled(), () -> new
ThreadPoolExecutor(
+ 1,
+ 1,
+ 60,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(),
+ new ThreadFactoryImpl("CommandExecutor", clientIdIndex)));
}
/**
@@ -588,6 +590,11 @@ public abstract class ClientImpl extends
AbstractIdleService implements Client,
return clientConfiguration.isSslEnabled();
}
+ @Override
+ public boolean isVirtualThreadsEnabled() {
+ return clientConfiguration.isVirtualThreadsEnabled();
+ }
+
/**
* Send heartbeat data to the appointed endpoint
*
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManagerImpl.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManagerImpl.java
index c7b3e1bb..d26798a8 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManagerImpl.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientManagerImpl.java
@@ -137,13 +137,14 @@ public class ClientManagerImpl extends ClientManager {
Runtime.getRuntime().availableProcessors(),
new ThreadFactoryImpl("ClientScheduler", clientIndex));
- this.asyncWorker = new ThreadPoolExecutor(
- Runtime.getRuntime().availableProcessors(),
- Runtime.getRuntime().availableProcessors(),
- 60,
- TimeUnit.SECONDS,
- new LinkedBlockingQueue<>(50000),
- new ThreadFactoryImpl("ClientAsyncWorker", clientIndex));
+ this.asyncWorker =
ExecutorServices.newExecutorService(client.isVirtualThreadsEnabled(),
+ () -> new ThreadPoolExecutor(
+ Runtime.getRuntime().availableProcessors(),
+ Runtime.getRuntime().availableProcessors(),
+ 60,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(50000),
+ new ThreadFactoryImpl("ClientAsyncWorker", clientIndex)));
}
/**
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeService.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeService.java
index 11075c1e..4bd2745e 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeService.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ConsumeService.java
@@ -25,8 +25,8 @@ import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import java.time.Duration;
import java.util.List;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
@@ -43,12 +43,12 @@ public abstract class ConsumeService {
protected final ClientId clientId;
protected final String consumerGroup;
private final MessageListener messageListener;
- private final ThreadPoolExecutor consumptionExecutor;
+ private final ExecutorService consumptionExecutor;
private final MessageInterceptor messageInterceptor;
private final ScheduledExecutorService scheduler;
public ConsumeService(ClientId clientId, String consumerGroup,
- MessageListener messageListener, ThreadPoolExecutor
consumptionExecutor,
+ MessageListener messageListener, ExecutorService consumptionExecutor,
MessageInterceptor messageInterceptor, ScheduledExecutorService
scheduler) {
this.clientId = clientId;
this.consumerGroup = consumerGroup;
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeService.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeService.java
index 81bea5a6..9d4fb8d5 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeService.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/FifoConsumeService.java
@@ -25,8 +25,8 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadPoolExecutor;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
import org.apache.rocketmq.client.java.hook.MessageInterceptor;
@@ -40,7 +40,7 @@ class FifoConsumeService extends ConsumeService {
private final boolean enableFifoConsumeAccelerator;
public FifoConsumeService(ClientId clientId, String consumerGroup,
MessageListener messageListener,
- ThreadPoolExecutor consumptionExecutor, MessageInterceptor
messageInterceptor,
+ ExecutorService consumptionExecutor, MessageInterceptor
messageInterceptor,
ScheduledExecutorService scheduler, boolean
enableFifoConsumeAccelerator) {
super(clientId, consumerGroup, messageListener, consumptionExecutor,
messageInterceptor, scheduler);
this.enableFifoConsumeAccelerator = enableFifoConsumeAccelerator;
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteFifoConsumeService.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteFifoConsumeService.java
index 5e79850a..5abeace9 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteFifoConsumeService.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteFifoConsumeService.java
@@ -23,8 +23,8 @@ import com.google.common.util.concurrent.MoreExecutors;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.ConsumeResultSuspend;
@@ -36,7 +36,7 @@ import org.apache.rocketmq.client.java.misc.ClientId;
public class LiteFifoConsumeService extends FifoConsumeService {
public LiteFifoConsumeService(ClientId clientId, String consumerGroup,
MessageListener messageListener,
- ThreadPoolExecutor consumptionExecutor, MessageInterceptor
messageInterceptor,
+ ExecutorService consumptionExecutor, MessageInterceptor
messageInterceptor,
ScheduledExecutorService scheduler, boolean
enableFifoConsumeAccelerator) {
super(clientId, consumerGroup, messageListener, consumptionExecutor,
messageInterceptor, scheduler, enableFifoConsumeAccelerator);
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteStandardConsumeService.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteStandardConsumeService.java
index 6044aac3..6869be92 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteStandardConsumeService.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/LiteStandardConsumeService.java
@@ -22,8 +22,8 @@ import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import java.util.List;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadPoolExecutor;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
import org.apache.rocketmq.client.java.hook.MessageInterceptor;
@@ -36,7 +36,7 @@ public class LiteStandardConsumeService extends
ConsumeService {
private static final Logger log =
LoggerFactory.getLogger(LiteStandardConsumeService.class);
public LiteStandardConsumeService(ClientId clientId, String consumerGroup,
MessageListener messageListener,
- ThreadPoolExecutor consumptionExecutor, MessageInterceptor
messageInterceptor,
+ ExecutorService consumptionExecutor, MessageInterceptor
messageInterceptor,
ScheduledExecutorService scheduler) {
super(clientId, consumerGroup, messageListener, consumptionExecutor,
messageInterceptor, scheduler);
}
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
index 52f7fe67..94bd7374 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
@@ -41,6 +41,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
@@ -90,7 +91,7 @@ class PushConsumerImpl extends ConsumerImpl implements
PushConsumer {
private static final Logger log =
LoggerFactory.getLogger(PushConsumerImpl.class);
protected final MessageListener messageListener;
- protected final ThreadPoolExecutor consumptionExecutor;
+ protected final ExecutorService consumptionExecutor;
protected final boolean enableFifoConsumeAccelerator;
final AtomicLong consumptionOkQuantity;
@@ -152,13 +153,14 @@ class PushConsumerImpl extends ConsumerImpl implements
PushConsumer {
this.processQueueTable = new ConcurrentHashMap<>();
- this.consumptionExecutor = new ThreadPoolExecutor(
- consumptionThreadCount,
- consumptionThreadCount,
- 60,
- TimeUnit.SECONDS,
- new LinkedBlockingQueue<>(),
- new ThreadFactoryImpl("MessageConsumption",
this.getClientId().getIndex()));
+ this.consumptionExecutor =
ExecutorServices.newConcurrencyLimitedExecutorService(
+ clientConfiguration.isVirtualThreadsEnabled(),
consumptionThreadCount, () -> new ThreadPoolExecutor(
+ consumptionThreadCount,
+ consumptionThreadCount,
+ 60,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(),
+ new ThreadFactoryImpl("MessageConsumption",
this.getClientId().getIndex())));
this.inflightRequestCountInterceptor = new
InflightRequestCountInterceptor();
this.addMessageInterceptor(inflightRequestCountInterceptor);
@@ -606,7 +608,7 @@ class PushConsumerImpl extends ConsumerImpl implements
PushConsumer {
return getSettings().getRetryPolicy();
}
- public ThreadPoolExecutor getConsumptionExecutor() {
+ public ExecutorService getConsumptionExecutor() {
return consumptionExecutor;
}
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeService.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeService.java
index 828fb8e0..10da2fe6 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeService.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/StandardConsumeService.java
@@ -22,8 +22,8 @@ import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import java.util.List;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadPoolExecutor;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
import org.apache.rocketmq.client.java.hook.MessageInterceptor;
@@ -37,7 +37,7 @@ public class StandardConsumeService extends ConsumeService {
private static final Logger log =
LoggerFactory.getLogger(StandardConsumeService.class);
public StandardConsumeService(ClientId clientId, String consumerGroup,
MessageListener messageListener,
- ThreadPoolExecutor consumptionExecutor, MessageInterceptor
messageInterceptor,
+ ExecutorService consumptionExecutor, MessageInterceptor
messageInterceptor,
ScheduledExecutorService scheduler) {
super(clientId, consumerGroup, messageListener, consumptionExecutor,
messageInterceptor, scheduler);
}
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/misc/ExecutorServices.java
b/java/client/src/main/java/org/apache/rocketmq/client/java/misc/ExecutorServices.java
index cc7fde8c..5757b748 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/misc/ExecutorServices.java
+++
b/java/client/src/main/java/org/apache/rocketmq/client/java/misc/ExecutorServices.java
@@ -17,15 +17,150 @@
package org.apache.rocketmq.client.java.misc;
+import java.lang.reflect.Method;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class ExecutorServices {
+ private static final Logger log =
LoggerFactory.getLogger(ExecutorServices.class);
+ private static final Method NEW_VIRTUAL_THREAD_PER_TASK_EXECUTOR =
findVirtualThreadExecutorFactory();
+ private static final AtomicBoolean VIRTUAL_THREAD_FALLBACK_LOGGED = new
AtomicBoolean(false);
+
private ExecutorServices() {
}
+ /**
+ * Creates a virtual-thread-per-task executor when requested and supported
by the runtime. Reflection keeps the
+ * client binary compatible with Java 8 while allowing it to use the JDK
21 API when available.
+ */
+ public static ExecutorService newExecutorService(boolean
virtualThreadsEnabled,
+ Supplier<ExecutorService> platformExecutorSupplier) {
+ if (!virtualThreadsEnabled) {
+ return platformExecutorSupplier.get();
+ }
+ if (null == NEW_VIRTUAL_THREAD_PER_TASK_EXECUTOR) {
+ logVirtualThreadFallback(null);
+ return platformExecutorSupplier.get();
+ }
+ try {
+ return (ExecutorService)
NEW_VIRTUAL_THREAD_PER_TASK_EXECUTOR.invoke(null);
+ } catch (ReflectiveOperationException | RuntimeException e) {
+ logVirtualThreadFallback(e);
+ return platformExecutorSupplier.get();
+ }
+ }
+
+ /**
+ * Creates an executor which uses one virtual thread for each task when
requested and supported, while limiting the
+ * number of concurrently running tasks. A semaphore is used instead of
pooling virtual threads so tasks waiting for
+ * a permit do not occupy carrier threads.
+ */
+ public static ExecutorService newConcurrencyLimitedExecutorService(boolean
virtualThreadsEnabled,
+ int maxConcurrency, Supplier<ExecutorService>
platformExecutorSupplier) {
+ if (!virtualThreadsEnabled) {
+ return platformExecutorSupplier.get();
+ }
+ if (maxConcurrency <= 0) {
+ throw new IllegalArgumentException("maxConcurrency should be
positive");
+ }
+ return new ConcurrencyLimitedExecutorService(
+ newExecutorService(true, platformExecutorSupplier),
maxConcurrency);
+ }
+
+ static boolean isVirtualThreadSupported() {
+ return null != NEW_VIRTUAL_THREAD_PER_TASK_EXECUTOR;
+ }
+
+ private static Method findVirtualThreadExecutorFactory() {
+ try {
+ return
java.util.concurrent.Executors.class.getMethod("newVirtualThreadPerTaskExecutor");
+ } catch (NoSuchMethodException | SecurityException ignored) {
+ return null;
+ }
+ }
+
+ private static void logVirtualThreadFallback(Throwable t) {
+ if (!VIRTUAL_THREAD_FALLBACK_LOGGED.compareAndSet(false, true)) {
+ return;
+ }
+ if (null == t) {
+ log.warn("Virtual threads were enabled, but the runtime does not
provide them; falling back to platform "
+ + "threads");
+ return;
+ }
+ log.warn("Failed to create a virtual-thread executor; falling back to
platform threads", t);
+ }
+
@SuppressWarnings("BooleanMethodIsAlwaysInverted")
public static boolean awaitTerminated(ExecutorService executor) throws
InterruptedException {
return executor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
}
+
+ private static class ConcurrencyLimitedExecutorService extends
AbstractExecutorService {
+ private final ExecutorService delegate;
+ private final Semaphore semaphore;
+
+ private ConcurrencyLimitedExecutorService(ExecutorService delegate,
int maxConcurrency) {
+ this.delegate = delegate;
+ this.semaphore = new Semaphore(maxConcurrency, true);
+ }
+
+ @Override
+ public void shutdown() {
+ delegate.shutdown();
+ }
+
+ @Override
+ public List<Runnable> shutdownNow() {
+ return delegate.shutdownNow();
+ }
+
+ @Override
+ public boolean isShutdown() {
+ return delegate.isShutdown();
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return delegate.isTerminated();
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit) throws
InterruptedException {
+ return delegate.awaitTermination(timeout, unit);
+ }
+
+ @Override
+ public void execute(Runnable command) {
+ Objects.requireNonNull(command, "command");
+ delegate.execute(() -> {
+ boolean acquired = false;
+ try {
+ semaphore.acquire();
+ acquired = true;
+ command.run();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ if (command instanceof Future<?>) {
+ // AbstractExecutorService.submit() wraps tasks in a
FutureTask before execute(); cancel it so
+ // callers are not left waiting.
+ ((Future<?>) command).cancel(false);
+ }
+ } finally {
+ if (acquired) {
+ semaphore.release();
+ }
+ }
+ });
+ }
+ }
}
diff --git
a/java/client/src/main/java/org/apache/rocketmq/client/java/misc/ExecutorServices.java
b/java/client/src/test/java/org/apache/rocketmq/client/apis/ClientConfigurationTest.java
similarity index 53%
copy from
java/client/src/main/java/org/apache/rocketmq/client/java/misc/ExecutorServices.java
copy to
java/client/src/test/java/org/apache/rocketmq/client/apis/ClientConfigurationTest.java
index cc7fde8c..d523db84 100644
---
a/java/client/src/main/java/org/apache/rocketmq/client/java/misc/ExecutorServices.java
+++
b/java/client/src/test/java/org/apache/rocketmq/client/apis/ClientConfigurationTest.java
@@ -15,17 +15,29 @@
* limitations under the License.
*/
-package org.apache.rocketmq.client.java.misc;
+package org.apache.rocketmq.client.apis;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
+import org.junit.Assert;
+import org.junit.Test;
-public class ExecutorServices {
- private ExecutorServices() {
+public class ClientConfigurationTest {
+
+ @Test
+ public void testVirtualThreadsDisabledByDefault() {
+ ClientConfiguration configuration = ClientConfiguration.newBuilder()
+ .setEndpoints("localhost:8081")
+ .build();
+
+ Assert.assertFalse(configuration.isVirtualThreadsEnabled());
}
- @SuppressWarnings("BooleanMethodIsAlwaysInverted")
- public static boolean awaitTerminated(ExecutorService executor) throws
InterruptedException {
- return executor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
+ @Test
+ public void testEnableVirtualThreads() {
+ ClientConfiguration configuration = ClientConfiguration.newBuilder()
+ .setEndpoints("localhost:8081")
+ .enableVirtualThreads(true)
+ .build();
+
+ Assert.assertTrue(configuration.isVirtualThreadsEnabled());
}
}
diff --git
a/java/client/src/test/java/org/apache/rocketmq/client/java/misc/ExecutorServicesTest.java
b/java/client/src/test/java/org/apache/rocketmq/client/java/misc/ExecutorServicesTest.java
new file mode 100644
index 00000000..a0e65834
--- /dev/null
+++
b/java/client/src/test/java/org/apache/rocketmq/client/java/misc/ExecutorServicesTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.rocketmq.client.java.misc;
+
+import java.lang.reflect.Method;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ExecutorServicesTest {
+
+ @Test
+ public void testVirtualThreadsDisabled() throws Exception {
+ ExecutorService executor = ExecutorServices.newExecutorService(false,
Executors::newSingleThreadExecutor);
+ try {
+ Future<Boolean> future = executor.submit(() ->
isVirtual(Thread.currentThread()));
+ Assert.assertFalse(future.get());
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testVirtualThreadsEnabledWhenSupported() throws Exception {
+ AtomicBoolean platformExecutorCreated = new AtomicBoolean(false);
+ ExecutorService executor = ExecutorServices.newExecutorService(true,
() -> {
+ platformExecutorCreated.set(true);
+ return Executors.newSingleThreadExecutor();
+ });
+ try {
+ Future<Boolean> future = executor.submit(() ->
isVirtual(Thread.currentThread()));
+ Assert.assertEquals(ExecutorServices.isVirtualThreadSupported(),
future.get());
+ Assert.assertEquals(!ExecutorServices.isVirtualThreadSupported(),
platformExecutorCreated.get());
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testConcurrencyLimitedExecutorService() throws Exception {
+ ExecutorService executor =
ExecutorServices.newConcurrencyLimitedExecutorService(
+ true, 1, Executors::newCachedThreadPool);
+ CountDownLatch firstTaskStarted = new CountDownLatch(1);
+ CountDownLatch releaseFirstTask = new CountDownLatch(1);
+ CountDownLatch secondTaskStarted = new CountDownLatch(1);
+ try {
+ Future<Boolean> first = executor.submit(() -> {
+ firstTaskStarted.countDown();
+ releaseFirstTask.await();
+ return isVirtual(Thread.currentThread());
+ });
+ Assert.assertTrue(firstTaskStarted.await(5, TimeUnit.SECONDS));
+
+ Future<Boolean> second = executor.submit(() -> {
+ secondTaskStarted.countDown();
+ return isVirtual(Thread.currentThread());
+ });
+ Assert.assertFalse(secondTaskStarted.await(200,
TimeUnit.MILLISECONDS));
+
+ releaseFirstTask.countDown();
+ Assert.assertEquals(ExecutorServices.isVirtualThreadSupported(),
first.get(5, TimeUnit.SECONDS));
+ Assert.assertEquals(ExecutorServices.isVirtualThreadSupported(),
second.get(5, TimeUnit.SECONDS));
+ Assert.assertTrue(secondTaskStarted.await(5, TimeUnit.SECONDS));
+ } finally {
+ releaseFirstTask.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ private static boolean isVirtual(Thread thread) throws Exception {
+ Method method;
+ try {
+ method = Thread.class.getMethod("isVirtual");
+ } catch (NoSuchMethodException ignored) {
+ return false;
+ }
+ return (Boolean) method.invoke(thread);
+ }
+}
