Go 使用 rabbitmq 详解

80 min read

RabbitMQ 是一种开源的消息队列软件，使用了标准协议来实现 AMQP（Advanced Message Queuing Protocol），是分布式应用解耦、异步处理等场景下常用的消息中间件之一。Go 在使用 RabbitMQ 时可以通过 AMQP 库和 RabbitMQ 的 API 直接进行交互。

下面就详细说明如何使用 Go 和 RabbitMQ：

  1. 安装 AMQP 库

使用 go get 命令安装 AMQP 库：

go get github.com/streadway/amqp
  1. 连接 RabbitMQ

创建连接时需要指定 RabbitMQ 的连接参数：

  • 连接参数

    • username：登录 RabbitMQ 的用户名
    • password：登录 RabbitMQ 的密码
    • host：RabbitMQ 服务器的 IP 地址或主机名
    • port：RabbitMQ 的 AMQP 端口号，默认为 5672
    • vhost：RabbitMQ 的虚拟主机，默认为 /

打开与 RabbitMQ 的连接和通道：

// 连接 RabbitMQ
conn, err := amqp.Dial("amqp://username:password@host:port/vhost")
if err != nil {
    // 连接失败
}

// 打开通道
ch, err := conn.Channel()
if err != nil {
    // 打开通道失败
}
defer ch.Close()
  1. 定义交换机

在 RabbitMQ 中，消息的生产者只发送消息给交换机，而消息的消费者从队列中获取消息。交换机根据自己的类型和路由键将消息路由到一个或多个队列中。

RabbitMQ 中支持四种类型的交换机：

  • direct：根据完全匹配的路由键将消息路由到队列中。
  • fanout：将消息广播到所有绑定到交换机的队列中。
  • topic：根据通配符匹配的路由键将消息路由到符合条件的队列中。
  • headers：根据消息的头部信息将消息路由到符合条件的队列中。

以下是创建一个 fanout 类型的交换机的示例代码：

// 声明交换机
err = ch.ExchangeDeclare(
    "my-exchange", // 交换机名称
    "fanout", // 交换机类型
    true, // 是否持久化交换机
    false, // 是否自动删除交换机
    false, // 是否内部交换机
    false, // 是否禁止等待确认
    nil, // 其他属性
)
if err != nil {
    // 声明交换机失败
}
  1. 定义队列

在 RabbitMQ 中，队列是存储消息的地方。当消息生产者向交换机发送消息时，如果没有任何队列与之绑定，那么消息将会被丢弃。因此，在消费者处理消息之前需要先声明并绑定一个队列。

以下是创建一个队列的示例代码：

// 声明队列
q, err := ch.QueueDeclare(
    "my-queue", // 队列名称
    true, // 是否持久化队列
    false, // 是否自动删除队列
    false, // 是否排他队列
    false, // 是否禁止等待确认
    nil, // 其他属性
)
if err != nil {
    // 声明队列失败
}

// 将队列绑定到交换机
err = ch.QueueBind(
    q.Name, // 队列名称
    "", // 路由键
    "my-exchange", // 交换机名称
    false, // 是否禁止等待确认
    nil, // 其他属性
)
if err != nil {
    // 绑定队列失败
}
  1. 发送消息

消息生产者向交换机发送消息时需要指定交换机的名称和路由键。交换机收到消息后会根据自己的类型和路由键将消息路由到一个或多个队列中。

以下是发送消息的示例代码：

// 设置消息内容
body := "hello world"

// 发送消息
err = ch.Publish(
    "my-exchange", // 交换机名称
    "", // 路由键
    false, // 是否禁止等待确认
    false, // 是否持久化消息
    amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(body),
    },
)
if err != nil {
    // 发送消息失败
}
  1. 消费消息

消息消费者从队列中接收消息时可以使用 Basic.Consume 方法创建一个消费者，并指定消费的队列名称。当有消息到达队列时，消费者会收到通知，然后可以在回调函数中处理消息。

以下是消费消息的示例代码：

// 创建消费者
msgs, err := ch.Consume(
    "my-queue", // 队列名称
    "",         // 消费者名称，为空时表示自动生成
    true,       // 是否自动确认消息
    false,      // 是否为独占消费者
    false,      // 是否禁止等待确认
    false,      // 是否为非阻塞操作
    nil,        // 其他属性
)
if err != nil {
    // 创建消费者失败
}

// 处理消息
for msg := range msgs {
    // 处理消息
}
  1. 关闭连接和通道

在程序退出时需要关闭与 RabbitMQ 的连接和通道：

// 关闭连接
err = conn.Close()
if err != nil {
    // 关闭连接失败
}

// 关闭通道
err = ch.Close()
if err != nil {
    // 关闭通道失败
}

至此，我们就完成了使用 Go 和 RabbitMQ 的基本操作，包括连接 RabbitMQ、定义交换机和队列、发送消息以及消费消息等。在实际应用中，我们还需要考虑一些其他因素，例如如何处理异常情况，如何确保消息的可靠性等。