下载视频

yzg-util-mq/src/main/java/com/yanzuoguang/mq/service/MessageSendService.java

@@ -77,4 +77,16 @@ public interface MessageSendService {
      */
     @ApiOperation(value = "动态初始化消息队列处理")
     SimpleMessageListenerContainer init(String queueName, int concurrency, ChannelAwareMessageListener messageListener);
+
+    /**
+     * 动态初始化消息队列处理
+     *
+     * @param queueName       队列名字
+     * @param concurrency     线程数量
+     * @param maxConcurrency  最大线程数量个
+     * @param messageListener 消息处理函数
+     * @return
+     */
+    @ApiOperation(value = "动态初始化消息队列处理")
+    SimpleMessageListenerContainer init(String queueName, int concurrency, int maxConcurrency, ChannelAwareMessageListener messageListener);
 }

yzg-util-mq/src/main/java/com/yanzuoguang/mq/service/impl/MessageSendServiceImpl.java

@@ -176,16 +176,26 @@ public class MessageSendServiceImpl implements MessageSendService {
      */
     @Override
     public SimpleMessageListenerContainer init(String queueName, int concurrency, ChannelAwareMessageListener messageListener) {
+        return this.init(queueName, concurrency, concurrency, messageListener);
+    }
+
+    /**
+     * 动态初始化消息队列处理
+     *
+     * @param queueName       队列名字
+     * @param concurrency     线程数量
+     * @param maxConcurrency  最大线程数量个
+     * @param messageListener 消息处理函数
+     * @return
+     */
+    @Override
+    public SimpleMessageListenerContainer init(String queueName, int concurrency, int maxConcurrency, ChannelAwareMessageListener messageListener) {
         SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
         container.setConnectionFactory(connectionFactory);
         container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
         container.setQueueNames(queueName);
-        if (concurrency > 0) {
-            container.setConcurrentConsumers(concurrency);
-        } else {
-            container.setConcurrentConsumers(this.mqConfig.getConcurrency());
-        }
-        container.setMaxConcurrentConsumers(this.mqConfig.getMaxConcurrency());
+        container.setConcurrentConsumers(StringHelper.getFirst(concurrency, this.mqConfig.getConcurrency()));
+        container.setMaxConcurrentConsumers(StringHelper.getFirst(maxConcurrency, this.mqConfig.getMaxConcurrency()));
         container.setPrefetchCount(this.mqConfig.getPrefetch());
         container.setTxSize(this.mqConfig.getTxSize());
         container.setMessageListener(new MessageListenerAdapter(messageListener));