demo

src/main/java/com/example/demo/mq/ConsumerConfiguration.java

@@ -9,6 +9,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.stereotype.Component;
@@ -33,6 +34,30 @@ public class ConsumerConfiguration {
         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
         //订阅主题和 标签（ * 代表所有标签)下信息
         consumer.subscribe("miyapaytest", "*");
+
+        //设置多线程消费
+        /**
+         * 1.rocketMQ用的是ThreadPoolExecutor
+         *
+         * 2.ThreadPoolExecutor只有当任务队列满了才会启动大于核心线程数的新线程
+         *
+         * 3.rocketMQ传给ThreadPoolExecutor的队列是没设置容量的无界队列
+         *
+         * 4.所以任务队列基本不会满
+         *
+         * 5.所以ThreadPoolExecutor基本不会新建大于核心线程数的线程
+         *
+         * 6.所以rocketMQ设置的ConsumeThreadMax是无效的
+         */
+        consumer.setConsumeThreadMax(50);
+        consumer.setConsumeThreadMin(50);
+
+        /**
+         * 设置广播模式加上这行代码
+         */
+//        consumer.setMessageModel(MessageModel.BROADCASTING);
+
+
         consumer.registerMessageListener(ledgerConsumerListener);
         log.info("消费者启动成功");
         return consumer;