脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 脚本专栏 → Golang → ​​​​​​​Golang RabbitMQ死信队列

​​​​​​​Golang实现RabbitMQ中死信队列几种情况

 更新时间：2023年03月01日 09:35:58   作者：ccgkk  
本文主要介绍了​​​​​​​Golang实现RabbitMQ中死信队列几种情况，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧

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

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 day07

import (
    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 day07

import (
    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 day07

import (
    "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死信队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

相关文章

  • Golang 数据库操作(sqlx)和不定字段结果查询

    Golang 数据库操作(sqlx)和不定字段结果查询

    本文主要介绍了Golang 数据库操作(sqlx)和不定字段结果查询，文中通过示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
    2021-09-09
  • 详解go语言 make(chan int, 1) 和 make (chan int) 的区别

    详解go语言 make(chan int, 1) 和 make (chan int) 的区别

    这篇文章主要介绍了go语言 make(chan int, 1) 和 make (chan int) 的区别,本文给大家介绍的非常详细，具有一定的参考借鉴价值，需要的朋友可以参考下
    2020-01-01
  • Go文件操作(新建打开写入读取删除关闭)学习笔记

    Go文件操作(新建打开写入读取删除关闭)学习笔记

    这篇文章主要为大家介绍了Go文件操作(新建打开写入读取删除关闭)学习笔记,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
    2024-01-01
  • Golang 如何判断数组某个元素是否存在 (isset)

    Golang 如何判断数组某个元素是否存在 (isset)

    这篇文章主要介绍了Golang 如何判断数组某个元素是否存在 (isset)，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
    2021-04-04
  • Go并发同步Mutex典型易错使用场景

    Go并发同步Mutex典型易错使用场景

    这篇文章主要为大家介绍了Go并发同步Mutex典型易错使用场景示例详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
    2022-08-08
  • go run main.go 一直提示找不到包的解决方案

    go run main.go 一直提示找不到包的解决方案

    这篇文章主要介绍了go run main.go 一直提示找不到包的解决方案，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
    2021-04-04
  • Go 修改map slice array元素值操作

    Go 修改map slice array元素值操作

    这篇文章主要介绍了Go 修改map slice array元素值操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
    2020-12-12
  • Golang实现AES加密和解密的示例代码

    Golang实现AES加密和解密的示例代码

    AES( advanced encryption standard)使用相同密钥进行加密和解密，也就是对称加密。本文将详细讲解Golang实现AES加密和解密的方法，感兴趣的可以学习一下
    2022-05-05
  • Go语言中排序的3种实现方法

    Go语言中排序的3种实现方法

    在写代码过程中，排序是经常会遇到的需求，这篇文章主要为大家介绍三种常用的方法，文中的示例代码简洁易懂，需要的小伙伴可以参考下
    2023-08-08
  • Go语言中切片使用的注意事项小结

    Go语言中切片使用的注意事项小结

    切片是引用类型，相信对大家来说都不陌生，下面这篇文章主要给大家总结介绍了关于Go语言中切片使用的一些注意事项，文中通过示例代码介绍的非常详细，需要的朋友可以参考借鉴，下面随着小编来一起学习学习吧。
    2018-01-01

最新评论