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

Posted forlife

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了rabbitMQ 死信 延时队列 [go版本]相关的知识,希望对你有一定的参考价值。

参考文章 https://juejin.cn/post/684490...

准备工作:

需要创建一个交换机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

以上是关于rabbitMQ 死信 延时队列 [go版本]的主要内容,如果未能解决你的问题,请参考以下文章

RabbitMq 实现延时队列-Springboot版本

RabbitMQ的死信队列和延时队列

利用rabbitMq的死信队列实现延时消息

RabbitMQ 死信队列 定时队列 延时队列

springBoot集成rabbitmq 之延时(死信)队列

RabbitMQ实现延时队列(死信队列)