Redis应用之实现延迟队列以及限流

Posted 2022-12-10 刘小豆豆豆

1、延迟队列实现

延迟队列的应用场景：

• 订单超过 30 分钟未支付，则自动取消。
• 外卖商家超时未接单，则自动取消。
• 医生抢单电话点诊，超过 30 分钟未打电话，则自动退款。
  等等场景都可以用定时任务去轮询实现，但是当数据量过大的时候，高频轮询数据库会消耗大量的资源，此时用延迟队列来应对这类场景比较好。

延迟队列可以通过Redis的ZSet数据结构实现，可以将消息序列化成为一个字符串作为zset的value，这个消息的到期处理时间作为score，然后用多线程轮询zset的到期的任务进行处理。多个线程轮询保证可用性，万一 挂掉一个线程还有其他线程可以继续处理。但因为有多个线程，需要考虑并发争抢任务。

具体做法：

利用 Redis 的 zset 结构，使用时间戳作为 score，比如你的任务是要延迟5分钟，那么就在当前时间上加5分钟作为 score ，轮询任务每秒只轮询 score 大于当前时间的 key即可，如果任务支持有误差，那么当没有扫描到有效数据的时候可以休眠对应时间再继续轮询。

通过主流的消息中间件也可以实现此需求。

2、限流

限流的作用除了控制流量外，限流还有一个应用目的是控制用户行为，避免垃圾请求，比如QQ空间的点赞，一般要现在某行为在规定时间内被允许的次数，超过次数就是非法行为。

限流除了常用的漏同法限流跟令牌桶限流外，还可以通过redis实现限流。

使用Redis限流

限定用户的某行为在指定时间内只能允许发生N次。

1、使用key键 限流

可以每秒生成一个key，每次请求来临时将value++，value值超过一定大小时，拒绝其他请求访问。

2、使用zset 限流

可以通过实现一个滑动时间窗口来实现限流，zset数据结构中的score值，可以通过score来圈出这个时间窗口来，我们只需要保留这个时间窗口，窗口外的数据拒绝访问。zset的value只需要保证唯一性即可，uuid比较浪费空间，采用毫秒的时间戳即可。

将用于行为（请求）作为key保存，通过统计滑动窗口内的行为数量与阈值max_count进行比较就可以得出当前的行为是否被允许。

整体思路：每一个行为到来时，都维护一次时间窗口，将滑动窗口以外的数据清理掉，只保留窗口内的记录。zset集合中，只有score值非常重要，value值没有特殊意义，保证其唯一即可。

高级限流算法

漏斗限流

漏斗的容量是有限的，如果将漏嘴堵住，然后一直往里灌水，它就会变满，直至再也装不进去。如果将漏嘴放开，水会以一定速率向下流，流走一部分后又可以进行灌水。如果漏嘴流速大于灌水速度，那漏斗永远装不满。如果灌水的速度大于漏嘴流速，灌水需要暂停，等待腾出部分空间。

漏斗的剩余空间可代表当前行为可以持续进行的数量，漏嘴的流水速度代表着系统允许该行为的最大频率。

思路：设置四个变量，漏斗的容量，漏嘴的流水速度，漏斗的剩余空间，上次漏水的时间。

(当前时间 - 上次漏水时间) x 漏嘴的流速 = 剩余空间。

此为单机版的限流算法，高并发的时候数据会不一致，原因是无法保证其原子性

分布式限流算法

Redis 4.0提供了限流Redis模块，即Redis-Cell，该模块也采用了限流算法，并提供原子的限流指令，有了这个模块，限流就比较简单了。

该模块只提供了一条指令 cl.throttle

127.0.0.1:6379> CL.THROTTLE test 15 30 60 1
1) (integer) 0    # 0表示允许， 1表示拒绝。
2) (integer) 15   # 漏斗的容量
3) (integer) 14   # 漏斗的剩余空间	
4) (integer) -1   # 如果被拒绝，需要多长时间后再试（漏斗已满，需要漏嘴流一会再打开）
5) (integer) 2    # 多长时间后，漏斗完全空出来

参数说明：

15表示漏斗的容量，30/60表示流速，即60秒允许30次请求。1代表可选参数，默认是1。

在执行限流指令时，如果被拒绝了，就需要丢弃或者重试。cl.throttle 指令考虑得非常周到，连重试时间都算好了，直接取返回结果数组的第四个值进行sleep即可，如果不想阻塞线程，也可以异步定时任务来重试。