大家有没有遇到过这样的场景，一个订单超过一个时间段未支付自动取消，退款申请一段时间未人工处理自动通过或者拒绝，这些场景都有一个特点，在一个事件发生之后，需要在固定的事件段以后处理另一个事件，如果使用定时任务，每秒扫描一次数据库，取出需要处理的事件数据。这样固然能实现这个需求，但是在数据量比较大的情况会对数据库造成不小的压力，而且这样显然不够优雅，这时候就需要一个延时队列。
首先大家需要了解一下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
}有疑问加站长微信联系（非本文作者）
