测试mq负载均衡

src/main/java/com/fdkankan/scene/controller/TestController.java

@@ -1,5 +1,9 @@
 package com.fdkankan.scene.controller;
 
+import com.fdkankan.rabbitmq.util.RabbitMqProducer;
+import com.fdkankan.web.response.ResultData;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
@@ -15,4 +19,15 @@ import org.springframework.web.bind.annotation.RestController;
 @RequestMapping("/test")
 public class TestController {
 
+    @Autowired
+    private RabbitMqProducer rabbitMqProducer;
+
+    @GetMapping("/test")
+    public ResultData test(){
+        for (int i = 0; i < 100; i++){
+            rabbitMqProducer.sendByWorkQueue("test_dsx", "{\"name\":123}");
+        }
+        return ResultData.ok();
+    }
+
 }

src/main/java/com/fdkankan/scene/listener/RabbitMqListener.java

@@ -53,6 +53,24 @@ public class RabbitMqListener {
         log.info("结束消费消息,id:{}", messageId);
     }
 
+    /**
+     * 开启了手动确认模式，如果没有手动确认，消费者不会重试，当服务重启时会再次消费，因为rabbitmq认为你还没有处理完你的业务
+     * queuesToDeclare = @Queue("${queue.modeling.modeling-test}"),  如果队列不不存在会自动创建队列
+     * concurrency = "3"    设置消费线程数，每个线程每次只拉取一条消息消费
+     */
+    @RabbitListener(
+        queuesToDeclare = @Queue("test_dsx")
+    )
+    public void test(Channel channel, Message message) throws Exception {
+        String messageId = message.getMessageProperties().getMessageId();
+        String msg = new String(message.getBody(), StandardCharsets.UTF_8);
+        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+        log.info("开始消费消息,id:{},queue:{},content:{}", messageId, "test_dsx", msg);
+        Thread.sleep(10000L);
+        log.info("deliverTag:" + message.getMessageProperties().getDeliveryTag());
+        log.info("结束消费消息,id:{}", messageId);
+    }
+