快速建设网站网站推广服务外包
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控制台就会看到一个队列：
之后运行生产者部分代码，生产者发送消息后，消费者侧就会接收到生产者发来的消息：
6、总结
以上就是rabbitmq直连交换机的使用方式，示例代码只是做了简单的演示，对于多重绑定，各种路由规则可以自行尝试，直连交换机模式，为开发者提供了灵活的路由规则，推荐使用。