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版本]的主要内容,如果未能解决你的问题,请参考以下文章