当前位置：首页 > news >正文

RabbitMQ-topic exchange使用方法

news 2025/8/18 14:15:31

RabbitMQ-默认读、写方式介绍

RabbitMQ-发布/订阅模式

RabbitMQ-直连交换机(direct)使用方法

1、概述

2、topic交换机使用方法

2.1 适用场景

2.2 解决方案

3、代码实现

3.1 源代码实现

3.2 运行记录

4、小结

1、概述

topic 交换机是比直连交换机功能更加强大的交换方式，通过不同的路由规则，可以实现fanout、direct两种交换机的功能。

2、topic交换机使用方法

2.1 适用场景

假设我们要对动物做一个描述，根据速度、颜色、种类等特征对其进行分别入到不同的mq队列中，routing key的格式为："<speed>.<colour>.<species>"，比如说，所有黄色动物入队列1，跑的速度慢的，还有小兔子入队列2，哪该如何实现该需求呢？

2.2 解决方案

结合2.1描述的需求，我们可以画出如下框图：

知识点解释：

* (star) ：和正则的功能类似，可以代表一整个单词。

# (hash) ：代表0个或者多个单词。

如果一条消息的routing key为「quick.orange.rabbit」，将会被同时路由到Q1和Q2，「lazy.orange.elephant」的routing key同样也将会被同时路由到Q1和Q2，「quick.orange.fox」的消息只会被路由Q1，【lazy.brown.fox】只会被路由到Q2，【lazy.pink.rabbit】只会被路由到Q2一次，虽然匹配了两个binding，【quick.brown.fox】没有匹配到任何的绑定，那么消息将会被丢弃。

如果一个队列绑定的是【#】，那么他将会接收到所有的消息，会忽略调binding key，实现类似扇形交换机的功能。

如果一个routing key中没有设计【#】和【*】，那么他会实现类似直连交换机的功能。

3、代码实现

3.1 源代码实现

生产者：

package mainimport ("fmt"amqp "github.com/rabbitmq/amqp091-go"
)func main() {conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")if err != nil {fmt.Println("Failed to connect to RabbitMQ")return}defer conn.Close()ch, err := conn.Channel()if err != nil {fmt.Println("Failed to open a channel")return}err = ch.ExchangeDeclare("logs_topic", // name"topic",      // typetrue,         // durablefalse,        // auto-deletedfalse,        // internalfalse,        // no-waitnil,          // arguments)if err != nil {fmt.Println("Failed to declare an exchange,err:", err)return}body := "Hello World by topic exchange"err = ch.Publish("logs_topic",       // exchange"quick.orange.fox", // routing keyfalse,false,amqp.Publishing{ContentType: "text/plain",Body:        []byte(body),})if err != nil {fmt.Println("Failed to publish a message")return}
}

代码示例中routing key为【quick.orange.fox】，这条消息将会被路由到2.2中的Q1队列中。

消费侧代码：

package mainimport ("fmt"amqp "github.com/rabbitmq/amqp091-go"
)func main() {conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")if err != nil {fmt.Println("Failed to connect to RabbitMQ")return}defer conn.Close()ch, err := conn.Channel()if err != nil {fmt.Println("Failed to open a channel")return}err = ch.ExchangeDeclare("logs", "direct", true, false, false, false, nil)if err != nil {fmt.Println("Failed to declare an exchange")return}q, err := ch.QueueDeclare("logs_topic", // nametrue,         // durablefalse,        // delete when unusedfalse,        // exclusivefalse,        // no-waitnil,          // arguments)err = ch.QueueBind(q.Name,       // queue name"*.orange.*", // routing key(binding key)"logs_topic", // exchangefalse,nil,)msgs, err := ch.Consume(q.Name, // queue"",     // consumertrue,   // auto-acktrue,   // exclusivefalse,  // no-localfalse,  // no-waitnil,    // args)var forever chan struct{}go func() {for d := range msgs {fmt.Printf(" [x] %s\n", d.Body)}}()fmt.Printf(" [*] Waiting for logs. To exit press CTRL+C")<-forever
}