feat:主题信息队列,一个生产者对应一个交换机,一个交换机对应多个队列,direct交换机根据路由键routingKey进行信息转发,路由键routingKey支持通配符的格式

RabbitMQ-learn/src/main/java/space/anyi/rabbitMQ_learn/topic/EmitLogTopic.java

@@ -0,0 +1,61 @@
+package space.anyi.rabbitMQ_learn.topic;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import java.util.Scanner;
+/**
+ * @ProjectName: RabbitMQ-learn
+ * @FileName: EmitLogTopic
+ * @Author: 杨逸
+ * @Data:2025/10/9 9:45
+ * @Description:
+ */
+
+public class EmitLogTopic {
+    public static final Scanner input = new Scanner(System.in);
+    private static final String EXCHANGE_NAME = "topic_logs";
+    public static final String[]  queueNames = new String[]{"q1","q2"};
+    public static final String[]  routingKeys = new String[]{"*.orange.*","a.#"};
+
+
+    public static void main(String[] argv) throws Exception {
+        ConnectionFactory factory = new ConnectionFactory();
+        factory.setHost("localhost");
+        try (Connection connection = factory.newConnection();
+             Channel channel = connection.createChannel()) {
+
+            //创建一个topic类型的交换机
+            /**
+             * topic类型交换机的路由键是支持通配符(类Ant格式)的,一个单词有多个字符,单词与单词之间以点(.)分割
+             *  * :匹配一个单词
+             *  # :匹配零个或多个单词
+             *
+             *  e.g:
+             *  *.orange.* :匹配a.orange.b,也匹配abc.orange.abc,但不匹配aorangeb
+             *  a.# :匹配a.b.c,也匹配a.b,也匹配a,但不匹配ab
+             */
+            channel.exchangeDeclare(EXCHANGE_NAME, "topic");
+            for (int i = 0; i < queueNames.length; i++) {
+                String queueName = queueNames[i];
+                String routingKey = routingKeys[i];
+                //创建信息队列
+                channel.queueDeclare(queueName, false, false, false, null);
+                //将队列绑定到交换机
+                channel.queueBind(queueName, EXCHANGE_NAME, routingKey);
+            }
+
+            System.out.println("请输入路由键:");
+            while (input.hasNext()) {
+                String routingKey = input.nextLine();
+                System.out.println("请输入消息:");
+                String message = input.nextLine();
+                //生产信息
+                channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
+                System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
+                System.out.println("请输入路由键:");
+            }
+
+        }
+    }
+}

RabbitMQ-learn/src/main/java/space/anyi/rabbitMQ_learn/topic/ReceiveLogsTopic.java

@@ -0,0 +1,41 @@
+package space.anyi.rabbitMQ_learn.topic;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.DeliverCallback;
+
+/**
+ * @ProjectName: RabbitMQ-learn
+ * @FileName: ReceiveLogsTopic
+ * @Author: 杨逸
+ * @Data:2025/10/9 10:05
+ * @Description:
+ */
+
+public class ReceiveLogsTopic {
+
+    private static final String EXCHANGE_NAME = "topic_logs";
+    public static final String[]  queueNames = new String[]{"q1","q2"};
+
+    public static void main(String[] argv) throws Exception {
+        ConnectionFactory factory = new ConnectionFactory();
+        factory.setHost("localhost");
+        Connection connection = factory.newConnection();
+        Channel channel = connection.createChannel();
+        //声明交换机
+        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
+        //获取信息队列的名称
+        //String queueName = channel.queueDeclare().getQueue();
+
+        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
+            String message = new String(delivery.getBody(), "UTF-8");
+            System.out.println(" [x] Received '" +
+                    delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
+        };
+        //消费信息
+        for (String queueName : queueNames) {
+            channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
+        }
+    }
+}