rabbitMQ 死信延时队列 [go版本]

forlife · · 746 次点击 · · 开始浏览

这是一个创建于的文章，其中的信息可能已经有所发展或是发生改变。

准备工作：

需要创建一个交换机j_dlx
需要创建一个队列j_dlx_queue
j_dlx 与 j_dlx_queue 绑定在一起交换机类型4种任一种

produer.go 生产者

package main

import (
    "fmt"
    "github.com/streadway/amqp"
    "time"
)

//因：快速实现逻辑，故：不处理错误逻辑
func main() {
    conn, _ := amqp.Dial("amqp://user:password@host:ip/vhost")
    ch, _ := conn.Channel()
    body := "Hello World " + time.Now().Format("2006-01-02 15:04:05")
    fmt.Println(body)
    //声明交换器
    args := amqp.Table{"x-dead-letter-exchange": "j_dlx"}
    // 声明一个j_queue_normal队列，并设置队列的死信交换机为"j_dlx"
    q, _ := ch.QueueDeclare("j_queue_normal", true, false, false, false, args)
    ch.Publish("", q.Name, false, false, amqp.Publishing{
        Body:       []byte(body),
        Expiration: "5000", // 设置TTL为5秒
    })

    // defer 关键字
    defer conn.Close() // 压栈 后进先出
    defer ch.Close()   // 压栈 后进先出
}

$ go run producer.go
Hello World 2021-03-19 17:40:45

consumer.go 消费者

package main

import (
    "fmt"
    "github.com/streadway/amqp"
)

func main() {
    conn, _ := amqp.Dial("amqp://user:password@host:ip/vhost")
    ch, _ := conn.Channel()

    //监听dlxQueue队列
    msgs, _ := ch.Consume("j_dxl_queue", "", true, false, false, false, nil)

    //fmt.Println(msgs)
    for d := range msgs {
        fmt.Printf("receive: %s\n", d.Body) // 收到消息，业务处理
    }
}

$ go run consumer.go
receive: Hello World 2021-03-19 17:40:37
receive: Hello World 2021-03-19 17:40:38
receive: Hello World 2021-03-19 17:40:40
receive: Hello World 2021-03-19 17:40:41
receive: Hello World 2021-03-19 17:40:44
receive: Hello World 2021-03-19 17:40:45

有疑问加站长微信联系（非本文作者）