您当前的位置： 首页 > 慢生活 > 程序人生 网站首页程序人生

尚硅谷RabbitMQ视频-012-基础篇-工作模式-主题模式

发布时间：2026-10-25 23:07:51编辑：雪饮阅读（）

Topics
• Topic 类型与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型 Exchange 可以让队列在绑定 Routing key 的时候使用通配符
• Routingkey 一般都是由一个或多个单词组成，多个单词之间以 "." 分割，例如：item.insert
• 通配符规则：
  • #：匹配零个或多个词
  • *：匹配一个词

Topic模式生产者
Producer.java:
package org.example.topic;

import org.example.workQueues.util.ConnectionUtil;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Producer {

    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        String exchangeName = "test_topic";
        // 声明topic类型交换机
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC, true, false, false, null);

        String queue1Name = "test_topic_queue1";
        String queue2Name = "test_topic_queue2";

        // 声明队列
        channel.queueDeclare(queue1Name, true, false, false, null);
        channel.queueDeclare(queue2Name, true, false, false, null);

        // 绑定队列和交换机
        // 参数1. queue：队列名称
        // 参数2. exchange：交换机名称
        // 参数3. routingKey：路由键，绑定规则
        //      如果交换机的类型为fanout ,routingKey设置为""
        // routing key 常用格式: 系统的名称.日志的级别。
        // 需求：所有error级别的日志存入数据库，所有order系统的日志存入数据库
        channel.queueBind(queue1Name, exchangeName, "#.error");
        channel.queueBind(queue1Name, exchangeName, "order.*");
        channel.queueBind(queue2Name, exchangeName, "*.*");

        // 分别发送消息到队列：order.info、goods.info、goods.error
        String body = "[所在系统: order][日志级别: info][日志内容: 订单生成，保存成功]";
        channel.basicPublish(exchangeName, "order.info", null, body.getBytes());

//        body = "[所在系统: goods][日志级别: info][日志内容: 商品发布成功]";
//        channel.basicPublish(exchangeName, "goods.info", null, body.getBytes());
//
//        body = "[所在系统: goods][日志级别: error][日志内容: 商品发布失败]";
//        channel.basicPublish(exchangeName, "goods.error", null, body.getBytes());

        channel.close();
        connection.close();
    }

}


Topic模式消费者1
Consumer1.java:
package org.example.topic;

import java.io.IOException;

import org.example.workQueues.util.ConnectionUtil;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Consumer1 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        String QUEUE_NAME = "test_topic_queue1";
        // queueDeclare(队列名, 持久化, 独占, 自动删除, 参数)
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body: " + new String(body));
            }
        };
        // basicConsume(队列名, 自动ack, 消费者)
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }

}


Topic模式消费者2
Consumer2.java:
package org.example.topic;

import java.io.IOException;

import org.example.workQueues.util.ConnectionUtil;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Consumer2 {
public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        String QUEUE_NAME = "test_topic_queue2";
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body: " + new String(body));
            }
        };
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}


查看routing key（交换机角度）
运行生产者后可以在rabbitmq后台中的exchanges中找到当前交换机test_topic可以查看routing key

也可以从queues and streams中找到对应队列中进行查看
消费者及绑定（队列角度）

补充说明：
        ✅ 能匹配 #.error 的 routingKey 示例：
goods.error → # 匹配 goods（1 段）
order.error → # 匹配 order（1 段）
user.add.error → # 匹配 user.add（2 段）
a.b.c.error → # 匹配 a.b.c（3 段）
error → # 匹配 0 段，直接就是 error ✅（这个很容易被忽略！）