如何在go语言中操作rabbitmq队列？1. 建立连接：使用amqp.dial函数并传入正确的连接字符串，通过go get github.com/rabbitmq/amqp091-go安装依赖库；2. 声明队列：通过conn.channel()创建通道后，使用ch.queuedeclare声明队列，并设置名称、持久化、自动删除等参数；3. 发送消息：调用ch.publishwithcontext函数，指定交换机、路由键和消息内容，建议设置上下文超时时间；4. 接收消息：使用ch.consume订阅队列，从返回的channel中读取消息，并设置自动应答以确保消息正确处理；5. 处理连接中断：通过conn.notifyclose监听连接关闭事件，并实现重连机制如循环重试或指数退避算法，确保程序健壮性。以上步骤完整覆盖了go语言操作rabbitmq的核心流程。
Go语言操作RabbitMQ队列，核心在于连接、声明、发送和接收。掌握这些，就能快速上手。
连接RabbitMQ，声明队列，然后就可以愉快的发送和接收消息了。当然，错误处理是少不了的，别忘了加上。
如何在Go中建立RabbitMQ连接？
连接RabbitMQ，首先你需要amqp这个库。go get github.com/rabbitmq/amqp091-go安装一下。然后，用amqp.Dial函数建立连接。连接字符串一般是amqp://user:password@host:port/vhost。别忘了处理连接错误，不然程序可能直接崩了。
立即学习“go语言免费学习笔记（深入）”；
package main
import (
"fmt"
"log"
amqp "github.com/rabbitmq/amqp091-go"
)
func main() {
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
log.Fatalf("连接RabbitMQ失败: %s", err)
}
defer conn.Close()
fmt.Println("成功连接到RabbitMQ!")
}这段代码只是建立连接，没什么复杂的。如果连接失败，程序会直接退出。
如何声明一个RabbitMQ队列？
有了连接，下一步就是声明队列。队列是消息的容器，发送消息之前必须先声明队列。使用channel.QueueDeclare函数声明队列，可以设置队列的名称、是否持久化、是否自动删除等等。
package main
import (
"fmt"
"log"
amqp "github.com/rabbitmq/amqp091-go"
)
func main() {
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
log.Fatalf("连接RabbitMQ失败: %s", err)
}
defer conn.Close()
ch, err := conn.Channel()
if err != nil {
log.Fatalf("打开通道失败: %s", err)
}
defer ch.Close()
q, err := ch.QueueDeclare(
"my_queue", // 队列名称
false, // 是否持久化
false, // 是否自动删除
false, // 是否排他
false, // 是否阻塞
nil, // 参数
)
if err != nil {
log.Fatalf("声明队列失败: %s", err)
}
fmt.Printf("声明队列成功: %v\n", q)
}这里声明了一个名为my_queue的队列，非持久化，非自动删除，非排他，非阻塞。你可以根据实际需求修改这些参数。
如何发送消息到RabbitMQ队列？
声明了队列，就可以发送消息了。使用channel.Publish函数发送消息，需要指定交换机、路由键、消息内容等。
package main
import (
"context"
"log"
"time"
amqp "github.com/rabbitmq/amqp091-go"
)
func main() {
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
log.Fatalf("连接RabbitMQ失败: %s", err)
}
defer conn.Close()
ch, err := conn.Channel()
if err != nil {
log.Fatalf("打开通道失败: %s", err)
}
defer ch.Close()
q, err := ch.QueueDeclare(
"my_queue", // 队列名称
false, // 是否持久化
false, // 是否自动删除
false, // 是否排他
false, // 是否阻塞
nil, // 参数
)
if err != nil {
log.Fatalf("声明队列失败: %s", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
body := "Hello, RabbitMQ!"
err = ch.PublishWithContext(ctx,
"", // 交换机
q.Name, // 路由键
false, // 是否强制
false, // 是否立即
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(body),
})
if err != nil {
log.Fatalf("发布消息失败: %s", err)
}
log.Printf("已发送消息: %s", body)
}这段代码发送了一条简单的文本消息到my_queue队列。注意，这里使用了context.WithTimeout设置了超时时间，防止程序一直阻塞。
如何从RabbitMQ队列接收消息？
发送了消息，当然要接收。使用channel.Consume函数订阅队列，然后就可以从返回的通道中读取消息了。
package main
import (
"fmt"
"log"
amqp "github.com/rabbitmq/amqp091-go"
)
func main() {
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
log.Fatalf("连接RabbitMQ失败: %s", err)
}
defer conn.Close()
ch, err := conn.Channel()
if err != nil {
log.Fatalf("打开通道失败: %s", err)
}
defer ch.Close()
q, err := ch.QueueDeclare(
"my_queue", // 队列名称
false, // 是否持久化
false, // 是否自动删除
false, // 是否排他
false, // 是否阻塞
nil, // 参数
)
if err != nil {
log.Fatalf("声明队列失败: %s", err)
}
msgs, err := ch.Consume(
q.Name, // 队列名称
"", // 消费者标签
true, // 是否自动应答
false, // 是否排他
false, // 是否本地
false, // 是否阻塞
nil, // 参数
)
if err != nil {
log.Fatalf("消费消息失败: %s", err)
}
forever := make(chan bool)
go func() {
for d := range msgs {
log.Printf("接收到消息: %s", d.Body)
}
}()
log.Printf("等待接收消息...")
<-forever
}这段代码订阅了my_queue队列，并打印接收到的消息。注意，这里使用了forever通道，防止程序退出。true表示自动应答，消息一旦被接收，就会自动从队列中删除。
如何处理RabbitMQ连接中断？
RabbitMQ连接可能会因为各种原因中断，比如网络故障、服务器重启等。为了保证程序的健壮性，需要处理连接中断的情况。可以使用amqp.Connection.NotifyClose函数监听连接关闭事件，并在连接关闭时重新建立连接。
package main
import (
"log"
"time"
amqp "github.com/rabbitmq/amqp091-go"
)
func connectToRabbitMQ() (*amqp.Connection, error) {
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
return nil, err
}
return conn, nil
}
func main() {
var conn *amqp.Connection
var err error
for {
conn, err = connectToRabbitMQ()
if err == nil {
break
}
log.Printf("连接RabbitMQ失败: %s, 5秒后重试", err)
time.Sleep(5 * time.Second)
}
defer conn.Close()
log.Println("成功连接到RabbitMQ!")
// 监听连接关闭事件
closeChan := conn.NotifyClose(make(chan *amqp.Error))
<-closeChan
log.Println("RabbitMQ连接已关闭，尝试重新连接...")
// 这里可以添加重新连接的逻辑，比如重新调用 connectToRabbitMQ 函数
}这段代码会在连接关闭时打印日志，并尝试重新连接。实际应用中，可以添加更复杂的重连逻辑，比如指数退避算法。
