基于Rabbitmq实现的延时队列(golang版)

虽然 rabbitmq 没有延时队列的功能,但是稍微变动一下也是可以实现的

实现延时队列的基本要素

  1. 存在一个倒计时机制:Time To Live(TTL)
  2. 当到达时间点的时候会触发一个发送消息的事件:Dead Letter Exchanges(DLX)

$~~$基于第一点,我利用的是消息存在过期时间这一特性, 消息一旦过期就会变成dead letter,可以让单独的消息过期,也可以设置整个队列消息的过期时间
rabbitmq会有限取两个值的最小值

$~~$基于第二点,是用到了rabbitmq的过期消息处理机制:
. x-dead-letter-exchange 将过期的消息发送到指定的 exchange
. x-dead-letter-routing-key 将过期的消息发送到自定的 route当中

在这里例子当中,我使用的是 过期消息+转发指定exchange

在 golang 中的实现

首先是消费者comsumer.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
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
}

然后是生产者productor.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
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
}

运行一下:

1
2
go run comsumer.go
go run productor.go

$~~$具体看代码和注释就行, 这里的关键点就是将要延时的消息发送到过期队列当中, 然后监听的是过期队列转发到的 exchange 下的队列
正常情况就是始终监听一个队列,然后把过期消息发送到延时队列中,当消息到达时间后就把消息发到正在监听的队列

一个自己写的mq工具