RabbitMQ入门指南

想象一下你正在构建一个电子商务网站。当一个用户下单时，系统需要完成好几个任务：

  • 扣减库存 (Inventory Service)
  • 处理支付 (Payment Service)
  • 创建订单 (Order Service)
  • 通知仓库发货 (Shipping Service)
  • 给用户发送邮件/短信确认 (Notification Service)

传统方式的问题 (The Problem)

在没有消息队列的传统模式下，"订单服务"可能会直接去调用其他所有服务。这叫做同步调用或紧密耦合（Tightly Coupled）。

这种方式会带来几个严重的问题：

  1. 性能瓶颈: 用户点击“下单”后，必须等待所有这些服务全部成功返回，才能收到“下单成功”的提示。如果其中任何一个服务（比如邮件服务）响应很慢，整个流程就会被拖慢，用户体验会很差。

  2. 脆弱性 (Cascading Failures): 如果其中一个不那么重要的服务，比如“通知服务”，因为故障宕机了，那么整个下单流程都会失败。一个次要环节的失败导致了核心业务的失败，这非常不合理。

  3. 扩展性差: 假如“双十一”大促，下单量暴增。订单服务会疯狂调用其他服务，可能瞬间就把库存、通知等服务给压垮了。我们很难针对性地去扩展某一个环节。

RabbitMQ 的解决方案 (The Solution)

现在，我们把 RabbitMQ 这个“智能邮局”放到系统的中间。

整个流程会变成这样：

  1. 用户下单，请求到达“订单服务”。

  2. “订单服务”创建一个“订单已创建”的消息，然后把这个消息扔给 RabbitMQ，就立刻告诉用户“下单成功！”

  3. 订单服务的任务到此结束，它完全不关心谁需要这个消息，也不用等待其他服务的回应。

  4. 然后，其他需要这个信息的服务，比如库存服务、发货服务、通知服务，它们会自己去 RabbitMQ “订阅”它们感兴趣的消息。当 RabbitMQ 收到“订单已创建”的消息后，它会把这条消息的副本分发给所有订阅了它的服务。

这样做带来了什么核心好处？

1. 解耦 (Decoupling)
这是最核心的价值。“订单服务”（生产者）和“库存服务”（消费者）之间不再有直接联系。它们都只和 RabbitMQ 打交道。
你随时可以增加新的消费者。比如，增加一个“数据分析服务”来分析订单，只需要让它也去订阅消息就行，完全不需要修改现有的任何代码。
任何一个消费者服务宕机，完全不影响生产者和其他消费者。

2. 异步通信 (Asynchronous Communication)
生产者把消息扔给 RabbitMQ 后就不用管了，可以立即处理下一个请求。这大大提高了应用的响应速度和吞吐量。就像发邮件，你点击发送后就可以关掉邮箱干别的事了，不用在线等着对方已读。

3. 削峰填谷 (Peak Shaving / Load Balancing)
当大促来临时，瞬间涌入成千上万的订单请求。生产者会把这些消息快速地堆积到 RabbitMQ 的队列中。而消费者服务则可以根据自己的处理能力，平稳地、一个个地从队列里取出消息进行处理。
这就像一个巨大的蓄水池，无论上游的洪水多大，下游都能保持稳定的水流。这保证了即使在流量高峰期，系统也不会被压垮。

4. 可靠性 (Reliability)
RabbitMQ 提供了多种机制来确保消息不会丢失。如果一个消费者在处理消息的过程中崩溃了，RabbitMQ 可以确保这个消息会被重新交给另一个健康的消费者来处理。消息被安全地存储在队列中，直到被正确处理完毕。

总结

所以，你可以把 RabbitMQ 理解为一个系统内部的中央调度中心和缓冲层。它不生产也不消费数据，它的唯一使命就是可靠、高效地传递消息，从而让整个系统架构变得更加灵活、健壮和可扩展。

和redis的pub-sub有什么区别？

它们的核心区别在于设计目标和可靠性承诺。

  • RabbitMQ 就像一个专业的联邦快递 (FedEx) 系统。它有严格的流程、签收、追踪、投递失败后的重试机制，确保你的包裹（消息）一定能准确、可靠地送达。

  • Redis Pub/Sub 更像一个大喇叭广播 (Loudspeaker)。它会用最快的速度把你的话喊出去，所有正在听的人都能立刻听到。但如果某个人当时没在场（客户端离线），他就永远错过了这条信息，而且广播者也不知道有谁听到了，有谁没听到。

在这里插入图片描述

值得注意的是，从 Redis 5.0 开始，引入了一个新的数据结构叫做 Streams。Redis Streams 在设计上借鉴了
Kafka，提供了消息持久化、消费组（类似 RabbitMQ 的竞争消费）和消息确认等功能，极大地增强了 Redis 作为消息队列的能力。

RabbitMQ 核心组件

想象一个大型的自动化包裹处理中心，这就是 RabbitMQ 的世界。在这个世界里，有几个关键角色：

  1. 生产者 (Producer): 就是寄件人。他写好包裹（创建消息），然后把它送到处理中心。

  2. 消费者 (Consumer): 就是收件人。他在家里等着，收取处理中心派送来的包裹。

  3. 交换机 (Exchange) 📜: 这是整个系统的分拣中心。它接收所有生产者送来的包裹，但它自己不存储。它的唯一工作是查看包裹上的“投递规则”，然后决定把它转发给哪个（或哪些）邮箱。

  4. 队列 (Queue) 📬: 这就是终端的**“邮政信箱”或“快递柜”**。它负责存储消息，直到有消费者来把它取走。一个队列可以被多个消费者共同监听，但一个消息只会被一个消费者取走（这就是任务分发）。

  5. 绑定 (Binding) 🔗: 这是一条**“订阅规则”**。它告诉交换机（分拣中心）和队列（邮箱）之间的关系。比如，“所有寄往‘A区’的包裹，都请投递到 1 号邮箱”。没有绑定，交换机就不知道该把消息发给谁。

消息的旅程

生产者 (寄件人) ➡️ 把消息 (包裹) 📦 发送给 交换机 (分拣中心) 📜 ➡️ 交换机根据绑定规则 (订阅关系) 🔗 决定把消息的副本推送到哪些 队列 (邮箱) 📬 ➡️ 消费者 (收件人) 从队列中取出消息进行处理。

这个结构就是 RabbitMQ 解耦和灵活性的关键。生产者根本不关心消息最终会去到哪个队列，它只管发给交换机。而消费者也只关心自己订阅的队列，不关心消息是谁发来的。

交换机类型（Exchange Types）/ 分拣策略

把消息发送给交换机时，我们通常会附带一个标签，叫做“路由键”（Routing Key）。你可以把它想象成包裹上写的地址信息，比如“收件人：张三”或“目的地：A区-01栋”。交换机就是根据这个路由键和它的类型规则来决定如何分拣的。

最常用的类型：

  1. Fanout (扇出/广播): 最简单粗暴的策略。
  2. Direct (直接): 点对点精确匹配的策略。
  3. Topic (主题): 最灵活、支持模糊匹配的策略。

Fanout Exchange (扇出/广播交换机)

这是最简单直接的一种交换机。它的工作原则是：忽略所有路由键 (Routing Key)，把收到的任何消息广播给所有绑定到它身上的队列。

可以把它想象成一个连接着多个喇叭的广播系统 📢。无论广播站说什么内容（消息），所有连接的喇叭（队列）都会一字不差地播放出来。

假设有一个体育新闻网站，需要实时更新比赛得分。当比分变化时：

生产者（后端服务）创建一个“比分更新”的消息，发送给一个 Fanout 类型的交换机。

这个交换机绑定了三个队列：一个给网站页面（Web Page Queue），一个给手机 App 推送（App Push Queue），还有一个给内部数据统计（Analytics Queue）。

Fanout 交换机收到消息后，会立刻把这条消息的三个副本分别发送到这三个队列中，完全不需要看消息上有没有写“发给App”之类的路由键。

这样，所有关心比分更新的服务都能同时收到通知。

Direct Exchange (直接交换机)

如果说 Fanout 是“广播”，那么 Direct 就是**“精确投递”**。

它的工作原则是：只有当消息的路由键 (Routing Key) 与队列的绑定键 (Binding Key) 完全匹配时，消息才会被投递到那个队列。

可以把它想象成一个非常严谨的邮政分拣员 👨‍💼。他会看包裹上的地址（Routing Key），然后对照自己的分拣手册（Binding Rules），只有地址一字不差地匹配上了，他才会把包裹放到对应的邮筒（Queue）里。

假设我们有一个后台任务系统，需要处理不同类型的日志：info（信息）、warning（警告）、error（错误）。

  1. 我们创建一个 Direct 类型的交换机，叫做 logs_direct。
  2. 我们创建两个队列：
    • error_logs_queue：专门用来处理错误日志。我们用 error 作为绑定键 (Binding Key) 把它绑定到 logs_direct 交换机。
    • all_logs_queue：用来处理所有其他日志。我们用 info 和 warning 这两个绑定键都把它绑定到 logs_direct 交换机。（是的，一个队列可以有多个绑定键）。
  3. 场景一: 生产者发送一条日志消息，路由键 (Routing Key) 设为 error。
    • 交换机发现 error 和 error_logs_queue 的绑定键匹配，于是把消息发给它。
  4. 场景二: 生产者发送一条日志消息，路由键设为 info。
    • 交换机发现 info 和 all_logs_queue 的一个绑定键匹配，于是把消息发给 all_logs_queue。error_logs_queue 不会收到这条消息。
核心用途：

这种模式非常适合用来做任务分发。你可以根据任务的类型或优先级，把它们发送到不同的队列，然后由专门的消费者来处理。

topic Exchange (主题交换机)

如果说 Direct Exchange 是“精确匹配”，那么 Topic Exchange 就是模糊匹配或“模式匹配”。

它的工作原则是：将消息的路由键 (Routing Key) 与队列的绑定键 (Binding Key) 进行模式匹配，如果匹配成功，消息就会被投递到该队列。

这里的匹配规则有点像正则表达式，但更简单，主要依赖两个特殊的通配符：

  • *(星号): 匹配一个单词。

  • #(井号): 匹配零个或多个单词。

假设我们在为一个新闻机构分发新闻稿。新闻稿的路由键格式为 <来源>.<类型>.<地区>，例如：

  • reuters.sports.usa (路透社的体育新闻，关于美国)
  • cnn.weather.europe (CNN 的天气新闻，关于欧洲)
  • reuters.finance.japan (路透社的财经新闻，关于日本)
    现在，不同的部门（队列）有不同的新闻订阅需求：
  1. 美国体育新闻部 (USA Sports Queue):
    绑定键: *.sports.usa
    它能收到 reuters.sports.usa, cnn.sports.usa 等消息。
  2. 所有路透社新闻部 (All Reuters Queue):
    绑定键: reuters.#
    它能收到 reuters.sports.usa, reuters.finance.japan 等所有来自路透社的消息。
  3. 所有关于日本的新闻部 (All Japan Queue):
    绑定键: ..japan
    它能收到 reuters.finance.japan, cnn.weather.japan 等消息。
  4. 天气新闻部 (Weather Queue):
    绑定键: .weather.
    它能收到 cnn.weather.europe, bbc.weather.asia 等所有天气新闻。
核心用途：

这种模式非常适合用于多标准、多维度的消息分类和订阅。发布者只需要按照规则定义好消息的“主题”，而消费者可以根据自己的兴趣，灵活地订阅一个或多个主题的消息。

Go代码实战

// send.go
package rabbitmq

import (
	"context"
	"github.com/rabbitmq/amqp091-go"
	"log"
	"time"
)

func failOnEror(err error, msg string) {
	if err != nil {
		log.Panicf("%s %s", msg, err)
	}
}
func NewProducer() {
	// 连接到RabbitMQ服务器
	conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
	failOnEror(err, "failed to connect to RabbitMQ")
	defer conn.Close()

	// 创建一个Channel
	ch, err := conn.Channel()
	failOnEror(err, "faided to open a channel")
	defer ch.Close()

	// 声明一个队列
	q, err := ch.QueueDeclare(
		"hello",
		false,
		false,
		false,
		false,
		nil,
	)
	failOnEror(err, "failed to declare a queue")

	// 创建一个有超时上下文的context
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// 发送消息
	body := "Hello World!"
	err = ch.PublishWithContext(ctx,
		"",     // 默认是direct exchange
		q.Name, // routing key (这里是队列名)
		false,
		false,
		amqp091.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		},
	)
	failOnEror(err, "failed to publish a message")
	log.Printf(" [x] Sent %s\n", body)
}


// receive.go
package rabbitmq

import (
	amqp "github.com/rabbitmq/amqp091-go"
	"log"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Panicf("%s %s", msg, err)
	}

}

func NewConsumer() {
	// 连接RabbitMQ服务器
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "failed to connect to RabbitMQ")
	defer conn.Close()

	// 创建Channel
	ch, err := conn.Channel()
	failOnError(err, "failed to open a channel")
	defer ch.Close()

	// 声明队列 (同样需要声明，以防消费者先于生产者启动)
	q, err := ch.QueueDeclare(
		"hello", // name
		false,   // durable
		false,   // delete when unused
		false,   // exclusive
		false,   // no-wait
		nil,     // arguments
	)
	failOnError(err, "Failed to declare a queue")

	// 4. 消费消息
	// ch.Consume 会返回一个 Go 的 channel，用于接收消息
	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		true,   // auto-ack (自动确认消息)
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	failOnError(err, "Failed to register a consumer")

	// 只消费一条然后退出
	d := <-msgs
	log.Printf("Received a message: %s", d.Body)
	d = <-msgs
	log.Printf("Received a message: %s", d.Body)
	d = <-msgs
	log.Printf("Received a message: %s", d.Body)
	d = <-msgs
	log.Printf("Received a message: %s", d.Body)
	d = <-msgs
	log.Printf("Received a message: %s", d.Body)

	return
}


// xx_test.go
package rabbitmq

import (
	"testing"
	"time"
)

func TestNewProducer(t *testing.T) {

	go NewProducer()
	go NewProducer()
	go NewProducer()
	go NewProducer()
	go NewProducer()
	go NewConsumer()
	time.Sleep(time.Second * 5)

}

工作队列模式：如何让多个消费者来共同处理一个队列里的任务

这个模式的核心目标是：

将一个耗时的任务分发给多个消费者（工作者/Worker）并行处理，从而提高整体效率。

1. 核心思想

想象一下，你有一个需要处理大量图片尺寸调整的任务。如果只有一个程序来做，可能需要好几个小时。但如果你启动 10 个程序，每个程序处理 1/10 的图片，总时间就会大大缩短。

工作队列就是用来实现这一点的。生产者（Task Publisher）不断地把任务（比如“请调整这张图片的尺寸”）作为消息发送到一个队列中。多个消费者（Workers）同时监听这一个队列。RabbitMQ 会把队列中的任务依次、轮流地分发给这些消费者。

默认情况下，RabbitMQ 使用轮询分发 (Round-robin) 策略。比如有 2 个 Worker（W1, W2）和 4 个任务（T1, T2, T3, T4），那么：

  • T1 会被发送给 W1
  • T2 会被发送给 W2
  • T3 会被发送给 W1
  • T4 会被发送给 W2

2. 消息确认 (Message Acknowledgment) - 关键！❗

在我们之前的 “Hello World” 例子中，我们设置了 auto-ack: true。这意味着当 RabbitMQ 把消息发给消费者后，就立刻在队列中把这条消息删除了，不管消费者是否真的处理成功。

这是很危险的！ 如果一个 Worker 在处理一个重要任务时突然崩溃了，那么这个任务就永远丢失了。

为了解决这个问题，我们需要：

  1. 关闭自动确认：设置 auto-ack: false。
  2. 手动确认：在 Worker 完成任务后，手动调用 d.Ack(false) 方法，明确告诉 RabbitMQ：“我已经处理完这条消息了，你可以安全地删除它了”。
    这样，如果一个 Worker 在处理过程中挂掉了，没有发送 ACK，RabbitMQ 就会知道这个任务没有被成功处理，然后会把它重新投递给另一个空闲的 Worker。这就保证了任务的可靠执行。

注意：RabbitMQ 的真实分发逻辑更准确的描述是：

“将消息发送给分发当时，下一个空闲的消费者。”
注意是：一次性地、采用轮询方式，把一批消息（一个预取缓冲区大小的量）快速地推送给每个消费者，让消息先在消费者的客户端缓存区里“排队”。

默认情况下，RabbitMQ的分发不是基于消费者是否“应用级别”的空闲，而是一种“尽早、尽快”的预取推送。这会导致负载不均，快的消费者处理完后可能会没事干，而慢的消费者自己缓存区里还屯着一堆任务。

3. Go 语言实战
生产者会发送几条模拟任务，消费者会模拟处理这些任务。

// send.go
func NewProducerGetFromParameter(newMsg string) {
	// 连接到RabbitMQ服务器
	conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
	failOnEror(err, "failed to connect to RabbitMQ")
	defer conn.Close()

	// 创建一个Channel
	ch, err := conn.Channel()
	failOnEror(err, "faided to open a channel")
	defer ch.Close()

	// 声明一个队列
	q, err := ch.QueueDeclare(
		"hello",
		false,
		false,
		false,
		false,
		nil,
	)
	failOnEror(err, "failed to declare a queue")

	// 创建一个有超时上下文的context
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// 发送消息
	body := newMsg
	err = ch.PublishWithContext(ctx,
		"",     // 默认是direct exchange
		q.Name, // routing key (这里是队列名)
		false,
		false,
		amqp091.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		},
	)
	failOnEror(err, "failed to publish a message")
	log.Printf(" [x] Sent %s\n", body)
}

func NewConsumerConsumTime(ConsumerID string) {
	// 连接RabbitMQ服务器
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "failed to connect to RabbitMQ")
	defer conn.Close()

	// 创建Channel
	ch, err := conn.Channel()
	failOnError(err, "failed to open a channel")
	defer ch.Close()

	// 声明队列 (同样需要声明，以防消费者先于生产者启动)
	q, err := ch.QueueDeclare(
		"hello", // name
		false,   // durable
		false,   // delete when unused
		false,   // exclusive
		false,   // no-wait
		nil,     // arguments
	)
	failOnError(err, "Failed to declare a queue")

	// 4. 消费消息
	// ch.Consume 会返回一个 Go 的 channel，用于接收消息
	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		false,
		false, // exclusive
		false, // no-local
		false, // no-wait
		nil,   // args
	)
	failOnError(err, "Failed to register a consumer")

	var forever chan struct{}
	// 只消费一条然后退出
	go func() {
		for d := range msgs {
			log.Printf("%s Received a message: %s", ConsumerID, d.Body)
			// 模拟耗时任务 根据msg中.的数量确定消耗时间
			dotCount := bytes.Count(d.Body, []byte("."))
			consumTime := time.Duration(dotCount) * time.Second
			time.Sleep(consumTime)
			log.Printf("%s Done, consumTime: %s", ConsumerID, consumTime)
			// !!! 手动发送确认消息
			d.Ack(false)
		}
	}()
	<-forever
}


func TestWorkQueues(t *testing.T) {

	go NewProducerGetFromParameter("..")
	go NewProducerGetFromParameter("...")
	go NewProducerGetFromParameter("....")
	go NewProducerGetFromParameter(".....")
	go NewConsumerConsumTime("消费者_1")
	go NewConsumerConsumTime("消费者_2")
	time.Sleep(time.Second * 30)
}

在这里插入图片描述

默认情况下，RabbitMQ的分发不是基于消费者是否“应用级别”的空闲，而是一种“尽早、尽快”的预取推送。这会导致负载不均，快的消费者处理完后可能会没事干，而慢的消费者自己缓存区里还屯着一堆任务。
这正是 ch.Qos(prefetchCount: 1) 的意义所在！
它等于你明确告诉Broker：“请不要给我预取消息。我手里一次只能拿一个任务。等我通过 d.Ack() 把这个任务明确告诉你我已经处理完了，你才能把下一个任务发给我。”
设置了 Qos(1) 之后，RabbitMQ的整个分发逻辑才会变成我们最初设想的那种公平分发”模式：真正地把任务交给下一个完全空闲的消费者。

因此，在消费者中设置Qos公平调度，就可以实现：（谁有空谁干活，负载均衡）
在这里插入图片描述

持久化 (Persistence)

为了防止数据丢失，我们需要做两件事情，缺一不可：

- 让队列本身持久化 (Durable Queues)
- 让消息本身持久化 (Persistent Messages)

1. 如何实现队列持久化？

声明队列的时候，把 durable 参数设置为 true 即可
生产者和消费者都必须将队列声明为 durable: true。

// 生产者和消费者都必须将队列声明为 durable: true。
q, err := ch.QueueDeclare(
    "task_queue", // name
    true,         // durable: <--- 从 false 改为 true
    false,        // delete when unused
    false,        // exclusive
    false,        // no-wait
    nil,          // arguments
)

不能将一个已经存在的、非持久化的队列（比如我们之前创建的 task_queue）重新声明为持久化的，这样做会报错。你需要先通过管理界面 (http://localhost:15672) 删除旧的 task_queue，或者声明一个新名字的队列。

2. 如何实现消息持久化？

在生产者发布消息时，需要将 amqp.Publishing 结构体中的 DeliveryMode 属性设置为 amqp.Persistent (其值为2)。

// 只在 生产者 中修改
err = ch.PublishWithContext(ctx,
    "",     // exchange
    q.Name, // routing key
    false,  // mandatory
    false,
    amqp.Publishing{
        DeliveryMode: amqp.Persistent, // <--- 关键在这里！
        ContentType:  "text/plain",
        Body:         []byte(body),
    },
)

现在，当你向一个持久化的队列发送一条持久化的消息时，RabbitMQ 会将这条消息写入磁盘。

严格来说，即使设置了消息持久化，RabbitMQ 也不是在收到消息的瞬间就 100% 写入了磁盘，中间有一个非常短暂的缓存过程。对于要求极度严格、完全不能容忍任何数据丢失的金融级应用，RabbitMQ 还提供了更高级的发布者确认 (Publisher Confirms) 机制。但对于绝大多数场景，我们刚才的设置已经足够健壮了。

发布/订阅 (Publish/Subscribe)

这个模式的目标是：将一条消息广播给多个消费者。每个消费者都能收到同样消息的完整副本。这与“工作队列”模式（一条消息只被一个消费者处理）形成了鲜明对比。

整个流程会是这样：

  1. 生产者 (Publisher) 不再将消息直接发送到队列，而是发送到一个 Fanout 类型的交换机。
  2. 每个消费者 (Subscriber) 启动时，都会创建一个属于自己的、临时的、匿名的队列。
  3. 然后，每个消费者都会把自己的这个临时队列绑定 (Bind) 到生产者所使用的那个 Fanout 交换机上。
  4. 当生产者发布一条消息到交换机时，交换机就会把这条消息的副本分发给所有绑定在它身上的队列，从而每个消费者都能从自己的队列中收到这条消息。

在这个发布/订阅模式下，必须先运行消费者，再发送消息，消费者才能收到。

1. 交换机是“无记忆”的
Fanout Exchange 就像一个实时的广播站 📢。它的唯一职责就是把收到的消息立刻转发给当前所有已连接的“收音机”（也就是绑定在它身上的队列）。
如果消息到达时，没有任何队列绑定在它身上，那么这条消息就会被直接丢弃。交换机本身不存储任何消息。它就像对着空无一人的山谷呐喊，声音发出去就消失了。
2. 消费者的队列是“临时”的
这意味着每个消费者都有一个属于自己的、独一无二的“收件箱”。但这个“收件箱”是在消费者启动并连接到 RabbitMQ 时才被创建的。当消费者程序退出，这个队列也会随之消失。

交换机不是存储的介质，队列才是，只有当消费者被创建时，其临时队列才能被创建

生产者把消息发给 logs 交换机。如果此时没有任何消费者在运行，那么就没有队列绑定到 logs 交换机上，消息自然就丢失了。只有当消费者先启动，创建好自己的临时队列并将其绑定到 logs 交换机之后，它才具备了接收消息的能力。

代码

// emit_logs.go
package rabbitmq

import (
	"context"
	"log"
	"time"
)
import amqp "github.com/rabbitmq/amqp091-go"

func newExchange(msg string) {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "failed to open a channel")
	defer ch.Close()

	err = ch.ExchangeDeclare(
		"logs",
		"fanout",
		true,
		false,
		false,
		false,
		nil,
	)
	failOnError(err, "failed to declare an exchange")

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	body := msg
	err = ch.PublishWithContext(
		ctx,
		"logs",
		"",
		false,
		false,
		amqp.Publishing{
			ContentType: "test/plain",
			Body:        []byte(body),
		},
	)
	failOnError(err, "failed to publish a message")
	log.Printf(" [x] Sent %s", body)

}


// receive_logs.go
package rabbitmq

import (
	amqp "github.com/rabbitmq/amqp091-go"
	"log"
)

func consumExchange() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	// 1. 声明交换机 (同样需要声明，以防消费者先启动)
	err = ch.ExchangeDeclare(
		"logs",   // name
		"fanout", // type
		true,     // durable
		false,    // auto-deleted
		false,    // internal
		false,    // no-wait
		nil,      // arguments
	)
	failOnError(err, "Failed to declare an exchange")

	// 2. 声明一个临时的、匿名的队列
	// name="" -> RabbitMQ会为我们生成一个随机的名字
	// exclusive=true -> 当连接关闭时，这个队列会被自动删除
	q, err := ch.QueueDeclare(
		"",    // name
		false, // durable
		false, // delete when unused
		true,  // exclusive
		false, // no-wait
		nil,   // arguments
	)
	failOnError(err, "Failed to declare a queue")

	// 3. 将我们的临时队列绑定到 "logs" 交换机
	err = ch.QueueBind(
		q.Name, // queue name
		"",     // routing key (fanout模式下忽略)
		"logs", // exchange
		false,
		nil,
	)
	failOnError(err, "Failed to bind a queue")

	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		true,   // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)

	failOnError(err, "Failed to register a consumer")

	var forever chan struct{}

	go func() {
		for d := range msgs {
			log.Printf(" [x] %s", d.Body)
		}
	}()

	log.Printf(" [*] Waiting for logs. To exit press CTRL+C")
	<-forever

}


// xx_test.go
package rabbitmq

import (
	"testing"
	"time"
)

func TestLogs(t *testing.T) {

	go consumExchange()
	go consumExchange()
	time.Sleep(1 * time.Second)
	go newExchange("Hello World!")
	time.Sleep(30 * time.Second)
}

在这里插入图片描述

# golang # rabbitmq # ruby
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

SeedVR2终极指南：如何用单步扩散模型实现专业级视频修复

还在为老旧视频的模糊画质而烦恼吗？面对抖动严重的监控录像或褪色的家庭影像，你是否曾希望有一种简单高效的修复方案？SeedVR2正是为解决这些痛点而生的一站式视频修复工具。作为字节跳动推出的先进视频修复模型，它能够在单步推理中实现专业级的视频质量提升，彻底改变了传统视频修复的工作流程。## 核心价值：为什么选择SeedVR2进行视频修复？**传统视频修复的三大瓶颈**：1. **计算成本高

avatar AI Agent技术社区

深度解析Amlogic S9xxx电视盒子Armbian系统：从闲置设备到Linux服务器的终极实战指南

你是否拥有一台闲置的Amlogic S9xxx系列电视盒子？想让它摆脱单一的视频播放功能，变身为功能强大的Linux服务器、开发平台或家庭媒体中心？本指南将为你详细解析如何通过Armbian系统实现这一转变，让你的电视盒子重获新生，成为高性能的Linux设备。Armbian系统基于Debian/Ubuntu构建，专为ARM芯片优化，提供了完整的Linux生态支持。## 🎯 核心价值与独特优势

avatar AI Agent技术社区

CANN/runtime Event管理API

本章节描述 CANN Runtime 的 Event 管理接口，用于事件的创建、记录、同步、计时及 IPC 跨进程共享。- [`aclError aclrtCreateEvent(aclrtEvent *event)`](#aclrtCreateEvent)：创建Event，创建出来的Event可用于统计两个Event之间的耗时、多Stream之间的任务同步等场景。- [`aclError

avatar AI Agent技术社区