golang使用rabbitmq发布/订阅

Posted 2023-02-21 吴冬冬

上一篇文章中讲了工作队列，但是工作队列只能做到一个生产者的消息被消费者消费一次。现在要实现一个生产者的消息被两个不同的消费者同时消费，我们把这种模式叫做发布订阅。

交换器

RabbitMQ消息传递模型的核心是生产者永远不会把消息直接发送到某个队列上，甚至根本不知道消息是否发送到队列上。生产者只向交换器发送消息。

交换器来决定怎么处理消息，是把消息传递到特定的队列，还是把消息传到多个队列，或者直接丢弃消息，这一切取决于交换器的类型。

交换器的类型有四种:direct, topic, headers, fanout。在这里我们只讲fanout这一种，fanout类型非常简单，他会把消息发到所有他知道的队列。(之前我们用的是)

定义交换器：

	err = ch.ExchangeDeclare("logs", amqp.ExchangeFanout, true, false, false, false, nil)
	failOnError(err, "Failed to declare an exchange")

然后把消息发到交换器

	body := bodyFrom(os.Args)
	err = ch.Publish("logs", "", false, false, amqp.Publishing
		DeliveryMode: amqp.Persistent,
		ContentType:  "text/plain",
		Body:         []byte(body),
	)
	failOnError(err, "Failed to publish a message")

临时队列

在之前的文章中，队列都有一个名字，但是有些情况下队列名字并不重要，因为只要订阅消息即可fanout会把消息发给他所有知道的队列。传入一个空字符串，rabbit会帮我们生成一个随机的队列名。

q, err := ch.QueueDeclare("", false, false, true, false, nil)
	failOnError(err, "Failed to declare a queue")

绑定

现在交换器和队列都有了，使队列成为交换器"已知"的过程过程就叫做绑定：

	err = ch.QueueBind(q.Name, "", "logs", false, nil)
	failOnError(err, "Failed to bind a queue")

现在启动多个订阅者，但是只发布一条消息，多个订阅者都会收到相同的消息。