RabbitMQ-默认读、写方式介绍
RabbitMQ-直连交换机(direct)使用方法
目录
在普通的生产者、消费者模式,rabbitmq会将消息依次传递给每一个消费者,一个worker一个,平均分配,这就是Round-robin调度方式,为了实现更加复杂的调度,我们就需要使用发布/订阅的方式。
RabbitMQ中,消息模型的核心理念就是,生产者从来不能直接将消息发送到队列,甚至生产者都不知道消息要被发送到队列中。
相反,生产者只能将消息发送到交换机中,交换机一侧从生产者接收消息,一侧将消息发送到队列中,交换机需要知道如何处理接收到的消息,是发送给一个队列还是多个队列?这是由交换机的类型决定的。
交换机共分为四类: direct
, topic
, headers
and fanout
. 本章节以扇形交换机为例说明rabbitmq的使用。
扇形交换机,就像你猜测的那样,他可以将他接收到的全部消息广播到所有队列里。
首先声明一个扇形交换机,type参数设置为『fanout』
- err = ch.ExchangeDeclare(
- "logs", // name
- "fanout", // type
- true, // durable
- false, // auto-deleted
- false, // internal
- false, // no-wait
- nil, // arguments
- )
交换机设定完成后,就可以往该交换机发送消息:
- body := "Hello World!"
- err = ch.Publish("logs", "", false, false, amqp.Publishing{
- ContentType: "text/plain",
- Body: []byte(body),
- })
如果要在rabbitmq的页面上查看发送的消息,需要提前创建一个队列,并绑定到该交换机[logs]上,就可以查看发送的消息:
扇形交换机的特性,就是他会将收到的消息广播给所有绑定到该交换机的队列,我们可以创建多个队列,并绑定到该交换机上,我们发送一次消息,就会看到,所有绑定到该交换机的队列中都会有一条消息,先创建三个队列,并分别绑定到logs交换机:
之后运行脚本,发送两次消息:
可以看到,三个队列当中都有两条消息。
- package main
-
- import (
- "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", "fanout", true, false, false, false, nil)
- if err != nil {
- fmt.Println("Failed to declare an exchange")
- return
- }
- body := "Hello World!"
- err = ch.Publish("logs", "", false, false, amqp.Publishing{
- ContentType: "text/plain",
- Body: []byte(body),
- })
- if err != nil {
- fmt.Println("Failed to publish a message")
- return
- }
- }
- q, err := ch.QueueDeclare(
- "", // name
- false, // durable
- false, // delete when unused
- true, // exclusive
- false, // no-wait
- nil, // arguments
- )
声明队列时,没有指定队列名称,这时,系统会返回一个随机名称存储在q变量中。
队列声明完成后,需要将该队列绑定到交换机上,这样交换机才能把消息广播给该队列:
绑定代码:
- err = ch.QueueBind(
- q.Name, // queue name
- "", // routing key
- "logs", // exchange
- false,
- nil,
- )
消费者侧全部代码如下:
- package main
-
- import (
- "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", "fanout", true, false, false, false, nil)
- if err != nil {
- fmt.Println("Failed to declare an exchange")
- return
- }
- q, err := ch.QueueDeclare(
- "", // name
- false, // durable
- false, // delete when unused
- true, // exclusive
- false, // no-wait
- nil, // arguments
- )
- err = ch.QueueBind(
- q.Name, // queue name
- "", // routing key
- "logs", // exchange
- false,
- nil,
- )
-
- msgs, err := ch.Consume(
- q.Name, // queue
- "", // consumer
- true, // auto-ack
- false, // exclusive
- false, // no-local
- false, // no-wait
- nil, // 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
- }
程序启动后,控制台上会增加一个随机命名的队列。
运行【3.2】的生产者程序,发送消息到扇形交换机,这个时候消费者就会同步消费到消息,并进行打印:
关于扇形交换机,核心的一点需要我们记住,发送到扇形交换机的消息,他会将消息广播给所有绑定到该交换机的队列上,无脑广播,所有队列会同时接受到交换机上全部的消息。