当前位置：首页 > news >正文

go语言zero框架对接阿里云消息队列MQ的rabbit的配置与调用

news 2025/7/15 1:43:06

在 Go 语言中对接阿里云消息队列（MQ）的 RabbitMQ 配置与调用，首先需要安装和配置相关的 Go 库，并了解如何通过 RabbitMQ 与阿里云消息队列进行交互。

### 步骤一：安装 RabbitMQ Go 客户端库

阿里云的消息队列（MQ）实际上是基于 **RabbitMQ** 实现的，因此可以使用标准的 RabbitMQ Go 客户端库来对接。

你可以使用 [github.com/rabbitmq/amqp091-go](github.com/rabbitmq/amqp091-go) 库，这是 Go 语言中常用的 RabbitMQ 客户端库。

```bash

go get github.com/rabbitmq/amqp091-go

```

### 步骤二：配置阿里云消息队列的连接信息

首先需要获取阿里云消息队列的连接信息，包括：
- **接入点（Endpoint）**：这是消息队列的服务器地址，通常可以在阿里云控制台的消息队列管理页面找到。
- **AccessKey 和 SecretKey**：这是用于认证和访问阿里云服务的凭证。

### 步骤三：创建连接和通道

在 Go 代码中，使用 AMQP 协议连接到 RabbitMQ 服务。阿里云的消息队列支持 AMQP 协议，所以可以直接通过它来连接。```go

package mainimport ("fmt""log""github.com/streadway/amqp"
)func main() {// 阿里云 MQ 的连接信息amqpURL := "amqp://<AccessKey>:<SecretKey>@<Endpoint>/vhost" // 根据实际情况填写conn, err := amqp.Dial(amqpURL)if err != nil {log.Fatalf("Failed to connect to RabbitMQ: %s", err)}defer conn.Close()// 创建一个通道（Channel）ch, err := conn.Channel()if err != nil {log.Fatalf("Failed to open a channel: %s", err)}defer ch.Close()// 创建一个队列q, err := ch.QueueDeclare("testQueue", // 队列名称true,        // 是否持久化false,       // 是否自动删除false,       // 是否具有独占权限false,       // 是否阻塞nil,         // 额外属性)if err != nil {log.Fatalf("Failed to declare a queue: %s", err)}fmt.Printf("Queue declared: %s\n", q.Name)// 发布一条消息到队列body := "Hello, this is a test message!"err = ch.Publish("",         // 默认交换机q.Name,     // 队列名称false,      // 是否等待服务器确认false,      // 是否设置强制推送amqp.Publishing{ContentType: "text/plain",Body:        []byte(body),})if err != nil {log.Fatalf("Failed to publish a message: %s", err)}fmt.Printf("Message sent: %s\n", body)
}

```

### 代码解释：
- **amqp.Dial**：连接到 RabbitMQ（阿里云 MQ）的消息队列服务，使用的是 AMQP 协议。在 URL 中需要包含 **AccessKey** 和 **SecretKey**，格式是 `amqp://<AccessKey>:<SecretKey>@<Endpoint>/vhost`。
- `<AccessKey>` 和 `<SecretKey>`：你在阿里云管理控制台中创建的 API 密钥。
- `<Endpoint>`：你可以从阿里云 MQ 控制台获取到的接入点地址。
- `/vhost`：虚拟主机，通常是 `/`，可以根据实际情况修改。

- **QueueDeclare**：声明队列。队列是消息的载体，在队列中存放的是等待被消费者取走的消息。这里使用了 **持久化队列**（`true`），表示即使服务器重启，队列和消息也不会丢失。

- **Publish**：将消息发送到指定的队列中。

### 步骤四：消费消息

除了发送消息，消费者（Consumer）也需要从队列中获取消息并进行处理。```go

package mainimport ("fmt""log""github.com/streadway/amqp"
)func main() {// 阿里云 MQ 的连接信息amqpURL := "amqp://<AccessKey>:<SecretKey>@<Endpoint>/vhost"conn, err := amqp.Dial(amqpURL)if err != nil {log.Fatalf("Failed to connect to RabbitMQ: %s", err)}defer conn.Close()// 创建一个通道（Channel）ch, err := conn.Channel()if err != nil {log.Fatalf("Failed to open a channel: %s", err)}defer ch.Close()// 声明队列（与生产者端一致）q, err := ch.QueueDeclare("testQueue", // 队列名称true,        // 持久化false,       // 非自动删除false,       // 非独占false,       // 非阻塞nil,         // 额外属性)if err != nil {log.Fatalf("Failed to declare a queue: %s", err)}// 获取消息msgs, err := ch.Consume(q.Name,    // 队列名称"",        // 消费者标签true,      // 自动应答false,     // 独占false,     // 不阻塞false,     // 不获取nil,       // 额外属性)if err != nil {log.Fatalf("Failed to register a consumer: %s", err)}fmt.Println("Waiting for messages. To exit press CTRL+C")// 消费消息for msg := range msgs {fmt.Printf("Received a message: %s\n", msg.Body)}
}

```

### 代码解释：
- **Consume**：消费者从队列中获取消息并处理。你可以设置 `auto-ack`（自动应答）为 `true`，表示 RabbitMQ 在消息被接收后自动确认消息。如果需要手动确认，可以设置为 `false`，并手动发送确认。
- 消费者会一直运行，并等待新的消息到来。

### 步骤五：测试和运行

1. 启动消费者程序（在一个终端中）。
2. 启动生产者程序（在另一个终端中）。生产者会向队列发送一条消息。
3. 消费者会自动收到并处理消息。

遇到错误如下

{"@timestamp":"2024-12-09T15:42:10.657+08:00","caller":"mq/consumer.go:55","content":"Dial: Exception (403) Reason: \"no access to this vhost\"","level":"error"}

{"@timestamp":"2024-12-09T15:42:10.657+08:00","caller":"mq/consumer.go:37","content":"Consumer encountered an error and needs to be restarted, error: Exception (403) Reason: \"no access to this vhost\"","level":"error"}

原因:主账号购买的服务,子账号创建的mq用户,没有授权给子账号的mq控制台管理权限

错入如下

{"@timestamp":"2024-12-09T17:56:39.870+08:00","caller":"mq/consumer.go:74","content":"Exchange Declare: Exception (406) Reason: \"ExchangeInBuilt[amq.topic], ReqId:6756BED74138333200852364, ErrorHelp[exchange=amq.topic, https://c.tb.cn/F3.Zro5uI]\"","level":"error"}

{"@timestamp":"2024-12-09T17:56:39.870+08:00","caller":"mq/consumer.go:37","content":"Consumer encountered an error and needs to be restarted, error: Exception (406) Reason: \"ExchangeInBuilt[amq.topic], ReqId:6756BED74138333200852364, ErrorHelp[exchange=amq.topic}

错误原因:官方默认的交换机 amq.topic 是 RabbitMQ 的内置交换机，不能重新声明。需要修改消配置的定义新的交换机go-amp.topic，跳过系统交换机声明的步骤。

### 注意事项：
- 确保你的阿里云 MQ 服务已经开启，并且配置了正确的接入点、密钥和虚拟主机。