快速入门：使用Go语言操作RabbitMQ队列-Golang-PHP中文网

快速入门：使用Go语言操作RabbitMQ队列

尼克

发布： 2025-07-07 12:14:24

原创

653人浏览过

如何在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队列

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 函数
}

登录后复制

这段代码会在连接关闭时打印日志，并尝试重新连接。实际应用中，可以添加更复杂的重连逻辑，比如指数退避算法。

以上就是快速入门：使用Go语言操作RabbitMQ队列的详细内容，更多请关注php中文网其它相关文章！