[rocketmq-spring] branch master updated: [ISSUE #546] Fix ConsumeMessageService#updateCorePoolSize" not support increase coreSize (#547)

[jinrongtong]

This is an automated email from the ASF dual-hosted git repository.

jinrongtong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-spring.git


The following commit(s) were added to refs/heads/master by this push:
     new 4147535  [ISSUE #546] Fix ConsumeMessageService#updateCorePoolSize" 
not support increase coreSize (#547)
4147535 is described below

commit 41475359a36cdc073b05883d8bc659ebf9a0aba9
Author: Cy <40804675+crzb...@users.noreply.github.com>
AuthorDate: Fri Jun 2 15:42:35 2023 +0800

    [ISSUE #546] Fix ConsumeMessageService#updateCorePoolSize" not support 
increase coreSize (#547)
    
    * Fix ConsumeMessageService#updateCorePoolSize" not support increase 
coreSize
    
    * Fix ConsumeMessageService#updateCorePoolSize" not support increase 
coreSize
    
    ---------
    
    Co-authored-by: Cy <dingchen...@hydee.cn>
---
 .../rocketmq/spring/annotation/RocketMQMessageListener.java    | 10 ++++------
 .../spring/support/DefaultRocketMQListenerContainer.java       |  6 +++---
 .../spring/support/DefaultRocketMQListenerContainerTest.java   |  2 +-
 3 files changed, 8 insertions(+), 10 deletions(-)

diff --git 
a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListener.java
 
b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListener.java
index 3647864..302d276 100644
--- 
a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListener.java
+++ 
b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListener.java
@@ -17,12 +17,12 @@
 
 package org.apache.rocketmq.spring.annotation;
 
+import org.apache.rocketmq.client.impl.consumer.ConsumeMessageService;
 import java.lang.annotation.Documented;
 import java.lang.annotation.ElementType;
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
-import java.util.concurrent.LinkedBlockingQueue;
 
 @Target(ElementType.TYPE)
 @Retention(RetentionPolicy.RUNTIME)
@@ -73,12 +73,10 @@ public @interface RocketMQMessageListener {
 
     /**
      * Max consumer thread number.
-     * @deprecated This property is not work well, because the consumer thread 
pool executor use
-     * {@link LinkedBlockingQueue} with default capacity bound 
(Integer.MAX_VALUE), use
-     * {@link RocketMQMessageListener#consumeThreadNumber} .
-     * @see <a 
href="https://github.com/apache/rocketmq-spring/issues/429";>issues#429</a>
+     * This property control consumer thread pool executor maximumPoolSize see
+     * {@link ConsumeMessageService#updateCorePoolSize(int)}
+     * @see <a 
href="https://github.com/apache/rocketmq-spring/issues/546";>issues#546</a>
      */
-    @Deprecated
     int consumeThreadMax() default 64;
 
     /**
diff --git 
a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java
 
b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java
index 0233a61..910e9af 100644
--- 
a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java
+++ 
b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java
@@ -236,7 +236,7 @@ public class DefaultRocketMQListenerContainer implements 
InitializingBean,
         this.rocketMQMessageListener = anno;
 
         this.consumeMode = anno.consumeMode();
-        this.consumeThreadMax = anno.consumeThreadNumber();
+        this.consumeThreadMax = anno.consumeThreadMax();
         this.consumeThreadNumber = anno.consumeThreadNumber();
         this.messageModel = anno.messageModel();
         this.selectorType = anno.selectorType();
@@ -641,8 +641,8 @@ public class DefaultRocketMQListenerContainer implements 
InitializingBean,
         if (accessChannel != null) {
             consumer.setAccessChannel(accessChannel);
         }
-        //set the consumer core thread number and maximum thread number has 
the same value
-        consumer.setConsumeThreadMax(consumeThreadNumber);
+
+        consumer.setConsumeThreadMax(consumeThreadMax);
         consumer.setConsumeThreadMin(consumeThreadNumber);
         consumer.setConsumeTimeout(consumeTimeout);
         consumer.setMaxReconsumeTimes(maxReconsumeTimes);
diff --git 
a/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainerTest.java
 
b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainerTest.java
index f7cd9cf..de15fcd 100644
--- 
a/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainerTest.java
+++ 
b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainerTest.java
@@ -244,7 +244,7 @@ public class DefaultRocketMQListenerContainerTest {
         container.setRocketMQMessageListener(anno);
 
         assertEquals(anno.consumeMode(), container.getConsumeMode());
-        assertEquals(anno.consumeThreadNumber(), 
container.getConsumeThreadMax());
+        assertEquals(anno.consumeThreadMax(), container.getConsumeThreadMax());
         assertEquals(anno.consumeThreadNumber(), 
container.getConsumeThreadNumber());
         assertEquals(anno.messageModel(), container.getMessageModel());
         assertEquals(anno.selectorType(), container.getSelectorType());