​​​​​​​Golang实现RabbitMQ中死信队列的情况有哪些

这篇文章主要讲解了“Golang实现RabbitMQ中死信队列的情况有哪些”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“Golang实现RabbitMQ中死信队列的情况有哪些”吧！

1、造成死信队列的主要原因

• 消费者超时未应答

• 队列的容量有限

• 消费者拒绝了的消息

2、操作逻辑图

3、代码实战

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

// 声明一个normal队列    _, err = ch.QueueDeclare(        constant.NormalQueue,        true,        false,        false,        false,        amqp.Table{            //"x-message-ttl":             5000,                    // 指定过期时间            //"x-max-length":              6,                        // 指定长度。超过这个长度的消息会发送到dead_exchange中            "x-dead-letter-exchange":    constant.DeadExchange,    // 指定死信交换机            "x-dead-letter-routing-key": constant.DeadRoutingKey,  // 指定死信routing-key        })

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

consumer1.go

package day07import (    amqp "github.com/rabbitmq/amqp091-go"    "log"    "v1/utils")type Constant struct {    NormalExchange   string    DeadExchange     string    NormalQueue      string    DeadQueue        string    NormalRoutingKey string    DeadRoutingKey   string}func Consumer1() {    // 获取连接    ch := utils.GetChannel()    // 创建一个变量常量    constant := Constant{        NormalExchange:   "normal_exchange",        DeadExchange:     "dead_exchange",        NormalQueue:      "normal_queue",        DeadQueue:        "dead_queue",        NormalRoutingKey: "normal_key",        DeadRoutingKey:   "dead_key",    }    // 声明normal交换机    err := ch.ExchangeDeclare(        constant.NormalExchange,        amqp.ExchangeDirect,        true,        false,        false,        false,        nil,    )    utils.FailOnError(err, "Failed to declare a normal exchange")    // 声明一个dead交换机    err = ch.ExchangeDeclare(        constant.DeadExchange,        amqp.ExchangeDirect,        true,        false,        false,        false,        nil,    )    utils.FailOnError(err, "Failed to declare a dead exchange")    // 声明一个normal队列    _, err = ch.QueueDeclare(        constant.NormalQueue,        true,        false,        false,        false,        amqp.Table{            "x-message-ttl": 5000, // 指定过期时间            //"x-max-length":              6,            "x-dead-letter-exchange":    constant.DeadExchange,   // 指定死信交换机            "x-dead-letter-routing-key": constant.DeadRoutingKey, // 指定死信routing-key        })    utils.FailOnError(err, "Failed to declare a normal queue")    // 声明一个dead队列:注意不要给死信队列设置消息时间，否者死信队列里面的信息会再次过期    _, err = ch.QueueDeclare(        constant.DeadQueue,        true,        false,        false,        false,        nil)    utils.FailOnError(err, "Failed to declare a dead queue")    // 将normal_exchange与normal_queue进行绑定    err = ch.QueueBind(constant.NormalQueue, constant.NormalRoutingKey, constant.NormalExchange, false, nil)    utils.FailOnError(err, "Failed to binding normal_exchange with normal_queue")    // 将dead_exchange与dead_queue进行绑定    err = ch.QueueBind(constant.DeadQueue, constant.DeadRoutingKey, constant.DeadExchange, false, nil)    utils.FailOnError(err, "Failed to binding dead_exchange with dead_queue")    // 消费消息    msgs, err := ch.Consume(constant.NormalQueue,        "",        false, // 这个地方一定要关闭自动应答        false,        false,        false,        nil)    utils.FailOnError(err, "Failed to consume in Consumer1")    var forever chan struct{}    go func() {        for d := range msgs {            if err := d.Reject(false); err != nil {                utils.FailOnError(err, "Failed to Reject a message")            }        }    }()    log.Printf(" [*] Waiting for logs. To exit press CTRL+C")    <-forever}

consumer2.go

package day07import (    amqp "github.com/rabbitmq/amqp091-go"    "log"    "v1/utils")func Consumer2() {    // 拿取信道    ch := utils.GetChannel()    // 声明一个交换机    err := ch.ExchangeDeclare(        "dead_exchange",        amqp.ExchangeDirect,        true,        false,        false,        false,        nil)    utils.FailOnError(err, "Failed to Declare a exchange")    // 接收消息的应答    msgs, err := ch.Consume("dead_queue",        "",        false,        false,        false,        false,        nil,    )    var forever chan struct{}    go func() {        for d := range msgs {            log.Printf("[x] %s", d.Body)            // 开启手动应答ß            d.Ack(false)        }    }()    log.Printf(" [*] Waiting for logs. To exit press CTRL+C")    <-forever}

produce.go

package day07import (    "context"    amqp "github.com/rabbitmq/amqp091-go"    "strconv"    "time"    "v1/utils")func Produce() {    // 获取信道    ch := utils.GetChannel()    // 声明一个交换机    err := ch.ExchangeDeclare(        "normal_exchange",        amqp.ExchangeDirect,        true,        false,        false,        false,        nil)    utils.FailOnError(err, "Failed to declare a exchange")    ctx, cancer := context.WithTimeout(context.Background(), 5*time.Second)    defer cancer()    // 发送了10条消息    for i := 0; i < 10; i++ {        msg := "Info:" + strconv.Itoa(i)        ch.PublishWithContext(ctx,            "normal_exchange",            "normal_key",            false,            false,            amqp.Publishing{                ContentType: "text/plain",                Body:        []byte(msg),            })    }}

3.2 针对原因2：限制一定的长度

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

// 声明一个normal队列    _, err = ch.QueueDeclare(        constant.NormalQueue,        true,        false,        false,        false,        amqp.Table{            //"x-message-ttl": 5000, // 指定过期时间            "x-max-length":              6,            "x-dead-letter-exchange":    constant.DeadExchange,   // 指定死信交换机            "x-dead-letter-routing-key": constant.DeadRoutingKey, // 指定死信routing-key        })

3.3 针对原因3：消费者拒绝的消息回到死信队列中

这里需要完成两点工作

工作1:需要在consumer1中作出拒绝的操作

go func() {        for d := range msgs {            if err := d.Reject(false); err != nil {                utils.FailOnError(err, "Failed to Reject a message")            }        }    }()

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

// 消费消息    msgs, err := ch.Consume(constant.NormalQueue,        "",        false, // 这个地方一定要关闭自动应答        false,        false,        false,        nil)

其他的部分不需要改变，按照问题1中的设计即可。

感谢各位的阅读，以上就是“Golang实现RabbitMQ中死信队列的情况有哪些”的内容了，经过本文的学习后，相信大家对Golang实现RabbitMQ中死信队列的情况有哪些这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是编程网，小编将为大家推送更多相关知识点的文章，欢迎关注！