关于用 Redis 限制用户行为频率的方法，麻烦 V 友帮看看

[yangbin9317]

需求：

这是昨天面试中遇到的一个问题，当用户 x 秒内请求 y 次，禁止用户访问

先写一些我自己想的方案，下面的windowStart为 x 秒之前的 timestamp，current为现在的 timestamp，threshold为请求次数限制。下列方法都需要设置和更新 key 过期时间，以免造成 key 泄漏。

方案 1:

用 Redis 的 List 数据结构记录用户访问的时间

当用户访问时 rpush 当前时间戳并删掉左侧 x 秒之前的时间戳，此时该列表的长度为 x 秒请求次数，注意此方法的timestamp需要用 Redis 的 Lua 脚本否则可能因为各机器时间不同出问题（例如某台机器时间为未来，则 ltrim 不到该条数据后面的正常时间戳）

伪代码：

redis.rpush(key, current);
redis.expire(key, current + y);
counter = 0;
while (redis.get(key, 0) < windowStart) {
	counter++;
}
redis.ltrim(0, counter);
if (redis.llen(key) > y) {
	doBanUser();
}

方案 1 的问题：

只需要最多存 y 条数据，但是这里用到了 List，Redis 中的 List 为双向链表，每个节点包含两个指针，在 64 位机器上占用 128bit，而实际有用的时间戳只要 64bit 。迭代链表时会更慢一些。

方案 2:

使用数组 ringbuffer 代替方案一的链表。

方案 2 的问题：

需要自己开发 Redis module，开发和运维成本较高。另外方案 1 和方案 2 中均存不方便重试和测试的问题。例如其他业务调自己服务但是自己服务超时，此时业务方重试，这种情况不应该算用户访问了两次。

方案 3:

使用 Redis 的 ZSet 数据结构

以用户请求的唯一 requestId 为 member，当前时间为 score，如果 requestId 已存在（ logn 复杂度)，不进行任何操作，如果不存在，则 ZADD，然后使用 ZRANGEBYSCORE 查看该时间段内请求数量。该方案同时需要使用 ZREMRANGEBYSCORE 来删除过老的元素，我认为可以每次或 N 次请求的时候删除，面试官说有更好的删除时机，但是我也没有问，这个问题就这么过去了。

想问问 V 友该方法应该如何删除，或者有没有更好的方法来用 Redis 限制用户行为频率？

[RihcardLu]

一种不太严密的算法，记录第一次请求次数和时间，每次访问后次数+1, 当达到 y 次时，和第一次请求时间做比较，小于 X，置 0

[yangbin9317]

@RihcardLu #1 嗯，很简单但不能做到任意窗口内小于 y

[Livid]

目前 V2EX 的做法的大概思路：

now = int(time.time())
x = 60
y = 10

redis_key = str(now - (now % x))

count = client.incr(redis_key)

if count >= y:

[Livid]

这个也基本上也就是 Redis 本身推荐的实现方式：

https://redis.io/commands/incr

[jugelizi]

令牌桶算法

[yangbin9317]

@Livid #4 谢谢 Livid，但是文档中的方法不能避免例如五十九秒的时候请求 9 次，下一分钟一秒的时候再请求 9 次这种情况。

[Livid]

@yangbin9317 这种情况的问题是？单位时间内的总数（也可以理解为是服务器能够接受的请求次数）是限制住了的。

[Mohanson]

令牌桶算法.

http://accu.cc/content/go/token_bucket/

[123444a]

y 不一样做法不同，如果单用户 QPS 等于 10 万，2 万用户，那么你要几万台 codis 集群

[123444a]

高 QPS 分布式一般使用预分配 token

[123444a]

脱离需求谈设计等于耍流氓

[Lax]

@yangbin9317
大部分生产环境用这种算法足够了。
在对齐的时间片上满足频率要求，不对齐的时间片上最多是频率限制的 2 倍，业务系统应该能满足这种波动。况且这是单个客户端的频率超出，对总体影响会比较小。

[yangbin9317]

@Lax 谢谢，但这是一个面试题，应该不能改需求，所以想搞清楚我的方法哪里还能优化

[yangbin9317]

@123444a 我错了，我一直在小公司工作没见过大场面

[yangbin9317]

@Livid 谢谢 Livid，这种方法在现实世界很好用，事实上我在日常开发中也是这么做的。

[gosansam]

推荐滑动窗口

[xuanbg]

咦，这不是简单生成一个 key 保存一下访问次数作为 value，并且设置一下 key 的过期时间就行了吗？如果没有 key 就写入一个 key，有 key 存在并且 value 超过次数就拦截掉，没有超过次数就对 value 进行累加。难道这样有问题？

[watzds]

@yangbin9317 其实面试官没你想的这么多

[skadi]

incr + ttl ?

[huntcool001]

看一下我这个设计怎么样:

假设 60 秒内允许访问 10 次

那么 redis 里插入一个 USER_ID_LIST, 里面存 10 个当前的时间戳

每次用户进来,调用 RPOPLPUSH 这个命令,会在 list 右边插入当前时间,左边返回一个时间.

当左边返回的返回小于当前时间戳减去 60 秒的时候,说明已经触发机制. 这时候禁止用户操作.


RPOPLPUSH 这个命令是 O(1)的

正常情况不用担心机器同步时间的问题,大部分情况也不会超过 1 秒吧. 除非是对精度要求很高,可以用 Lua 脚本返回服务端的时间.