4、Go连接RabbitMQ：使用amqp库、建立连接、创建Channel、声明Queue、发送消息

好，咱们正式开始动手了。

前面聊了那么多理论，什么消息模型、交换机、路由键……但说一千道一万，不如亲手写一行代码来得实在。这一节，我就带你从零开始，用 Go 连上 RabbitMQ，然后发一条消息出去。

你想想看，这其实就是整个异步处理体系的「第一公里」。连不上，后面全是空谈。

4.1 选对库：为什么是 amqp？

Go 生态里操作 RabbitMQ，最主流的库就是 github.com/streadway/amqp。这个库我用了好几年，社区活跃，API 设计也比较符合 Go 的习惯。

当然，现在也有个官方维护的 rabbitmq/amqp091-go，本质上是同一个东西的 fork。我个人习惯还是用 streadway 的版本，因为文档和示例更丰富，踩过的坑网上基本都能搜到。

安装很简单：

go get github.com/streadway/amqp

嗯，就这一行。然后在你代码里 import 就行。

4.2 建立连接：别小看这一步

连接 RabbitMQ 服务器，用的是 amqp.Dial()。你需要提供一个 AMQP 协议的 URL。

格式长这样：

amqp://用户名:密码@主机地址:端口号/vhost

举个例子：

conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
    log.Fatal("连接失败：", err)
}
defer conn.Close()

这里有个细节我想强调一下——连接是重量级的。它底层维护了一个 TCP 连接，还有心跳检测。所以千万不要在每次发消息时都新建一个连接，那样性能会非常差。

我曾经见过一个项目，每次发消息都 Dial 一次，结果生产环境一压测，连接数直接打爆了 RabbitMQ 的服务端。后来改成连接池复用，问题才解决。

正确的做法是：应用启动时建立连接，整个生命周期复用这一个连接。

4.3 创建 Channel：真正的干活通道

连接建好了，但你不能直接用连接发消息。RabbitMQ 的设计里，Channel（通道）才是执行操作的实体。

你可以把连接想象成一条光纤，Channel 就是光纤里的一根根信道。大部分协议操作——声明队列、发布消息、消费消息——都是在 Channel 上完成的。

ch, err := conn.Channel()
if err != nil {
    log.Fatal("创建 Channel 失败：", err)
}
defer ch.Close()

这里要注意：Channel 是轻量级的，你可以创建很多个。但也不是无限制，默认一个连接最多 2047 个 Channel。我一般一个业务场景用一个 Channel，够用了。

4.4 声明 Queue：没有队列，消息往哪放？

有了 Channel，下一步就是声明一个队列。说白了，就是告诉 RabbitMQ：「嘿，帮我创建一个叫 xxx 的队列，我要往里面放消息。」

q, err := ch.QueueDeclare(
    "hello",   // 队列名称
    false,     // 是否持久化
    false,     // 是否自动删除
    false,     // 是否排他
    false,     // 是否阻塞等待
    nil,       // 额外参数
)
if err != nil {
    log.Fatal("声明队列失败：", err)
}

这几个参数我解释一下：

参数 说明 我常用的值
name 队列名称 按业务命名，如 order_paid_queue
durable 是否持久化。true 表示 RabbitMQ 重启后队列还在 重要消息用 true
autoDelete 最后一个消费者断开后是否自动删除 临时队列用 true
exclusive 是否排他。仅当前连接可见 基本不用
noWait 是否非阻塞 一般 false，确保服务端确认
我个人习惯：生产环境的队列，durable 一定设为 true。否则 RabbitMQ 一重启，队列没了，消息全丢。这个坑我踩过，血的教训。

4.5 发送消息：终于到了这一步

队列有了，发消息就简单了。用 Channel 的 Publish 方法：

err = ch.Publish(
    "",     // exchange，空字符串表示使用默认交换机
    q.Name, // routing key，这里直接填队列名
    false,  // mandatory
    false,  // immediate
    amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte("Hello, RabbitMQ!"),
    },
)

这里有几个点我想说一下：

  • exchange 为空：表示使用默认的直连交换机，消息会直接路由到 routing key 指定的队列。初学者这样用没问题，但生产环境我建议显式声明交换机。
  • ContentType：我一般用 text/plain 或 application/json。如果是 JSON 消息，记得设成后者，方便消费者解析。
  • Body：必须是 []byte。所以结构体要发的话，先序列化。

4.6 完整示例：跑起来看看

把上面几段拼起来，就是一个完整的发送端：

package main

import (
    "log"
    "github.com/streadway/amqp"
)

func main() {
    // 1. 建立连接
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatal("连接失败：", err)
    }
    defer conn.Close()

    // 2. 创建 Channel
    ch, err := conn.Channel()
    if err != nil {
        log.Fatal("创建 Channel 失败：", err)
    }
    defer ch.Close()

    // 3. 声明队列
    q, err := ch.QueueDeclare(
        "hello",
        false,
        false,
        false,
        false,
        nil,
    )
    if err != nil {
        log.Fatal("声明队列失败：", err)
    }

    // 4. 发送消息
    body := "Hello, RabbitMQ!"
    err = ch.Publish(
        "",
        q.Name,
        false,
        false,
        amqp.Publishing{
            ContentType: "text/plain",
            Body:        []byte(body),
        },
    )
    if err != nil {
        log.Fatal("发送消息失败：", err)
    }

    log.Printf(" [x] 发送成功：%s", body)
}

运行这段代码前，确保你的 RabbitMQ 已经启动。然后执行：

go run main.go

如果看到 [x] 发送成功：Hello, RabbitMQ!，恭喜你，第一条消息发出去了！

验证方法：打开 RabbitMQ 管理界面（默认 http://localhost:15672），进入 Queues 标签页，找到 hello 队列，点进去就能看到消息数变成了 1。

4.7 避坑指南

最后，分享几个我实际项目中遇到的坑：

  • 连接超时：默认 Dial 没有超时设置。我建议用 amqp.DialConfig() 自己配一个超时，比如 5 秒。否则网络不通时，程序会一直卡着。
  • Channel 关闭后复用：Channel 一旦发生错误（比如网络闪断），就会自动关闭。这时候如果你还拿它发消息，会 panic。我一般会在每次 Publish 前检查一下 Channel 的状态。
  • 消息体不要太大：RabbitMQ 默认消息大小限制是 4MB。超过这个值，消息会被拒绝。如果你要发大文件，建议先存到对象存储，消息里只放个路径。

好了，这一节的内容就到这。你学会了怎么连 RabbitMQ、怎么建队列、怎么发消息。下一节，咱们聊聊怎么把消息消费掉——那才是异步处理的精髓所在。

← 上一章 📖 返回目录 下一章 →