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

本文来自:Segmentfault

感谢作者:forlife

查看原文:rabbitMQ 死信 延时队列 [go版本]

• 编辑
• 预览