Golang RabbitMQ 推送延迟消息

在 Go 语言中使用 RabbitMQ 推送延迟消息可以通过以下步骤实现：

安装 RabbitMQ 客户端库

可以使用以下命令在终端中安装 RabbitMQ 的 Go 客户端库：

go get github.com/streadway/amqp

创建一个 RabbitMQ 连接和通道

使用以下代码创建一个 RabbitMQ 连接和通道：

conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
    log.Fatalf("Failed to connect to RabbitMQ: %s", err)
}
defer conn.Close()

ch, err := conn.Channel()
if err != nil {
    log.Fatalf("Failed to open a channel: %s", err)
}
defer ch.Close()

声明一个延迟队列

使用以下代码声明一个延迟队列：

args := make(amqp.Table)
args["x-delayed-type"] = "direct"
err = ch.ExchangeDeclare(
    "delayed", // exchange name
    "x-delayed-message", // exchange type
    true, // durable
    false, // auto-deleted
    false, // internal
    false, // no-wait
    args, // arguments
)
if err != nil {
    log.Fatalf("Failed to declare an exchange: %s", err)
}

在这里我们使用了延迟消息插件提供的x-delayed-message交换机类型来声明一个延迟队列。

发送一个延迟消息

使用以下代码发送一个延迟消息：

msg := amqp.Publishing{
    ContentType: "text/plain",
    Body:        []byte("Hello, world!"),
    Headers: amqp.Table{
        "x-delay": 5000, // delay in milliseconds
    },
}
err = ch.Publish(
    "delayed", // exchange
    "my-delayed-queue", // routing key
    false, // mandatory
    false, // immediate
    msg, // message
)
if err != nil {
    log.Fatalf("Failed to publish a message: %s", err)
}

在这里我们在消息的Header中添加了一个x-delay字段来指定消息的延迟时间，单位为毫秒。

完整代码示例：

package main

import (
    "log"
    "time"

    "github.com/streadway/amqp"
)

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %s", err)
    }
defer conn.Close()

    ch, err := conn.Channel()
if err != nil {
        log.Fatalf("Failed to open a channel: %s", err)
    }
defer ch.Close()

    args := make(amqp.Table)
args["x-delayed-type"] = "direct"
    err = ch.ExchangeDeclare(
        "delayed", // exchange name
        "x-delayed-message", // exchange type
        true, // durable
        false, // auto-deleted
        false, // internal
        false, // no-wait
        args, // arguments
    )
if err != nil {
        log.Fatalf("Failed to declare an exchange: %s", err)
    }

    msg := amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte("Hello, world!"),
        Headers: amqp.Table{
            "x-delay": 5000, // delay in milliseconds
        },
    }
    err = ch.Publish(
        "delayed", // exchange
        "my-delayed-queue", // routing key
        false, // mandatory
        false, // immediate
        msg, // message
    )
if err != nil {
        log.Fatalf("Failed to publish a message: %s", err)
    }

    log.Println("Message sent")

    // Wait for the message to be delivered
    time.Sleep(10 * time.Second)
}