内容介绍
- 前言
- 简单的实现
- 定义消息
- Push
- Consume
- 存在的问题
- 多消费者实现
- 定义消息
- Push
- Consume
- 存在的问题
- 总结
前言
延迟队列是一种非常使用的数据结构，我们经常有需要延迟推送处理消息的场景，比如延迟60秒发送短信，延迟30分钟关闭订单，消息消费失败延迟重试等等。
zset哈希表+跳表RocketMQ我们这里先定一下我们要实现的几个目标：
- 消息必须至少被消费一次
- 多个生产者
- 多个消费者
然后我们定义一个简单的接口：
- Push(msg) error：添加消息到队列
- Consume(topic, batchSize, func(msg) error)：消费消息
简单的实现
- 每个主题最多可以被一个消费者消费，因为不会对主题进行分区
- 但是可以多个生产者同时进行生产，因为Push操作是原子的
- 同时需要消费操作返回值error为nil才删除消息，保证消息至少被消费一次
定义消息
这个消息参考了Kafka的消息结构：
- Topic可以是某个队列的名字
- Key是消息的唯一标识，在一个队列里面不可以重复
- Body是消息的内容
- Delay是消息的延迟时间
- ReadyTime是消息准备好执行的时间
// Msg 消息
type Msg struct {
Topic string // 消息的主题
Key string // 消息的Key
Body []byte // 消息的Body
Delay time.Duration // 延迟时间（秒）
ReadyTime time.Time // 消息准备好执行的时间（now + delay）
}
Push
HashZSet原子的同时我们不会覆盖已经存在的相同Key的消息。
const delayQueuePushRedisScript = `
-- KEYS[1]: topicZSet
-- KEYS[2]: topicHash
-- ARGV[1]: 消息的Key
-- ARGV[2]: 消息的Body
-- ARGV[3]: 消息准备好执行的时间
local topicZSet = KEYS[1]
local topicHash = KEYS[2]
local key = ARGV[1]
local body = ARGV[2]
local readyTime = tonumber(ARGV[3])
-- 添加readyTime到zset
local count = redis.call("zadd", topicZSet, readyTime, key)
-- 消息已经存在
if count == 0 then
return 0
end
-- 添加body到hash
redis.call("hsetnx", topicHash, key, body)
return 1
`
func (q *SimpleRedisDelayQueue) Push(ctx context.Context, msg *Msg) error {
// 如果设置了ReadyTime，就使用RedisTime
var readyTime int64
if !msg.ReadyTime.IsZero() {
readyTime = msg.ReadyTime.Unix()
} else {
// 否则使用Delay
readyTime = time.Now().Add(msg.Delay).Unix()
}
success, err := q.pushScript.Run(ctx, q.client, []string{q.topicZSet(msg.Topic), q.topicHash(msg.Topic)},
msg.Key, msg.Body, readyTime).Bool()
if err != nil {
return err
}
if !success {
return ErrDuplicateMessage
}
return nil
}
Consume
batchSizefnerrornilconst delayQueueDelRedisScript = `
-- KEYS[1]: topicZSet
-- KEYS[2]: topicHash
-- ARGV[1]: 消息的Key
local topicZSet = KEYS[1]
local topicHash = KEYS[2]
local key = ARGV[1]
-- 删除zset和hash关于这条消息的内容
redis.call("zrem", topicZSet, key)
redis.call("hdel", topicHash, key)
return 1
`
func (q *SimpleRedisDelayQueue) Consume(topic string, batchSize int, fn func(msg *Msg) error) {
for {
// 批量获取已经准备好执行的消息
now := time.Now().Unix()
zs, err := q.client.ZRangeByScoreWithScores(context.Background(), q.topicZSet(topic), &redis.ZRangeBy{
Min: "-inf",
Max: strconv.Itoa(int(now)),
Count: int64(batchSize),
}).Result()
// 如果获取出错或者获取不到消息，则休眠一秒
if err != nil || len(zs) == 0 {
time.Sleep(time.Second)
continue
}
// 遍历每个消息
for _, z := range zs {
key := z.Member.(string)
// 获取消息的body
body, err := q.client.HGet(context.Background(), q.topicHash(topic), key).Bytes()
if err != nil {
continue
}
// 处理消息
err = fn(&Msg{
Topic: topic,
Key: key,
Body: body,
ReadyTime: time.Unix(int64(z.Score), 0),
})
if err != nil {
continue
}
// 如果消息处理成功，删除消息
q.delScript.Run(context.Background(), q.client, []string{q.topicZSet(topic), q.topicHash(topic)}, key)
}
}
}
存在的问题
Consume多消费者实现
- 每个主题最多可以被分区个数个消费者消费，会对主题进行分区
定义消息
- 我们添加了一个Partition字段表示消息的分区号
// Msg 消息
type Msg struct {
Topic string // 消息的主题
Key string // 消息的Key
Body []byte // 消息的Body
Partition int // 分区号
Delay time.Duration // 延迟时间（秒）
ReadyTime time.Time // 消息准备好执行的时间
}
Push
代码与SimpleRedisDelayQueue的Push相同，只是我们会使用Msg里面的Partition字段对主题进行分区。
func (q *PartitionRedisDelayQueue) topicZSet(topic string, partition int) string {
return fmt.Sprintf("%s:%d:z", topic, partition)
}
func (q *PartitionRedisDelayQueue) topicHash(topic string, partition int) string {
return fmt.Sprintf("%s:%d:h", topic, partition)
}
Consume
partitionfunc (q *PartitionRedisDelayQueue) Consume(topic string, batchSize, partition int, fn func(msg *Msg) error) {
// ...
}
存在的问题
RocketMQKafka总结
RocketMQ