开发青年网站,建设局网站项目负责人资质要求,网络推广合作资源平台,四大工业设计软件文章目录 前言一、延时队列与应用场景二、RabbitMQ如何实现延时队列实现延时队列的基本要素整体的实现原理如下 三、Go语言实战生产者消费者 前言
之前做秒杀商城项目的时候使用到了延时队列来解决订单超时问题#xff0c;本博客就总结一下Golang是如何利用RabbitMQ实现的延时… 文章目录 前言一、延时队列与应用场景二、RabbitMQ如何实现延时队列实现延时队列的基本要素整体的实现原理如下 三、Go语言实战生产者消费者 前言
之前做秒杀商城项目的时候使用到了延时队列来解决订单超时问题本博客就总结一下Golang是如何利用RabbitMQ实现的延时队列的。
一、延时队列与应用场景
延迟队列是一种特殊类型的消息队列用于在一定时间后将消息投递给消费者。它可以用于处理需要延迟执行的任务或者具有定时特性的业务场景。使用延迟队列可以灵活地控制消息的发送和处理时间适用于很多场景如订单超时处理、提醒任务等
具体应用场景有如下 订单取消当订单生成时将订单消息发送到延迟队列中并设置延迟时间为十分钟。消费者在十分钟后接收到订单消息并进行关闭操作。 店铺商品提醒在店铺创建时将提醒消息发送到延迟队列中并设置延迟时间为十天。消费者在十天后接收到消息并发送提醒通知。 用户登录提醒用户注册成功后将提醒消息发送到延迟队列中并设置延迟时间为三天。消费者在三天后接收到消息并发送短信提醒。 退款通知当用户发起退款时将通知消息发送到延迟队列中并设置延迟时间为三天。消费者在三天后接收到消息并通知相关运营人员。 会议提醒在会议预定时将提醒消息发送到延迟队列中并设置延迟时间为预定时间前十分钟。消费者在指定时间点前十分钟接收到消息并发送会议参加通知。
通过使用延迟队列可以在指定的时间点触发任务避免了轮询的低效方式并且能够满足大量数据和时效性的需求。这种方法提供了更高的性能和实时性并有效减轻了系统的负载。 下图是订单超时处理的流程图。
二、RabbitMQ如何实现延时队列
虽然 rabbitmq 没有延时队列的功能,但是稍微变动一下也是可以实现的。 通过设置消息的 TTL 和 DLX 等参数可以将消息转发到一个指定的队列中以便在一定的时间后再进行处理。
实现延时队列的基本要素
1、存在一个倒计时机制:Time To Live(TTL) 2、当到达时间点的时候会触发一个发送消息的事件:Dead Letter ExchangesDLX 基于第一点,我利用的是消息存在过期时间这一特性, 消息一旦过期就会变成dead letter,可以让单独的消息过期,也可以设置整个队列消息的过期时间 而rabbitmq会有限取两个值的最小 **基于第二点,**是用到了rabbitmq的过期消息处理机制: . x-dead-letter-exchange 将过期的消息发送到指定的 exchange 中 . x-dead-letter-routing-key 将过期的消息发送到自定的 route当中
整体的实现原理如下
发送者将消息发送到延时队列上并设置过期时间当过期时间到达时消息会被自动转发到指定的交换机和队列中供接收者消费。 1、建立与 RabbitMQ 服务器的连接并创建通道。 2、发送者通过 ch.Publish 方法将消息发送到延时队列“test_delay”上设置消息的过期时间。 3、延时队列中的消息在到达过期时间后会自动被发送到 “logs” 交换机由交换机将消息广播给所有绑定的队列。 4、接收者通过监听 “test_logs” 队列接收并处理消息。当有消息到达时会触发回调函数进行处理。 也就是说要实现延时队列消费者必须试实现两个队列。 一个是延时队列“test_delay”另一个是接收延时消息的队列“test_logs”。
这两个队列的作用如下 延时队列“test_delay”这个队列用于接收需要延时发送的消息。发送者通过将消息发送到延时队列设置消息的过期时间。当消息过期时RabbitMQ 会自动将消息转发到指定的交换机和队列中。 接收延时消息的队列“test_logs”这个队列用于接收延时消息。在示例中这个队列是通过将 “test_logs” 队列绑定到 “logs” 交换机上来实现的。交换机会将消息广播给所有绑定的队列因此当延时消息到达过期时间后会被发送到这个队列中供消费者进行处理。 通过使用两个队列消息可以被延时发送到指定的队列并在过期后自动转发到接收队列实现了延时发送和消费的功能。
三、Go语言实战
生产者
首先建立与 RabbitMQ 服务器的连接并创建一个通道。然后通过 ch.Publish 方法将消息发送到延时队列上。这里使用的是空字符串作为交换机exchange表示不选择任何交换机只将消息发送到指定的队列“test_delay”。在消息的属性中设置了消息的过期时间为 5 秒。 func main() {conn, err : amqp.Dial(amqp://guest:guestlocalhost:5672/)failOnError(err, Failed to connect to RabbitMQ)defer conn.Close()ch, err : conn.Channel()failOnError(err, Failed to open a channel)defer ch.Close()body : hello// 将消息发送到延时队列上err ch.Publish(, // exchange 这里为空则不选择 exchangetest_delay, // routing keyfalse, // mandatoryfalse, // immediateamqp.Publishing{ContentType: text/plain,Body: []byte(body),Expiration: 5000, // 设置五秒的过期时间})failOnError(err, Failed to publish a message)log.Printf( [x] Sent %s, body)
}
消费者
同样建立与 RabbitMQ 服务器的连接并创建一个通道。然后声明了一个名为 “logs” 的交换机类型为 “fanout”并且可持久化表示该交换机会将消息广播给所有绑定的队列。接着声明了一个常规的队列 “test_logs”并将其绑定到 “logs” 交换机上。之后声明了一个延时队列 “test_delay”并设置了该队列的 x-dead-letter-exchange 参数为 “logs”即当消息过期时将消息发送到 “logs” 交换机。最后通过 ch.Consume 方法监听 “test_logs” 队列并在回调函数中处理接收到的消息。 func main() {// 建立链接conn, err : amqp.Dial(amqp://guest:guestlocalhost:5672/)failOnError(err, Failed to connect to RabbitMQ)defer conn.Close()ch, err : conn.Channel()failOnError(err, Failed to open a channel)defer ch.Close()// 声明一个主要使用的 exchangeerr ch.ExchangeDeclare(logs, // namefanout, // typetrue, // durablefalse, // auto-deletedfalse, // internalfalse, // no-waitnil, // arguments)failOnError(err, Failed to declare an exchange)// 声明一个常规的队列, 其实这个也没必要声明,因为 exchange 会默认绑定一个队列q, err : ch.QueueDeclare(test_logs, // namefalse, // durablefalse, // delete when unusedtrue, // exclusivefalse, // no-waitnil, // arguments)failOnError(err, Failed to declare a queue)/*** 注意,这里是重点!!!!!* 声明一个延时队列, ß我们的延时消息就是要发送到这里*/_, errDelay : ch.QueueDeclare(test_delay, // namefalse, // durablefalse, // delete when unusedtrue, // exclusivefalse, // no-waitamqp.Table{// 当消息过期时把消息发送到 logs 这个 exchangex-dead-letter-exchange:logs,}, // arguments)failOnError(errDelay, Failed to declare a delay_queue)err ch.QueueBind(q.Name, // queue name, 这里指的是 test_logs, // routing keylogs, // exchangefalse,nil)failOnError(err, Failed to bind a queue)// 这里监听的是 test_logsmsgs, err : ch.Consume(q.Name, // queue name, 这里指的是 test_logs, // consumertrue, // auto-ackfalse, // exclusivefalse, // no-localfalse, // no-waitnil, // args)failOnError(err, Failed to register a consumer)forever : make(chan bool)go func() {for d : range msgs {log.Printf( [x] %s, d.Body)}}()log.Printf( [*] Waiting for logs. To exit press CTRLC)-forever
}
