Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[ISSUE #8354] fix the consumer's threadpool cannot be expanded to the maximum number of threads. #8355

Open
wants to merge 13 commits into
base: develop
Choose a base branch
from
Open

[ISSUE #8354] fix the consumer's threadpool cannot be expanded to the maximum number of threads. #8355

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
cf28b9e
remove responseFuture when exception
CLFutureX Aug 2, 2023
ae1d6fa
Empty-Commit
CLFutureX Aug 3, 2023
b0e4027
Merge branch 'develop' into PiteChen_fix_7111
CLFutureX Aug 25, 2023
8379354
Merge branch 'PiteChen_fix_7111' into develop
CLFutureX Aug 26, 2023
8a14c2f
Merge branch 'apache:develop' into develop
CLFutureX Sep 6, 2023
3c03c89
Merge branch 'apache:develop' into develop
CLFutureX Sep 16, 2023
bfbf575
Merge branch 'apache:develop' into develop
CLFutureX Sep 23, 2023
216bc13
Merge branch 'apache:develop' into develop
CLFutureX Sep 27, 2023
2990861
Merge branch 'apache:develop' into develop
CLFutureX Oct 28, 2023
a7b82b6
Merge branch 'apache:develop' into develop
CLFutureX May 20, 2024
bbe87f5
Merge branch 'apache:develop' into develop
CLFutureX Jul 2, 2024
fb3201e
[ISSUE #8354] fix the consumer's threadpool cannot be expanded to the…
CLFutureX Jul 3, 2024
5c03509
code optimize
CLFutureX Jul 3, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 9 additions & 7 deletions ...main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,16 +67,18 @@ public ConsumeMessageConcurrentlyService(DefaultMQPushConsumerImpl defaultMQPush

this.defaultMQPushConsumer = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer();
this.consumerGroup = this.defaultMQPushConsumer.getConsumerGroup();
this.consumeRequestQueue = new LinkedBlockingQueue<>();
int subTopics = Math.max(this.defaultMQPushConsumerImpl.getSubscriptionInner().size(), 1);
int queueSize = this.defaultMQPushConsumer.getPullThresholdForQueue() * subTopics;
this.consumeRequestQueue = new LinkedBlockingQueue<>(queueSize);

String consumerGroupTag = (consumerGroup.length() > 100 ? consumerGroup.substring(0, 100) : consumerGroup) + "_";
this.consumeExecutor = new ThreadPoolExecutor(
this.defaultMQPushConsumer.getConsumeThreadMin(),
this.defaultMQPushConsumer.getConsumeThreadMax(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.consumeRequestQueue,
new ThreadFactoryImpl("ConsumeMessageThread_" + consumerGroupTag));
this.defaultMQPushConsumer.getConsumeThreadMin(),
this.defaultMQPushConsumer.getConsumeThreadMax(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.consumeRequestQueue,
new ThreadFactoryImpl("ConsumeMessageThread_" + consumerGroupTag));

this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("ConsumeMessageScheduledThread_" + consumerGroupTag));
this.cleanExpireMsgExecutors = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("CleanExpireMsgScheduledThread_" + consumerGroupTag));
Expand Down
16 changes: 9 additions & 7 deletions ...n/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessagePopConcurrentlyService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,15 +68,17 @@ public ConsumeMessagePopConcurrentlyService(DefaultMQPushConsumerImpl defaultMQP

this.defaultMQPushConsumer = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer();
this.consumerGroup = this.defaultMQPushConsumer.getConsumerGroup();
this.consumeRequestQueue = new LinkedBlockingQueue<>();
int subTopics = Math.max(this.defaultMQPushConsumerImpl.getSubscriptionInner().size(), 1);
int queueSize = this.defaultMQPushConsumer.getPullThresholdForQueue() * subTopics;
this.consumeRequestQueue = new LinkedBlockingQueue<>(queueSize);

this.consumeExecutor = new ThreadPoolExecutor(
this.defaultMQPushConsumer.getConsumeThreadMin(),
this.defaultMQPushConsumer.getConsumeThreadMax(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.consumeRequestQueue,
new ThreadFactoryImpl("ConsumeMessageThread_"));
this.defaultMQPushConsumer.getConsumeThreadMin(),
this.defaultMQPushConsumer.getConsumeThreadMax(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.consumeRequestQueue,
new ThreadFactoryImpl("ConsumeMessageThread_"));

this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("ConsumeMessageScheduledThread_"));
}
Expand Down
Loading