Redis延迟队列

2018-08-27 本文已影响116人吳下阿蒙

不同的微服务之间做异步通迅时通常会使用Kafka，它非常适用于对消费次序或时间没有强一致性需要的场景。如果消息需要在指定的时间才可以被消费，Kafka并没有原生支持此类消费场景，需要较复杂的实现。我们把支持此类场景的队列称为延迟队列。

逻辑上，延迟队列中的消息按延迟时间排序

延迟队列的主要特性是进入队列的消息会被推迟到指定的时间才出队被消费。而对于Kafka队列，消息入队后会排在队尾等待被消费，并不能指定出队时间。因此，延迟队列中的一条消息，除了消息本身外，还需要附加一个“何时出队”的信息。

例如，如果我把这篇文章设置为2018年3月14日发布，Medium可以把这如下的消息加入到延迟队列中；等到2018年3月14日这条消息才会出队，触发发布这篇文章的流程。

{
    "time": "2018-03-14T00:00:00Z",
    "message": "发布《Redis延迟队列》"
}

既然Kafka不支持延迟队列，我们有什么选择呢？

Java Collections Framework中包含的DelayQueue便是延迟队列的实现。然而这毕竟只是一个数据结构，基于它建设一个队列服务的工作量不少。
RabbitMQ通过RabbitMQ Delayed Message Plugin可支持延迟队列。可惜我司的基础设施中没有RabbitMQ。
Redis的Sorted Set可被用于实现简单的延迟队列。利用Redis的Lua支持我们也可以将基建设成一个功能全面的延迟队列服务。

下文将介绍如何使用Redis的Sorted Set实现简单的延迟队列。

Sorted Set是一个有序的集合，集合内元素的排序基于其加入集合时指定的score。通过ZRANGEBYSCORE命令，我们可以取得score在指定区间内的元素。将集合中的元素做为消息，score视为延迟的时间，这便是一个延迟队列的模型。

生产者通过ZADD将消息发送到队列中：

127.0.0.1:6379> ZADD delay-queue 1520985600 "publish article"

消费者通过ZRANGEBYSCORE获取消息。如果时间未到，将得不到消息；当时间已到或已超时，都可以得到消息：

127.0.0.1:6379> ZRANGEBYSCORE delay-queue -inf 1520985599
(empty list or set)
127.0.0.1:6379> ZRANGEBYSCORE delay-queue -inf 1520985600 WITHSCORES
1) "publish article"
2) "1520985600"
127.0.0.1:6379> ZRANGEBYSCORE delay-queue -inf 1520985601 WITHSCORES
1) "publish article"
2) "1520985600"

使用ZRANGEBYSCORE取得消息后，消息并没有从集合中删出。需要调用ZREM删除消息：

127.0.0.1:6379> ZREM delay-queue "publish article"

美中不足的是，消费者组合使用ZRANGEBYSCORE和ZREM的过程不是原子的，当有多个消费者时会存在竞争，可能使得一条消息被消费多次。此时需要使用Lua脚本保证消费操作的原子性：

local message = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', ARGV[1], 'WITHSCORES', 'LIMIT', 0, 1);
if #message > 0 then
  redis.call('ZREM', KEYS[1], message[1]);
  return message;
else
  return {};
end

Redis延迟队列

猜你喜欢

热点阅读