面试题：Redis限流熔断状态监控与预警机制之基础实现

Question

请阐述如何利用Redis的原子操作实现简单的限流功能，并描述在监控限流状态时，Redis的哪些数据结构比较适合用于记录相关信息，为什么？

Answer 1

利用Redis原子操作实现简单限流功能

1. 计数器方式：
   • 使用INCR命令，它是Redis的原子操作。例如，假设要限制某个API每分钟最多被调用100次。可以在Redis中使用一个键（比如api:limit:123，其中123是API的标识）来记录调用次数。
   • 每次API被调用时，执行INCR api:limit:123命令。这个命令会原子性地将键的值加1，并返回加1后的值。
   • 当返回值小于等于100时，允许请求通过；当返回值大于100时，限制请求。同时，需要配合EXPIRE命令设置键的过期时间为1分钟，这样每分钟都会重新计数。示例代码（以Python为例）：

   import redis
   
   r = redis.Redis(host='localhost', port=6379, db = 0)
   key = 'api:limit:123'
   max_calls = 100
   call_count = r.incr(key)
   if call_count == 1:
       r.expire(key, 60)
   if call_count > max_calls:
       print('请求被限流')
   else:
       print('请求通过')
   

2. 滑动窗口方式（更精确限流）：
   • 利用ZADD和ZCARD命令。例如，还是限制每分钟100次调用。以时间戳作为分值，每次调用时，使用ZADD api:limit:123 {当前时间戳} {当前时间戳}（第二个参数是成员，这里使用时间戳作为成员也可以使用其他唯一标识）将调用记录添加到有序集合中。
   • 通过ZCOUNT api:limit:123 {当前时间戳 - 60} {当前时间戳}获取过去一分钟内的调用次数，若小于等于100则允许请求，否则限制。同时，还需要定期清理有序集合中过期的记录（比如通过ZREMRANGEBYSCORE命令删除时间戳小于当前时间戳 - 60的记录）。示例代码（以Python为例）：

   import redis
   import time
   
   r = redis.Redis(host='localhost', port=6379, db = 0)
   key = 'api:limit:123'
   max_calls = 100
   current_time = time.time()
   r.zadd(key, {current_time: current_time})
   call_count = r.zcount(key, current_time - 60, current_time)
   if call_count > max_calls:
       print('请求被限流')
   else:
       print('请求通过')
   r.zremrangebyscore(key, 0, current_time - 60)
   

监控限流状态适合的Redis数据结构及原因

1. 计数器方式：
   • 适合的数据结构：字符串（String）。
   • 原因：在计数器方式中，只需要记录一个简单的计数值，字符串类型足够存储这个计数值，并且INCR等原子操作对字符串类型支持良好，操作简单高效，占用内存空间小。
2. 滑动窗口方式：
   • 适合的数据结构：有序集合（Sorted Set）。
   • 原因：滑动窗口需要记录每个调用的时间戳信息，并且要按照时间顺序进行排序，以便计算在特定时间范围内的调用次数。有序集合正好满足这一需求，它可以根据分值（这里是时间戳）对成员进行排序，通过ZCOUNT等命令能够方便地统计在某个分值范围内的成员数量，即调用次数，非常适合实现滑动窗口限流的监控。