RabbitMQ延迟消息配置异常：设置x-delay头后消息仍立即投递

核心原因 #

直接给普通队列发送带x-delay头的消息不会生效，这个头是RabbitMQ延迟消息插件的专属特性，只有发送到延迟交换机时才会被处理。另外，你使用的rabbitmq:3-management-alpine镜像默认未预装该插件，需手动安装。

步骤1：安装延迟消息插件 #

1. 查看RabbitMQ容器ID：docker ps
2. 进入容器终端：docker exec -it <容器ID> sh
3. 启用延迟插件：rabbitmq-plugins enable rabbitmq_delayed_message_exchange
4. 重启容器使插件生效：docker restart <容器ID>

步骤2：修改代码使用延迟交换机 #

修改后的send.go #

package main

import (
    "context"
    "log"
    "time"

    amqp "github.com/rabbitmq/amqp091-go"
)

func failOnError(err error, msg string) {
    if err != nil {
        log.Panicf("%s: %s", msg, err)
    }
}

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    ch, err := conn.Channel()
    failOnError(err, "Failed to open a channel")
    defer ch.Close()

    // 声明延迟交换机
    err = ch.ExchangeDeclare(
        "delayed_exchange",       // 交换机名称
        "x-delayed-exchange",     // 延迟交换机专属类型
        true,                     // 持久化
        false,                    // 不再使用时删除
        false,                    // 排他
        false,                    // 不等待
        amqp.Table{
            "x-delayed-type": "direct", // 指定底层路由类型，此处用direct
        },
    )
    failOnError(err, "Failed to declare delayed exchange")

    // 声明队列
    q, err := ch.QueueDeclare(
        "hello", // 队列名
        true,    // 持久化（建议开启，避免重启丢失）
        false,   // 不再使用时删除
        false,   // 排他
        false,   // 不等待
        nil,     // 参数
    )
    failOnError(err, "Failed to declare a queue")

    // 绑定交换机与队列
    err = ch.QueueBind(
        q.Name,               // 队列名
        q.Name,               // 路由键，与队列名一致确保消息路由到目标队列
        "delayed_exchange",   // 交换机名
        false,
        nil,
    )
    failOnError(err, "Failed to bind queue to exchange")

    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    body := "Hello World!"
    err = ch.PublishWithContext(ctx,
        "delayed_exchange", // 发送到延迟交换机
        q.Name,             // 路由键
        false,
        false,
        amqp.Publishing{
            Headers: amqp.Table{
                "x-delay": 5000, // 延迟5秒，单位毫秒
            },
            ContentType: "text/plain",
            Body:        []byte(body),
        })
    failOnError(err, "Failed to publish a message")
    log.Printf(" [x] Sent %s (will be delivered after 5s)\n", body)
}

修改后的receive.go #

package main

import (
    "log"

    amqp "github.com/rabbitmq/amqp091-go"
)

func failOnError(err error, msg string) {
    if err != nil {
        log.Panicf("%s: %s", msg, err)
    }
}

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    ch, err := conn.Channel()
    failOnError(err, "Failed to open a channel")
    defer ch.Close()

    // 同步声明延迟交换机（与发送端配置一致）
    err = ch.ExchangeDeclare(
        "delayed_exchange",
        "x-delayed-exchange",
        true,
        false,
        false,
        false,
        amqp.Table{
            "x-delayed-type": "direct",
        },
    )
    failOnError(err, "Failed to declare delayed exchange")

    // 声明队列（与发送端配置一致）
    q, err := ch.QueueDeclare(
        "hello",
        true,
        false,
        false,
        false,
        nil,
    )
    failOnError(err, "Failed to declare a queue")

    // 绑定队列到交换机（与发送端配置一致）
    err = ch.QueueBind(
        q.Name,
        q.Name,
        "delayed_exchange",
        false,
        nil,
    )
    failOnError(err, "Failed to bind queue to exchange")

    msgs, err := ch.Consume(
        q.Name, // 监听目标队列
        "",
        true,
        false,
        false,
        false,
        nil,
    )
    failOnError(err, "Failed to register a consumer")

    var forever chan struct{}

    go func() {
        for d := range msgs {
            log.Printf("Received a message: %s", d.Body)
        }
    }()

    log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
    <-forever
}

关键说明 #

• 延迟交换机必须声明为x-delayed-exchange类型，并通过x-delayed-type指定底层路由规则（如direct、topic等）
• 消息必须发送到延迟交换机，x-delay头才会被插件识别并执行延迟投递逻辑
• 建议开启队列和交换机的持久化配置，避免RabbitMQ重启后丢失绑定关系与未投递的延迟消息

内容的提问来源于stack exchange，提问作者Lexx