增加 rabbitmq 示例

Author: YunaiV

lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/config/RabbitConfig.java

@@ -1,10 +1,8 @@
 package cn.iocoder.springboot.lab04.rabbitmqdemo.config;
 
 import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo01Message;
-import org.springframework.amqp.core.Binding;
-import org.springframework.amqp.core.BindingBuilder;
-import org.springframework.amqp.core.DirectExchange;
-import org.springframework.amqp.core.Queue;
+import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo02Message;
+import org.springframework.amqp.core.*;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -25,7 +23,7 @@ public Queue demo01Queue() {
                     false); // autoDelete: 是否自动删除
         }
 
-        // 创建 Exchange
+        // 创建 Direct Exchange
         @Bean
         public DirectExchange demo01Exchange() {
             return new DirectExchange(Demo01Message.EXCHANGE,
@@ -44,4 +42,38 @@ public Binding demo01Binding() {
 
     }
 
+    /**
+     * Topic Exchange 示例的配置类
+     */
+    public static class TopicExchangeDemoConfiguration {
+
+        // 创建 Queue
+        @Bean
+        public Queue demo02Queue() {
+            return new Queue(Demo02Message.QUEUE, // Queue 名字
+                    true, // durable: 是否持久化
+                    false, // exclusive: 是否排它
+                    false); // autoDelete: 是否自动删除
+        }
+
+        // 创建 Topic Exchange
+        @Bean
+        public TopicExchange demo02Exchange() {
+            return new TopicExchange(Demo02Message.EXCHANGE,
+                    true,  // durable: 是否持久化
+                    false);  // exclusive: 是否排它
+        }
+
+        // 创建 Binding
+        // Exchange：Demo02Message.EXCHANGE
+        // Routing key：Demo02Message.ROUTING_KEY
+        // Queue：Demo02Message.QUEUE
+        @Bean
+        public Binding demo02Binding() {
+            return BindingBuilder.bind(demo02Queue()).to(demo02Exchange()).with(Demo01Message.ROUTING_KEY);
+        }
+
+
+    }
+
 }

lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/consumer/Demo02Consumer.java

@@ -0,0 +1,21 @@
+package cn.iocoder.springboot.lab04.rabbitmqdemo.consumer;
+
+import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo02Message;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.rabbit.annotation.RabbitHandler;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+@Component
+@RabbitListener(queues = Demo02Message.QUEUE)
+public class Demo02Consumer {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @RabbitHandler
+    public void onMessage(Demo02Message message) {
+        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
+    }
+
+}

lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo01Message.java

@@ -8,7 +8,7 @@ public class Demo01Message implements Serializable {
 
     public static final String EXCHANGE = "EXCHANGE_DEMO_01";
 
-    public static final String ROUTING_KEY = "ROUTING_KEY_01";
+    public static final String ROUTING_KEY = "#.yu.nai";
 
     /**
      * 编号

lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/message/Demo02Message.java

@@ -0,0 +1,34 @@
+package cn.iocoder.springboot.lab04.rabbitmqdemo.message;
+
+import java.io.Serializable;
+
+public class Demo02Message implements Serializable {
+
+    public static final String QUEUE = "QUEUE_DEMO_02";
+
+    public static final String EXCHANGE = "EXCHANGE_DEMO_02";
+
+    public static final String ROUTING_KEY = "ROUTING_KEY_02";
+
+    /**
+     * 编号
+     */
+    private Integer id;
+
+    public Demo02Message setId(Integer id) {
+        this.id = id;
+        return this;
+    }
+
+    public Integer getId() {
+        return id;
+    }
+
+    @Override
+    public String toString() {
+        return "Demo02Message{" +
+                "id=" + id +
+                '}';
+    }
+
+}

lab-04/lab-04-rabbitmq-demo/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo02Producer.java

@@ -0,0 +1,22 @@
+package cn.iocoder.springboot.lab04.rabbitmqdemo.producer;
+
+import cn.iocoder.springboot.lab04.rabbitmqdemo.message.Demo02Message;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+public class Demo02Producer {
+
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+
+    public void syncSend(Integer id, String routingKey) {
+        // 创建 Demo02Message 消息
+        Demo02Message message = new Demo02Message();
+        message.setId(id);
+        // 同步发送消息
+        rabbitTemplate.convertAndSend(Demo02Message.EXCHANGE, routingKey, message);
+    }
+
+}

lab-04/lab-04-rabbitmq-demo/src/test/java/cn/iocoder/springboot/lab04/rabbitmqdemo/producer/Demo02ProducerTest.java

@@ -0,0 +1,44 @@
+package cn.iocoder.springboot.lab04.rabbitmqdemo.producer;
+
+import cn.iocoder.springboot.lab04.rabbitmqdemo.Application;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import java.util.concurrent.CountDownLatch;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = Application.class)
+public class Demo02ProducerTest {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Autowired
+    private Demo02Producer producer;
+
+    @Test
+    public void testSyncSendSuccess() throws InterruptedException {
+        int id = (int) (System.currentTimeMillis() / 1000);
+        producer.syncSend(id, "da.yu.nai");
+        logger.info("[testSyncSend][发送编号：[{}] 发送成功]", id);
+
+        // 阻塞等待，保证消费
+        new CountDownLatch(1).await();
+    }
+
+    @Test
+    public void testSyncSendFailure() throws InterruptedException {
+        int id = (int) (System.currentTimeMillis() / 1000);
+        producer.syncSend(id, "yu.nai.shuai");
+        logger.info("[testSyncSend][发送编号：[{}] 发送成功]", id);
+
+        // 阻塞等待，保证消费
+        new CountDownLatch(1).await();
+    }
+
+
+}