coder-zzzz opened a new issue #3312:
URL: https://github.com/apache/rocketmq/issues/3312
```java
public ConsumeMessageConcurrentlyService(DefaultMQPushConsumerImpl
defaultMQPushConsumerImpl,
MessageListenerConcurrently messageListener) {
this.defaultMQPushConsumerImpl = defaultMQPushConsumerImpl;
this.messageListener = messageListener;
this.defaultMQPushConsumer =
this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer();
this.consumerGroup = this.defaultMQPushConsumer.getConsumerGroup();
this.consumeRequestQueue = new LinkedBlockingQueue<Runnable>();
this.consumeExecutor = new ThreadPoolExecutor(
this.defaultMQPushConsumer.getConsumeThreadMin(),
this.defaultMQPushConsumer.getConsumeThreadMax(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.consumeRequestQueue,
new ThreadFactoryImpl("ConsumeMessageThread_"));
this.scheduledExecutorService =
Executors.newSingleThreadScheduledExecutor(new
ThreadFactoryImpl("ConsumeMessageScheduledThread_"));
this.cleanExpireMsgExecutors =
Executors.newSingleThreadScheduledExecutor(new
ThreadFactoryImpl("CleanExpireMsgScheduledThread_"));
}
```
See the code above, the consumeExecutor use a default capacity(Int.MaxValue)
LinkedBlockingQueue.If there is more than getConsumeThreadMin thread is working
,then new task will put in the queue, will never create new thread to process
new task.
--
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]
