Golang RabbitMQ 延迟消费队列实现指南

在 RabbitMQ 中，延迟队列是指消息在一定时间后才会被消费者接收到。这种队列通常用于一些需要延迟处理的任务，比如订单超时处理、定时任务等。

Golang 中使用 RabbitMQ 创建延迟消费队列可以通过以下步骤实现：

安装 RabbitMQ Go 客户端

在终端中执行以下命令：

go get github.com/streadway/amqp

创建 RabbitMQ 连接

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

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

创建延迟队列和死信队列

// 声明延迟队列
delayExchange := 'delay_exchange'
delayQueue := 'delay_queue'
err = ch.ExchangeDeclare(delayExchange, 'direct', true, false, false, false, nil)
if err != nil {
    log.Fatalf('failed to declare delay exchange: %v', err)
}

args := amqp.Table{
    'x-dead-letter-exchange':    'dlx_exchange', // 死信交换器
    'x-dead-letter-routing-key': 'dlx_routing_key',
    'x-message-ttl':            10000, // 延迟时间，10s
}
_, err = ch.QueueDeclare(delayQueue, true, false, false, false, args)
if err != nil {
    log.Fatalf('failed to declare delay queue: %v', err)
}

err = ch.QueueBind(delayQueue, '', delayExchange, false, nil)
if err != nil {
    log.Fatalf('failed to bind delay queue to exchange: %v', err)
}

// 声明死信队列
dlxEchange := 'dlx_exchange'
dlxQueue := 'dlx_queue'
err = ch.ExchangeDeclare(dlxEchange, 'direct', true, false, false, false, nil)
if err != nil {
    log.Fatalf('failed to declare dlx exchange: %v', err)
}

_, err = ch.QueueDeclare(dlxQueue, true, false, false, false, nil)
if err != nil {
    log.Fatalf('failed to declare dlx queue: %v', err)
}

err = ch.QueueBind(dlxQueue, 'dlx_routing_key', dlxEchange, false, nil)
if err != nil {
    log.Fatalf('failed to bind dlx queue to exchange: %v', err)
}

发送延迟消息

msg := amqp.Publishing{
    Body: []byte('hello, world'),
}

err = ch.Publish(delayExchange, '', false, false, msg)
if err != nil {
    log.Fatalf('failed to publish message to delay queue: %v', err)
}

接收延迟消息

msgs, err := ch.Consume(delayQueue, '', false, false, false, false, nil)
if err != nil {
    log.Fatalf('failed to consume messages from delay queue: %v', err)
}

for msg := range msgs {
    log.Printf('received message: %s', msg.Body)
    err = ch.Publish(dlxEchange, 'dlx_routing_key', false, false, msg)
    if err != nil {
        log.Fatalf('failed to publish message to dlx: %v', err)
    }
}

在上述代码中，我们首先创建了一个延迟队列和一个死信队列，然后通过向延迟队列发送消息来触发延迟消费。在消费者端，我们使用 Consume 方法从延迟队列中获取消息，并将消息重新发送到死信队列中，以便进行后续处理。

需要注意的是，在消息重新发送到死信队列时，需要将消息的 DeliveryTag 设置为原始消息的 DeliveryTag，以便 RabbitMQ 能够正确地标识和处理消息。