[ISSUE apache#8695] fix DefaultLitePullConsumer PullThreadNums Parame…

…ter not effective. (apache#8696)
luozongle01 · Sep 14, 2024 · f872b68 · f872b68
1 parent 57d9f8d
commit f872b68
Showing 1 changed file with 9 additions and 4 deletions.
diff --git a/...t/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java b/...t/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
@@ -164,10 +164,6 @@ private enum SubscriptionType {
     public DefaultLitePullConsumerImpl(final DefaultLitePullConsumer defaultLitePullConsumer, final RPCHook rpcHook) {
         this.defaultLitePullConsumer = defaultLitePullConsumer;
         this.rpcHook = rpcHook;
-        this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(
-            this.defaultLitePullConsumer.getPullThreadNums(),
-            new ThreadFactoryImpl("PullMsgThread-" + this.defaultLitePullConsumer.getConsumerGroup())
-        );
         this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("MonitorMessageQueueChangeThread"));
         this.pullTimeDelayMillsWhenException = defaultLitePullConsumer.getPullTimeDelayMillsWhenException();
     }
@@ -293,6 +289,8 @@ public synchronized void start() throws MQClientException {
                     this.defaultLitePullConsumer.changeInstanceNameToPID();
                 }
 
+                initScheduledThreadPoolExecutor();
+
                 initMQClientFactory();
 
                 initRebalanceImpl();
@@ -324,6 +322,13 @@ public synchronized void start() throws MQClientException {
         }
     }
 
+    private void initScheduledThreadPoolExecutor() {
+        this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(
+                this.defaultLitePullConsumer.getPullThreadNums(),
+                new ThreadFactoryImpl("PullMsgThread-" + this.defaultLitePullConsumer.getConsumerGroup())
+        );
+    }
+
     private void initMQClientFactory() throws MQClientException {
         this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultLitePullConsumer, this.rpcHook);
         boolean registerOK = mQClientFactory.registerConsumer(this.defaultLitePullConsumer.getConsumerGroup(), this);