文章目录1、造成死信队列的主要原因2、操作逻辑图3、代码实战3.1

针对原因1#xff1…下面这段教程针对是你已经有一些基本的MQ的知识比如说能够很清楚的理解queue、exchange等概念如果你还不是很理解我建议你先访问官网查看基本的教程。

文章目录1、造成死信队列的主要原因2、操作逻辑图3、代码实战3.1

针对原因1消费者超出时间未应答3.3

针对原因3消费者拒绝的消息回到死信队列中1、造成死信队列的主要原因

2、操作逻辑图

其实整体的思路就是分别创建一个normal_exchange、dead_exchange、normal_queue、dead_queue,然后将normal_exchange与normal_queue进行绑定将dead_exchange与dead_queue进行绑定这里比较关键的一个点在于说如何将normal_queue与dead_exchange进行绑定这样才能将错误的消息传递过来。

下面就是这段代码的关键。

声明一个normal队列_,

ch.QueueDeclare(constant.NormalQueue,true,false,false,false,amqp.Table{//x-message-ttl:

5000,

指定长度。

超过这个长度的消息会发送到dead_exchange中x-dead-letter-exchange:

指定死信交换机x-dead-letter-routing-key:

指定死信routing-key})3.1

github.com/rabbitmq/amqp091-gologv1/utils

)type

ch.ExchangeDeclare(constant.NormalExchange,amqp.ExchangeDirect,true,false,false,false,nil,)utils.FailOnError(err,

Failed

ch.ExchangeDeclare(constant.DeadExchange,amqp.ExchangeDirect,true,false,false,false,nil,)utils.FailOnError(err,

Failed

ch.QueueDeclare(constant.NormalQueue,true,false,false,false,amqp.Table{x-message-ttl:

5000,

指定死信交换机x-dead-letter-routing-key:

指定死信routing-key})utils.FailOnError(err,

Failed

声明一个dead队列:注意不要给死信队列设置消息时间否者死信队列里面的信息会再次过期_,

err

ch.QueueDeclare(constant.DeadQueue,true,false,false,false,nil)utils.FailOnError(err,

Failed

将normal_exchange与normal_queue进行绑定err

ch.QueueBind(constant.NormalQueue,

false,

将dead_exchange与dead_queue进行绑定err

ch.QueueBind(constant.DeadQueue,

false,

ch.Consume(constant.NormalQueue,,false,

这个地方一定要关闭自动应答false,false,false,nil)utils.FailOnError(err,

Failed

github.com/rabbitmq/amqp091-gologv1/utils

)func

ch.ExchangeDeclare(dead_exchange,amqp.ExchangeDirect,true,false,false,false,nil)utils.FailOnError(err,

Failed

ch.Consume(dead_queue,,false,false,false,false,nil,)var

forever

开启手动应答ßd.Ack(false)}}()log.Printf(

[*]

github.com/rabbitmq/amqp091-gostrconvtimev1/utils

)func

ch.ExchangeDeclare(normal_exchange,amqp.ExchangeDirect,true,false,false,false,nil)utils.FailOnError(err,

Failed

context.WithTimeout(context.Background(),

5*time.Second)defer

strconv.Itoa(i)ch.PublishWithContext(ctx,normal_exchange,normal_key,false,false,amqp.Publishing{ContentType:

text/plain,Body:

只需要改变consumer1.go中的对normal_queue的声明

声明一个normal队列_,

ch.QueueDeclare(constant.NormalQueue,true,false,false,false,amqp.Table{//x-message-ttl:

5000,

指定死信交换机x-dead-letter-routing-key:

指定死信routing-key})3.3

message)}}}()工作2:如果你consume的时候开启了自动应答一定要关闭

消费消息msgs,

ch.Consume(constant.NormalQueue,,false,

这个地方一定要关闭自动应答false,false,false,nil)其他的部分不需要改变按照问题1中的设计即可。