RabbitMQ Fanout Exchange 多消费者正确实现指南

在 Go 中使用 RabbitMQ Fanout Exchange 时，如果多个消费者只能交替接收消息，而不是同时收到，多半是交换器类型定义或队列绑定逻辑出了问题。解决方案其实很明确：显式声明为 fanout 类型，并确保每个消费者绑定到同一交换器下的独立队列。

在 Go 中使用 RabbitMQ Fanout Exchange，如果发现多个消费者只能轮着干活——你一条我一条，而不是各拿一份——那基本可以断定是交换器定义或队列绑定环节出了问题。要解决，核心就两步：把交换器类型明确设为 fanout，再让每个消费者绑定自己的独立队列。

Fanout Exchange 设计的初衷就是广播。所有绑上去的队列，都会完整复制并收到每一条发布的消息。但要实现这种效果，需要确保两个基础条件成立：

1. 交换器必须被正确声明为 fanout 类型（而不是稀里糊涂用了默认的 direct 或根本没声明）；
2. 每个消费者要用各自独立的队列名（别共用 “example.queue”），并分别绑定到同一个 fanout 交换器。

看看你当前代码里常见的问题：

• ❌ 没调用 channel.ExchangeDeclare(...) 去声明一个 fanout 类型交换器；
• ❌ 两个消费者都用了相同的队列名 “example.queue”，RabbitMQ 因此把它们视作同一个队列的多个消费者实例，于是采用轮询分发（Round-Robin），而不是广播。

下面是正确的处理方式：

✅ 步骤一：统一声明 Fanout Exchange

无论是在发布端还是消费者初始化时，建议在连接建立后、正式开始消费之前，提前声明一次交换器：

err := channel.ExchangeDeclare(
    "logs",   // 交换器名称，推荐语义化命名，比如 "logs"、"broadcast"
    "fanout", // 类型必须是 "fanout"
    true,     // durable: 持久化，重启后不会丢失
    false,    // auto-deleted: 不自动删除
    false,    // internal: 非内部交换器
    false,    // no-wait
    nil,      // arguments
)
if err != nil {
    log.Fatalf("Failed to declare exchange: %v", err)
}

⚠️ 注意：ExchangeDeclare 是幂等操作，调用一次就行。多个消费者或生产者可以复用同一个交换器，不必重复声明。

✅ 步骤二：为每个消费者创建专属队列并绑定

修改你的 HandleMessageFanout1 和 HandleMessageFanout2，让它们分别使用不同的队列名，并显式绑定到 logs 交换器：

// HandleMessageFanout1 —— 使用队列 "queue-fanout-1"
func HandleMessageFanout1() {
    conn := system.EltropyAppContext.RabbitMQConn
    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Failed to open channel: %v", err)
    }
    defer ch.Close()

    // 声明专属队列（也可以不指定名称，让 RabbitMQ 自动生成；这里显式命名方便调试）
    q, err := ch.QueueDeclare(
        "queue-fanout-1", // 唯一队列名
        true,             // durable
        false,            // delete when unused
        false,            // exclusive
        false,            // no-wait
        nil,              // args
    )
    if err != nil {
        log.Fatalf("Failed to declare queue: %v", err)
    }

    // 绑定队列到 fanout 交换器（routingKey 在 fanout 中会被忽略，传空字符串就行）
    err = ch.QueueBind(
        q.Name,    // queue name
        "",        // routing key (ignored for fanout)
        "logs",    // exchange name
        false,     // no-wait
        nil,
    )
    if err != nil {
        log.Fatalf("Failed to bind queue to exchange: %v", err)
    }

    // 开始消费
    msgs, err := ch.Consume(
        q.Name,    // queue
        "",        // consumer tag (empty = auto-generated)
        true,      // auto-ack
        false,     // exclusive
        false,     // no-local
        false,     // no-wait
        nil,
    )
    if err != nil {
        log.Fatalf("Failed to register consumer: %v", err)
    }

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

同理，HandleMessageFanout2 应当使用 "queue-fanout-2" 作为队列名，并完成相同的声明与绑定流程。

✅ 补充：生产者示例（Go）—— 向 logs 交换器发布消息

// 示例：Go 生产者（在同一 channel 上操作）
err := ch.Publish(
    "logs",    // exchange
    "",        // routing key (ignored)
    false,     // mandatory
    false,     // immediate
    amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte("Hello from Fanout!"),
    },
)

? 总结与注意事项

• ? Fanout Exchange 不依赖 routing key，所有绑定的队列无条件接收全部消息；
• ? 每个消费者必须对应独立的队列（不能共用 queue name），否则会退化为竞争消费模式；
• ⚙️ 建议把 ExchangeDeclare 和 QueueDeclare 放在应用启动时集中初始化，避免重复声明；
• ? 如果测试过程中有旧的队列残留，可以通过 RabbitMQ Management UI 手动清理，或者在开发阶段使用 autoDelete: true；
• ? 官方权威参考：RabbitMQ Tutorial 3 — Publish/Subscribe (Go)。

按照上面这个结构来调整，两个 Go 消费者就能够同时、独立、完整地收到每一条 fanout 消息——广播语义才算真正落地。