创建网站好的平台,西宁知名网站制作公司,海口官网设计,cname解析对网站影响1、概念 死信#xff08;Dead Letter#xff09;#xff0c;字面上可以理解为未被消费者成功消费的信息#xff0c;正常来说#xff0c;生产者将消息放入到队列中#xff0c;消费者从队列获取消息#xff0c;并进行处理#xff0c;但是由于某种原因#xff0c;队列中的…1、概念 死信Dead Letter字面上可以理解为未被消费者成功消费的信息正常来说生产者将消息放入到队列中消费者从队列获取消息并进行处理但是由于某种原因队列中的消息未被消费者拿到这样的消息就会成为死信存放死信消息的队列也就被称为死信队列(Dead Letter Queue简称DLQ)。
2、死信产生的原因
文心一言的回答如下
消息被拒绝当消费者使用basic.reject或basic.nack方法拒绝消息并且requeue参数被设置为false时消息会被视为死信。这意味着消费者明确表示无法或不愿意处理该消息并且不希望该消息重新进入队列等待其他消费者处理。消息处理失败消费者由于代码错误、消息格式不正确、业务规则冲突等原因无法成功处理消息时该消息也可以被标记为死信。这种情况下尽管消费者尝试处理消息但由于某些无法克服的错误消息无法被成功消费。消息过期如果消息设置了生存时间TTLTime To Live并且在这个时间内没有被消费那么消息会过期并被视为死信。TTL是RabbitMQ中用于指定消息在队列中存活时间的参数超过该时间的消息将被视为过期并丢弃或转发到死信队列。队列长度限制当队列中的消息数量超过了设置的最大长度时新到达的消息无法进入队列这些消息也会被视为死信。队列长度限制是RabbitMQ中用于控制队列大小的一种机制当队列达到最大容量时新到达的消息将无法被接收并可能被丢弃或转发到死信队列。
总结来说主要原因就三个消息被拒绝、消息过期、队列满
在一些重要的场景比如支付场景提交的订单超时未支付的可以设计为进入死信队列。
3、死信队列使用实践
3.1 消息过期
设置正常队列ttl过期时间为5s如果5s内消息没有被消费则会自动放入死信队列中。
关键点设置正常队列属性ttl5s过期
// 设置TTL并让消息过期进入死信队列,5s不消费就为进入死信队列args : amqp.Table{x-message-ttl: int64(5000), // 5秒TTLx-dead-letter-exchange: ,x-dead-letter-routing-key: dlx.Name,}// 声明正常队列q, err : ch.QueueDeclare(normal_queue, // nametrue, // durablefalse, // delete when unusedfalse, // exclusivefalse, // no-waitargs, // arguments)
全部代码如下
package mainimport (fmtamqp github.com/rabbitmq/amqp091-go
)func main() {conn, err : amqp.Dial(amqp://guest:guestxxxx.xx.xx.xxx:5672/)if err ! nil {fmt.Println(Failed to connect to RabbitMQ)return}defer conn.Close()ch, err : conn.Channel()if err ! nil {fmt.Println(Failed to open a channel)return}// 声明死信队列dlx, err : ch.QueueDeclare(dead_letter_queue, // nametrue, // durablefalse, // delete when unusedfalse, // exclusivefalse, // no-waitnil, // arguments)if err ! nil {fmt.Println(Failed to declare a queue: dead_letter_queue,err: err.Error())return}err ch.ExchangeDeclare(my_exchange, // namedirect, // typetrue, // durablefalse, // auto-deletedfalse, // internalfalse, // no-waitnil, // arguments)if err ! nil {fmt.Println(Failed to declare exchange: my_exchange,err: err.Error())return}// 设置TTL并让消息过期进入死信队列,5s不消费就为进入死信队列args : amqp.Table{x-message-ttl: int64(5000), // 5秒TTLx-dead-letter-exchange: ,x-dead-letter-routing-key: dlx.Name,}// 声明正常队列,注意必须在声明队列时就要设置死信队列信息q, err : ch.QueueDeclare(normal_queue, // nametrue, // durablefalse, // delete when unusedfalse, // exclusivefalse, // no-waitargs, // arguments)if err ! nil {fmt.Println(Failed to declare a queue: normal_queue,err: err.Error())return}// 将正常队列绑定到交换机并设置死信交换机和路由键err ch.QueueBind(q.Name, // queue nameq.Name, // routing keymy_exchange, // exchangefalse,nil,)if err ! nil {fmt.Println(Failed to bind queue to exchange: my_exchange,err: err.Error())return}err ch.Publish(my_exchange, normal_queue, false, false, amqp.Publishing{Body: []byte(hello world)})if err ! nil {fmt.Println(Failed to publis msg to exchange: my_exchange,err: err.Error())return}
}队列信息包括绑定的死信队列信息、ttl等信息如下 运行上方代码会向队列发送一条信息我们先不创建消费者5s后消息会被自动放入死信队列。 3.2 队列满
当mq队列由于消息量过多导致队列打满时这个时候过来的消息将会被自动放入到死信队列中。
设置队列长度属性代码如下
args : amqp.Table{// x-message-ttl: int64(5000), // 5秒TTLx-max-length: 2,x-dead-letter-exchange: ,x-dead-letter-routing-key: dlx.Name,}// 声明正常队列q, err : ch.QueueDeclare(normal_queue, // nametrue, // durablefalse, // delete when unusedfalse, // exclusivefalse, // no-waitargs, // arguments)
队列属性如下 发送两条信息 继续发送第三个 测试代码
package mainimport (fmtamqp github.com/rabbitmq/amqp091-go
)func main() {conn, err : amqp.Dial(amqp://guest:guestxx.xx.xx.xx:5672/)if err ! nil {fmt.Println(Failed to connect to RabbitMQ)return}defer conn.Close()ch, err : conn.Channel()if err ! nil {fmt.Println(Failed to open a channel)return}// 声明死信队列dlx, err : ch.QueueDeclare(dead_letter_queue, // nametrue, // durablefalse, // delete when unusedfalse, // exclusivefalse, // no-waitnil, // arguments)if err ! nil {fmt.Println(Failed to declare a queue: dead_letter_queue,err: err.Error())return}err ch.ExchangeDeclare(my_exchange, // namedirect, // typetrue, // durablefalse, // auto-deletedfalse, // internalfalse, // no-waitnil, // arguments)if err ! nil {fmt.Println(Failed to declare exchange: my_exchange,err: err.Error())return}// 设置TTL并让消息过期进入死信队列,5s不消费就为进入死信队列args : amqp.Table{// x-message-ttl: int64(5000), // 5秒TTLx-max-length: 2,x-dead-letter-exchange: ,x-dead-letter-routing-key: dlx.Name,}// 声明正常队列q, err : ch.QueueDeclare(normal_queue, // nametrue, // durablefalse, // delete when unusedfalse, // exclusivefalse, // no-waitargs, // arguments)if err ! nil {fmt.Println(Failed to declare a queue: normal_queue,err: err.Error())return}// 将正常队列绑定到交换机并设置死信交换机和路由键err ch.QueueBind(q.Name, // queue nameq.Name, // routing keymy_exchange, // exchangefalse,nil,)if err ! nil {fmt.Println(Failed to bind queue to exchange: my_exchange,err: err.Error())return}err ch.Publish(my_exchange, normal_queue, false, false, amqp.Publishing{Body: []byte(hello world)})if err ! nil {fmt.Println(Failed to publis msg to exchange: my_exchange,err: err.Error())return}
}3.3 消息被拒绝 消息被拒绝的情况当消费者无法处理某条信息时客户端想rabbitmq服务器发送一个【负确认】应答表示消费者未能成功处理此条消息并且希望RabbitMQ根据配置重新发送这条消息例如将其重新排队或者将其丢弃。
客户端函数ch.Nack函数原型
func (ch *Channel) Nack(tag uint64, multiple bool, requeue bool) error {ch.m.Lock()defer ch.m.Unlock()return ch.send(basicNack{DeliveryTag: tag,Multiple: multiple,Requeue: requeue,})
}
入参含义如下
tag 这是一个唯一标识符用于标识消费者之前接收到的特定消息。当消费者调用 ch.Ack、ch.Nack 或 ch.Reject 时必须提供这个标识符以便RabbitMQ知道是对哪条消息进行确认或拒绝。 multiple 这是一个布尔值bool用于指示是否应该同时确认或拒绝多条消息。如果设置为 true则RabbitMQ将认为从上一个被确认的消息开始包括该消息直到当前消息为止的所有未确认消息都被拒绝。这通常用于批量处理消息确认但在使用 ch.Nack 时它的作用更多是关于是否应该重新排队当前消息之后的消息取决于RabbitMQ的配置和消息的属性。 requeue 这也是一个布尔值bool用于指示被拒绝的消息是否应该被重新放入队列的末尾以便稍后重试。如果设置为 true则消息将被重新排队如果设置为 false则消息将被丢弃或者根据RabbitMQ的配置可能被发送到死信队列如果配置了的话。
测试过程首先使用3.1或者3.2的代码向mq中写入几条信息: 之后使用如下代码进行消费
package mainimport (fmttimeamqp github.com/rabbitmq/amqp091-go
)func main() {conn, err : amqp.Dial(amqp://guest:guestxx.xx.xx.xx:5672/)if err ! nil {fmt.Println(Failed to connect to RabbitMQ)return}defer conn.Close()ch, err : conn.Channel()if err ! nil {fmt.Println(Failed to open a channel)return}err ch.ExchangeDeclare(my_exchange, // namedirect, // typetrue, // durablefalse, // auto-deletedfalse, // internalfalse, // no-waitnil, // arguments)if err ! nil {fmt.Println(Failed to declare exchange: my_exchange,err: err.Error())return}// 声明正常队列// q, err : ch.QueueDeclare(// normal_queue, // name// true, // durable// false, // delete when unused// false, // exclusive// false, // no-wait// nil, // arguments// )// if err ! nil {// fmt.Println(Failed to declare a queue: normal_queue,err: err.Error())// return// }// 将正常队列绑定到交换机并设置死信交换机和路由键err ch.QueueBind(normal_queue, // queue namenormal_queue, // routing keymy_exchange, // exchangefalse,nil,)if err ! nil {fmt.Println(Failed to bind queue to exchange: my_exchange,err: err.Error())return}msgs, _ : ch.Consume(normal_queue, // queue, // consumerfalse, // auto-acktrue, // exclusivefalse, // no-localfalse, // no-waitnil, // args)go func() {for d : range msgs {// 模拟处理失败,全部放入死信队列ch.Nack(d.DeliveryTag, false, false)}}()time.Sleep(10 * time.Second)
}运行代码后3条消息全部进入到死信队列中 4、总结 RabbitMQ的死信队列Dead Letter Queue简称DLQ是一种用于处理消息失败或无法路由的消息的机制死信队列中的所有消息都是无法被正常消费的死信这使得开发者可以集中对这些消息进行管理和分析。通过分析死信队列中的消息开发者可以了解系统的运行状态、发现潜在的问题并进行相应的优化和改进以提升系统的稳定性和可靠性。
