RockteMQ-AI commented on code in PR #1319:
URL: https://github.com/apache/rocketmq-clients/pull/1319#discussion_r3728057109
##########
java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java:
##########
@@ -152,13 +153,14 @@ public PushConsumerImpl(ClientConfiguration
clientConfiguration, String consumer
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(
Review Comment:
The `getConsumptionExecutor()` return type changed from `ThreadPoolExecutor`
to `ExecutorService`. Since `PushConsumerImpl` is package-private this is safe
within the module, but if any downstream code (e.g., rocketmq-dashboard or
operators) reflectively accesses this method, it could break. Worth a brief
mention in the PR description or commit message for release notes.
##########
java/client/src/main/java/org/apache/rocketmq/client/java/misc/ExecutorServices.java:
##########
@@ -17,15 +17,148 @@
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) {
Review Comment:
The `command instanceof Future<?>` check in the `InterruptedException`
handler is a thoughtful touch — it correctly handles the case where
`AbstractExecutorService.submit()` wraps the callable in a `FutureTask` before
calling `execute()`. Consider adding a brief inline comment explaining why this
check exists, since it's not immediately obvious that `execute()` can receive
`FutureTask` instances from the inherited `submit()` methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
