add initialization value of rocketmq consumer

superxcv · May 31, 2018 · e5cda5b · e5cda5b
1 parent 0db55ea
commit e5cda5b
Show file tree

Hide file tree

Showing 6 changed files with 16 additions and 249 deletions.
diff --git a/...itease.uav.messaging/src/main/java/com/creditease/uav/messaging/api/MessagingFactory.java b/...itease.uav.messaging/src/main/java/com/creditease/uav/messaging/api/MessagingFactory.java
@@ -77,17 +77,19 @@ public static MessageProducer getMessageProducer(String producerId) {
      * @return
      */
     public static MessageConsumer createMessageConsumer(String name, String[] bizIDs, int maxConcurrent,
-            long stopInterval, MQFactory.QueueType queueType) {
+            int initConcurrent, long stopInterval, MQFactory.QueueType queueType) {
 
-        MessageConsumer consumer = new RMQMessageConsumer(name, bizIDs, maxConcurrent, stopInterval, queueType);
+        MessageConsumer consumer = new RMQMessageConsumer(name, bizIDs, maxConcurrent, initConcurrent, stopInterval,
+                queueType);
 
         return consumer;
     }
 
     public static MessageConsumer createMessageConsumer(String name, String[] bizIDs, int maxConcurrent,
-            long stopInterval) {
+            int initConcurrent, long stopInterval) {
 
-        return createMessageConsumer(name, bizIDs, maxConcurrent, stopInterval, MQFactory.QueueType.QUEUE);
+        return createMessageConsumer(name, bizIDs, maxConcurrent, initConcurrent, stopInterval,
+                MQFactory.QueueType.QUEUE);
     }
 
     /**

diff --git a/...ase.uav.messaging/src/main/java/com/creditease/uav/messaging/impl/RMQMessageConsumer.java b/...ase.uav.messaging/src/main/java/com/creditease/uav/messaging/impl/RMQMessageConsumer.java
@@ -48,8 +48,9 @@ public class RMQMessageConsumer extends MessageConsumer {
     private long stopInterval = 0;
 
     @SuppressWarnings("unchecked")
-    public RMQMessageConsumer(String name, String[] bizIds, int maxConcurrent, long stopInterval,
+    public RMQMessageConsumer(String name, String[] bizIds, int maxConcurrent, int initConcurrent, long stopInterval,
             MQFactory.QueueType queueType) {
+
         super(name, bizIds);
 
         this.context = MessagingContext.instance();
@@ -61,7 +62,7 @@ public RMQMessageConsumer(String name, String[] bizIds, int maxConcurrent, long
         // group不同的consumer不能重复
         config.setComsumerGroup(MessagingContext.DefaultConsumerGroup + "-" + this.getName());
         config.setConsumeThreadMax(maxConcurrent);
-        config.setConsumeThreadMin(1);
+        config.setConsumeThreadMin(initConcurrent);
         // 暂时使用默认值，调优的时候可能需要调整这个参数
         config.setPullBatchSize(0);
         // get bizId2TopicMap
@@ -79,9 +80,9 @@ public RMQMessageConsumer(String name, String[] bizIds, int maxConcurrent, long
         consumer = MQFactory.createMQConsumer(config, queueInfo);
     }
 
-    public RMQMessageConsumer(String name, String[] bizIds, int maxConcurrent, long stopInterval) {
+    public RMQMessageConsumer(String name, String[] bizIds, int maxConcurrent, int initConcurrent, long stopInterval) {
 
-        this(name, bizIds, maxConcurrent, stopInterval, MQFactory.QueueType.QUEUE);
+        this(name, bizIds, maxConcurrent, initConcurrent, stopInterval, MQFactory.QueueType.QUEUE);
     }
 
     @Override

diff --git a/com.creditease.uav.messaging/src/test/java/test/messaging/TestRMQMessageConsumer.java b/com.creditease.uav.messaging/src/test/java/test/messaging/TestRMQMessageConsumer.java
@@ -44,7 +44,7 @@ public static void main(String[] args) {
         bizIDs[0] = "test";
         String[] handlerNames = new String[1];
         handlerNames[0] = "test";
-        MessageConsumer consumer = MessagingFactory.createMessageConsumer("name", bizIDs, 10, 0);
+        MessageConsumer consumer = MessagingFactory.createMessageConsumer("name", bizIDs, 10, 10, 0);
         consumer.start();
     }
 }
diff --git a/...r/src/main/java/com/creditease/agent/feature/common/MessageProducerResourceComponent.java b/...r/src/main/java/com/creditease/agent/feature/common/MessageProducerResourceComponent.java
diff --git a/...src/main/java/com/creditease/agent/feature/common/messaging/StandardMessagingBuilder.java b/...src/main/java/com/creditease/agent/feature/common/messaging/StandardMessagingBuilder.java
diff --git a/...ommon/src/main/java/org/uavstack/resources/common/messaging/StandardMessagingBuilder.java b/...ommon/src/main/java/org/uavstack/resources/common/messaging/StandardMessagingBuilder.java
@@ -43,6 +43,7 @@
 public class StandardMessagingBuilder extends AbstractComponent {
 
     public StandardMessagingBuilder(String cName, String feature) {
+
         super(cName, feature);
     }
 
@@ -103,6 +104,8 @@ public MessageConsumer buildConsumer(String groupName, String msgType, MQFactory
 
         String ConsumeThreadMax = ConfigurationManager.getInstance().getFeatureConfiguration(feature,
                 msgType + ".consumethreadmax");
+        String ConsumeThreadInit = ConfigurationManager.getInstance().getFeatureConfiguration(feature,
+                msgType + ".consumethreadinit");
         String ConsumeStopInterval = ConfigurationManager.getInstance().getFeatureConfiguration(feature,
                 msgType + ".consumestopinterval");
 
@@ -117,6 +120,7 @@ public MessageConsumer buildConsumer(String groupName, String msgType, MQFactory
         bizIDs[0] = msgType;
         MessageConsumer consumer = MessagingFactory.createMessageConsumer(groupName, bizIDs,
                 ConsumeThreadMax == null ? 10 : Integer.parseInt(ConsumeThreadMax),
+                ConsumeThreadInit == null ? 10 : Integer.parseInt(ConsumeThreadInit),
                 ConsumeStopInterval == null ? 0 : Long.parseLong(ConsumeStopInterval), queueType);
 
         return consumer;