rabbitmq实现延迟队列

Posted 2021-09-27 润雨冰雪

大家有没有遇到过这样的场景，一个订单超过一个时间段未支付自动取消，退款申请一段时间未人工处理自动通过或者拒绝，这些场景都有一个特点，在一个事件发生之后，需要在固定的事件段以后处理另一个事件，如果使用定时任务，每秒扫描一次数据库，取出需要处理的事件数据。这样固然能实现这个需求，但是在数据量比较大的情况会对数据库造成不小的压力，而且这样显然不够优雅，这时候就需要一个延时队列。
首先大家需要了解一下rabbitmq的死信队列，如果消息具备下面三个条件之一

1.消息被否定确认
2.消息的存活时间超过设置的最大TTL时间
3.消息队列的长度已经超过最大长度

这时候这些消息就被称做死信消息

如果配置了死信队列，这些消息会被投递到死信队列，如果没有，这些消息会被丢弃

基于上面的规则，我们就可以利用死信队列实现延迟消息

定义两个队列A和B，和交换机（exchange）X，定义A的“x-dead-letter-exchange”为X。然后建立B与X的bind关系（路由规则需要自己把握），我们发送的消息到达A，然后发送的时候定义消息的存活时间（这个时间就是期望多久收到消息的时间），之后我们并不从A消费消息直到消息过期。然后消息就会被投递到交换机X，自然就能进入队列B。所以我们从B消费就能实现延迟消息的效果啦。

golang代码如下
//消费端

package main
import (
   "log"
 "github.com/streadway/amqp")
func failOnError(err error, msg string) {
   if err != nil {
      log.Fatalf("%s: %s", msg, err)
   }
}
func main() {
   // 建立链接
 conn, err := amqp.Dial("amqp://guest:guest@localhost: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()
   // 声明一个主要使用的 exchange
 err = ch.ExchangeDeclare(
      "logs", // name
 "fanout", // type
 true, // durable
 false, // auto-deleted
 false, // internal
 false, // no-wait
 nil, // arguments
 )
   failOnError(err, "Failed to declare an exchange")
   // 声明一个常规的队列, 其实这个也没必要声明,因为 exchange 会默认绑定一个队列
 q, err := ch.QueueDeclare(
      "test_logs", // name
 false, // durable
 false, // delete when unused
 true, // exclusive
 false, // no-wait
 nil, // arguments
 )
   failOnError(err, "Failed to declare a queue")
   /**
 * 注意,这里是重点!!!!! * 声明一个延时队列, ß我们的延时消息就是要发送到这里 */ _, errDelay := ch.QueueDeclare(
      "test_delay", // name
 false, // durable
 false, // delete when unused
 true, // exclusive
 false, // no-wait
 amqp.Table{
         // 当消息过期时把消息发送到 logs 这个 exchange
 "x-dead-letter-exchange":"logs",
 }, // arguments
 )
   failOnError(errDelay, "Failed to declare a delay_queue")
   err = ch.QueueBind(
      q.Name, // queue name, 这里指的是 test_logs
 "", // routing key
 "logs", // exchange
 false,
 nil)
   failOnError(err, "Failed to bind a queue")
   // 这里监听的是 test_logs
 msgs, err := ch.Consume(
      q.Name, // queue name, 这里指的是 test_logs
 "", // consumer
 true, // auto-ack
 false, // exclusive
 false, // no-local
 false, // no-wait
 nil, // 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 CTRL+C")
   <-forever
}

//发送端

package main
import (
   "log"
 "os" "strings"
 "github.com/streadway/amqp")
func failOnError(err error, msg string) {
   if err != nil {
      log.Fatalf("%s: %s", msg, err)
   }
}
func main() {
   conn, err := amqp.Dial("amqp://guest:guest@localhost: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 := bodyFrom(os.Args)
   // 将消息发送到延时队列上
 err = ch.Publish(
      "", // exchange 这里为空则不选择 exchange
 "test_delay", // routing key
 false, // mandatory
 false, // immediate
 amqp.Publishing{
         ContentType: "text/plain",
 Body:        []byte(body),
 Expiration: "5000", // 设置五秒的过期时间
 })
   failOnError(err, "Failed to publish a message")
   log.Printf(" [x] Sent %s", body)
}
func bodyFrom(args []string) string {
   var s string
 if (len(args) < 2) || os.Args[1] == "" {
      s = "hello"
 } else {
      s = strings.Join(args[1:], " ")
   }
   return s
}