大家好,欢迎来到IT知识分享网。
publish/subscribe
发布订阅模式中,生产者不再直接与队列绑定,而是将数据发送至交换机Exchange
交换机Exchange用于将数据按某种规则送入与之绑定的队列,进而供消费者使用。
发布订阅模式中,交换机将无差别的将所有消息送入与之绑定的队列,所有消费者拿到的消息完全相同,交换机的类型为fanout
exchange的类型有
- direct
- topic
- headers
- fanout
生产者
package main
import (
"log"
"os"
"rabbit/utils"
"strings"
"github.com/streadway/amqp"
)
func main() {
ch, _ := utils.GetRabbitMQChannel()
_ = ch.ExchangeDeclare(
"logs", // name
"fanout", // type
true, // durable
false, // auto-deleted
false, // internal
false, // no-wait
nil, // arguments
)
body := bodyFrom(os.Args)
_ = ch.Publish(
"logs", // exchange
"", // routing key
false, // mandatory
false, // immediate
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(body),
})
log.Printf(" [x] Sent %s", body)
}
func bodyFrom(args []string) string {
var s string
if (len(args) < 2) || os.Args[1] == "" {
s = "hello"
} else {
s = strings.Join(args[1:], " ")
}
return s
}
消费者
package main
import (
"log"
"rabbit/utils"
)
func main() {
ch, _ := utils.GetRabbitMQChannel()
_ = ch.ExchangeDeclare(
"logs", // name
"fanout", // type
true, // durable
false, // auto-deleted
false, // internal
false, // no-wait
nil, // arguments
)
q, _ := ch.QueueDeclare(
"", // name
false, // durable
false, // delete when unused
true, // exclusive
false, // no-wait
nil, // arguments
)
_ = ch.QueueBind(
q.Name, // queue name
"", // routing key
"logs", // exchange
false,
nil,
)
msgs, _ := ch.Consume(
q.Name, // queue
"", // consumer
true, // auto-ack
false, // exclusive
false, // no-local
false, // no-wait
nil, // args
)
forever := make(chan bool)
go func() {
for d := range msgs {
log.Printf(" [x] %s", d.Body)
}
}()
log.Printf(" [*] Waiting for logs. To exit press CTRL+C")
<-forever
}
开启的两个消费者收到了相同的消息
临时队列
队列实例包含RabbitMQ生成的随机队列名称, 当声明它的连接关闭时,队列将被删除,因为它被声明为exclusive
。
免责声明:本站所有文章内容,图片,视频等均是来源于用户投稿和互联网及文摘转载整编而成,不代表本站观点,不承担相关法律责任。其著作权各归其原作者或其出版社所有。如发现本站有涉嫌抄袭侵权/违法违规的内容,侵犯到您的权益,请在线联系站长,一经查实,本站将立刻删除。 本文来自网络,若有侵权,请联系删除,如若转载,请注明出处:https://yundeesoft.com/24273.html