go语言zero框架对接阿里云消息队列MQ的rabbit的配置与调用
在 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 main
import (
"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 main
import (
"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 服务已经开启，并且配置了正确的接入点、密钥和虚拟主机。
- 可以根据需求选择是否启用持久化、自动确认等选项。
- 在生产环境中，确保对消息队列进行有效的监控和错误处理，以保证系统的稳定性和可靠性。
通过这种方式，你就可以在 Go 语言中对接阿里云的消息队列（MQ）并实现与 RabbitMQ 的基本交互。