[ISSUE #429]Use 'consumeThreadNumber' instead of 'consumeThreadMax' (#431)

Co-authored-by: joewee <371499220@qq.com>

Author: joewee

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListener.java

@@ -22,6 +22,7 @@
 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)
@@ -72,9 +73,19 @@
 
     /**
      * 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>
      */
+    @Deprecated
     int consumeThreadMax() default 64;
 
+    /**
+     * consumer thread number.
+     */
+    int consumeThreadNumber() default 20;
+
     /**
      * Max re-consume times.
      *

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java

@@ -105,6 +105,8 @@ public class DefaultRocketMQListenerContainer implements InitializingBean,
 
     private int consumeThreadMax = 64;
 
+    private int consumeThreadNumber = 20;
+
     private String charset = "UTF-8";
 
     private MessageConverter messageConverter;
@@ -186,6 +188,10 @@ public int getConsumeThreadMax() {
         return consumeThreadMax;
     }
 
+    public int getConsumeThreadNumber() {
+        return consumeThreadNumber;
+    }
+
     public String getCharset() {
         return charset;
     }
@@ -227,7 +233,8 @@ public void setRocketMQMessageListener(RocketMQMessageListener anno) {
         this.rocketMQMessageListener = anno;
 
         this.consumeMode = anno.consumeMode();
-        this.consumeThreadMax = anno.consumeThreadMax();
+        this.consumeThreadMax = anno.consumeThreadNumber();
+        this.consumeThreadNumber = anno.consumeThreadNumber();
         this.messageModel = anno.messageModel();
         this.selectorType = anno.selectorType();
         this.selectorExpression = anno.selectorExpression();
@@ -612,10 +619,9 @@ private void initRocketMQPushConsumer() throws MQClientException {
         if (accessChannel != null) {
             consumer.setAccessChannel(accessChannel);
         }
-        consumer.setConsumeThreadMax(consumeThreadMax);
-        if (consumeThreadMax < consumer.getConsumeThreadMin()) {
-            consumer.setConsumeThreadMin(consumeThreadMax);
-        }
+        //set the consumer core thread number and maximum thread number has the same value
+        consumer.setConsumeThreadMax(consumeThreadNumber);
+        consumer.setConsumeThreadMin(consumeThreadNumber);
         consumer.setConsumeTimeout(consumeTimeout);
         consumer.setMaxReconsumeTimes(maxReconsumeTimes);
         switch (messageModel) {

rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainerTest.java

@@ -244,7 +244,8 @@ public void testSetRocketMQMessageListener() {
         container.setRocketMQMessageListener(anno);
 
         assertEquals(anno.consumeMode(), container.getConsumeMode());
-        assertEquals(anno.consumeThreadMax(), container.getConsumeThreadMax());
+        assertEquals(anno.consumeThreadNumber(), container.getConsumeThreadMax());
+        assertEquals(anno.consumeThreadNumber(), container.getConsumeThreadNumber());
         assertEquals(anno.messageModel(), container.getMessageModel());
         assertEquals(anno.selectorType(), container.getSelectorType());
         assertEquals(anno.selectorExpression(), container.getSelectorExpression());
@@ -256,7 +257,7 @@ public void testSetRocketMQMessageListener() {
 
     @RocketMQMessageListener(consumerGroup = "abc1", topic = "test",
             consumeMode = ConsumeMode.ORDERLY,
-            consumeThreadMax = 3456,
+            consumeThreadNumber = 3456,
             messageModel = MessageModel.BROADCASTING,
             selectorType = SelectorType.SQL92,
             selectorExpression = "selectorExpression",