“发布／订阅”模式，将分发一个消息给多个消费者（consumers）。
image.png

交换机（Exchanges）

前面的教程中，我们发送消息到队列并从中取出消息。现在是时候介绍RabbitMQ中完整的消息模型了。
让我们简单的概括一下之前的教程：

  • 发布者（producer）是发布消息的应用程序。
  • 队列（queue）用于消息存储的缓冲。
  • 消费者（consumer）是接收消息的应用程序。

RabbitMQ消息模型的核心理念是：发布者（producer）不会直接发送任何消息给队列。事实上，发布者（producer）甚至不知道消息是否已经被投递到队列。
发布者（producer）只需要把消息发送给一个交换机（exchange）。交换机非常简单，它一边从发布者方接收消息，一边把消息推送到队列。交换机必须知道如何处理它接收到的消息，是应该推送到指定的队列还是是多个队列，或者是直接忽略消息。这些规则是通过交换机类型（exchange type）来定义的。

交换机类型

有几个可供选择的交换机类型：direct, topic, headers和fanout。

实现

  2. func NewRabbitMQPubSub(exchangeName string) *RabbitMQ {
  3.     rmq := NewRabbitMQ("", exchangeName, "")
  4.     var err error
  5.     // 获取 connection
  6.     rmq.conn, err = amqp.Dial(rmq.MqUrl)
  7.     rmq.failOnErr(err, "创建连接错误")
  8.     rmq.channel, err = rmq.conn.Channel()
  9.     rmq.failOnErr(err, "创建连接错误")
  10.     return rmq
  11. }
  13. // 订阅生成模式
  14. func (r *RabbitMQ) PublishPub(message string) {
  15.     // 1.尝试创建交换机
  16.     err := r.channel.ExchangeDeclare(
  17.         r.Exchange,
  18.         "fanout",
  19.         true,
  20.         false,
  21.         //true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
  22.         false,
  23.         false,
  24.         nil,
  25.         )
  26.     r.failOnErr(err, "Failed to declare exchange.")
  28.     // 2.发送消息
  29.     err = r.channel.Publish(
  30.         r.Exchange,
  31.         "",
  32.         false,
  33.         false,
  34.         amqp.Publishing{
  35.             ContentType: "text/plain",
  36.             Body: []byte(message),
  37.         })
  38.     r.failOnErr(err, "Failed to publish message.")
  39. }
  41. func (r *RabbitMQ) ReceiveSub() {
  42.     // 1.尝试创建交换机
  43.     err := r.channel.ExchangeDeclare(
  44.         r.Exchange,
  45.         // 交换机类型
  46.         "fanout",
  47.         true,
  48.         false,
  49.         false,
  50.         false,
  51.         nil,
  52.         )
  53.     r.failOnErr(err, "Failed to declare exchange")
  55.     // 2.试探创建队列
  56.     q, err := r.channel.QueueDeclare(
  57.         "",
  58.         false,
  59.         false,
  60.         true,
  61.         false,
  62.         nil,
  63.         )
  64.     r.failOnErr(err, "Failed to decalre queue")
  66.     // 绑定队列到 exchange中
  67.     err = r.channel.QueueBind(
  68.         q.Name,
  69.         // 在pub/sub模式下, 这里的key必须为空
  70.         "",
  71.         r.Exchange,
  72.         false,
  73.         nil,
  74.         )
  75.     r.failOnErr(err, "Failed to decalre queue")
  77.     // 消费信息
  78.     msgs, err := r.channel.Consume(
  79.         q.Name,
  80.         "",
  81.         true,
  82.         false,
  83.         false,
  84.         false,
  85.         nil,
  86.         )
  88.     forever := make(chan bool)
  89.     go func() {
  90.         for msg := range msgs {
  91.             log.Printf("Received a message: %s", msg.Body)
  92.         }
  93.     }()
  95.     <-forever
  96. }

让我们来试下
pub.go

  1. package main
  3. import (
  4.     "fmt"
  5.     "github.com/WenkeZhou/flash-sale/pkg/rabbitmq"
  6.     "strconv"
  7.     "time"
  8. )
  10. func main() {
  11.     rmq := rabbitmq.NewRabbitMQPubSub("exchange_test")
  12.     for i:=1; i<=100; i++ {
  13.         rmq.PublishPub("this is pub message: " + strconv.Itoa(i))
  14.         fmt.Printf("this is pub message: " + strconv.Itoa(i))
  15.         time.Sleep(1 * time.Second)
  16.     }
  17. }

sub.go

  1. package main
  3. import (
  4.     "github.com/WenkeZhou/flash-sale/pkg/rabbitmq"
  5. )
  7. func main() {
  8.     rmq := rabbitmq.NewRabbitMQPubSub("exchange_test")
  9.     rmq.ReceiveSub()
  10. }