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

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

news 来源：原创 2024/9/29 17:23:58

RabbitMQ-默认读、写方式介绍

RabbitMQ-发布/订阅模式

1、概述

2、直连交换机

3、多重绑定

4、具体代码实现

4.1 生产者部分

4.2 消费者部分

5、运行代码

6、总结

1、概述

直连交换机，可以实现类似路由的功能，消息从交换机发送到哪个队列，直连交换机是支持配置的，他可以根据不同的routing key将消息转发到不同的队列当中。

在上一篇《RabbitMQ-发布/订阅模式》中，介绍过绑定过程，类似：

err = ch.QueueBind(q.Name, // queue name"",     // routing key"logs", // exchangefalse,nil)

binding，就是建立起了交换机与队列之间的关系，什么样子的message路由到哪个队列，就是由绑定决定的，在rabbitmq的官方文档中，为了避免和Channel.Publish函数的key参数混淆，在bind函数中的routing key称之为binding key，比如：

err = ch.QueueBind(q.Name,    // queue name"black",   // binding key"logs",    // exchangefalse,nil)

在上面的代码中，routing key参数，在扇形交换机是无效的，这点大家要注意。

2、直连交换机

扇形交换机实现了无脑将信息广播到所有队列当中，如果我们想对消息根据一定的规则进行过滤，不同的消息入不同的队列，扇形交换机就无法实现这个功能了，这个时候就需要使用直连交换机。

上图，声明了直连交换机，并将两个队列绑定到该交换机上，第一个队列的binding key为【orange】，第二个队列设计了两个绑定，第一个binding key为【black】，另外一个为【green】，在这种设计下，routing key为【orange】的消息将会被路由到Q1队列，routing key为【black】【green】的消息将会被路由到Q2队列，其他类型的消息就会被丢弃。

3、多重绑定

在这种模式下，其实现的功能类似扇形交换机，交换机可以将同一个消息路由到多个队列当中。

在上图的设计方式中，routing key为【black】的消息会同时路由到Q1和Q2两个队列中。

4、具体代码实现

4.1 生产者部分

第一步，和扇形交换机一样，声明交换机：

err = ch.ExchangeDeclare("logs_direct", // name"direct",      // typetrue,          // durablefalse,         // auto-deletedfalse,         // internalfalse,         // no-waitnil,           // arguments
)

第二步，发送消息：

	body := "Hello World by dircet exchange"err = ch.Publish("logs", // exchange"info", // routing keyfalse,false,amqp.Publishing{ContentType: "text/plain",Body:        []byte(body),})

4.2 消费者部分

声明队列：

	q, err := ch.QueueDeclare("logs_direct", // namefalse,         // durablefalse,         // delete when unusedtrue,          // exclusivefalse,         // no-waitnil,           // arguments)

绑定：

err = ch.QueueBind(q.Name, // queue name"info", // routing key(binding key)"logs", // exchangefalse,nil,)

5、运行代码

生产者部分全部代码：

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",   // exchange name"direct", // exchange typetrue,false,false,false,nil)if err != nil {fmt.Println("Failed to declare an exchange")return}body := "Hello World by dircet exchange"err = ch.Publish("logs", // exchange"info", // routing keyfalse,false,amqp.Publishing{ContentType: "text/plain",Body:        []byte(body),})if err != nil {fmt.Println("Failed to publish a message")return}
}

消费者部分全部代码：

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_direct", // namefalse,         // durablefalse,         // delete when unusedtrue,          // exclusivefalse,         // no-waitnil,           // arguments)err = ch.QueueBind(q.Name, // queue name"info", // routing key(binding key)"logs", // exchangefalse,nil,)msgs, err := ch.Consume(q.Name, // queue"",     // consumertrue,   // auto-ackfalse,  // 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
}

启动消费者，程序启动后，从RabbitMQ控制台就会看到一个队列：