面试题概览：

• 限流的主要目的是什么，有哪些常见的限流算法？
• 请说说看计数器限流的基本原理和它存在的问题？
• 滑动窗口限流算法如何优化计数器限流？请简单用代码实现以下滑动窗口限流；
• 滑动窗口限流在面对突发流量的情况下依旧不够平滑，还有没有什么限流算法能够让系统在面对突发流量的时候更均匀的接收请求？
• 说说看漏桶算法和令牌桶算法的原理以及它们的差异？
• 如何实现不仅对总流量限流，也对单个用户的指定时间段限流（基于用户身份限流）？
  

  面试官：能不能说说看在系统架构中，限流的主要目的是什么，有哪些常见的限流算法

在系统架构中，限流的主要目的是保护系统免受过量请求的冲击，确保系统资源能够得到合理分配和有效利用，从而维持系统的稳定性和可靠性。具体来说，限流有助于防止以下几种情况的发生：

1. 系统过载：当请求量超过系统的处理能力时，系统可能会因为资源耗尽（如CPU、内存、数据库连接等）而崩溃或响应变慢。
2. 数据库压力：大量的并发请求可能导致数据库负载过高，进而影响数据的一致性和完整性。
3. 服务雪崩：在微服务架构中，一个服务的过载可能会引发连锁反应，导致整个系统的崩溃。

限流常见的实现方式包括以下几种：

1. 计数器限流：

• 通过记录单位时间窗口内的请求数量，当数量达到设定的阈值时，拒绝后续请求。
• 优点：实现简单，易于理解。
• 缺点：存在临界问题，即在时间窗口切换时的大量请求突发会影响下一个窗口对限流的判断。
  10. 滑动窗口限流：
• 改进了计数器限流，通过维护一个滑动的时间窗口来记录请求数量，从而更精细地控制请求速率。
• 优点：解决了临界问题，能够更平滑地控制请求流量。
• 缺点：实现相对复杂，且依旧无法很好应对突发流量给系统带来的高负荷。
  12. 令牌桶限流：
• 以恒定速率向令牌桶中添加令牌，请求到来时从桶中取出令牌，若桶中无令牌则拒绝请求。
• 优点：能够平滑突发流量，适应性强。
• 缺点：需要维护令牌桶的状态，开销稍大。
  14. 漏桶限流：
• 请求以固定速率流入漏桶，桶满则拒绝请求，桶不满则请求继续处理。
• 优点：实现简单，易于理解。
• 缺点：应对突发流量时能通过恒定速度放开请求避免系统高负荷，但是可能造成高并发请求延迟。
  16. 基于Redis的限流：
• 利用Redis的原子操作特性，如INCR、EXPIRE等，实现限流逻辑。
  

  面试官：能详细说说看计数器限流的基本原理吗？

计数器限流（也称为固定窗口限流算法）是一种简单且直观的限流方法，其基本原理是在一段时间间隔内对请求进行计数，并将计数结果与设置的最大请求数进行比较，以决定是否允许后续请求通过。以下是对计数器限流基本原理的详细解释：

一、时间间隔与计数器

1. 时间间隔：计数器限流通常设定一个固定的时间间隔，如1分钟、5分钟等。在这个时间间隔内，系统会对接收到的请求进行计数。
2. 计数器：系统维护一个计数器变量，用于记录在当前时间间隔内接收到的请求数量。每当有请求到达时，计数器就会加1。
   

   二、请求计数与比较

3. 请求计数：在设定的时间间隔内，每当有请求到达系统时，计数器就会增加相应的数值。这样，系统就可以实时地跟踪当前时间间隔内的请求数量。
4. 比较操作：系统会将计数器的当前值与预设的最大请求数进行比较。如果计数器的值小于或等于最大请求数，则允许请求通过；如果计数器的值大于最大请求数，则拒绝后续请求，直到时间间隔结束并重置计数器。
   

   三、时间间隔结束与计数器重置

5. 时间间隔结束：当设定的时间间隔结束时，系统会自动将计数器重置为0，并开始下一个时间间隔的计数。
6. 计数器重置：重置计数器是为了确保在新的时间间隔内，系统能够重新接收和处理请求。这样，系统就可以持续地对请求进行限流控制。

以下是一个使用Python实现的简单计数器限流示例。这个示例使用了一个全局变量来模拟计数器，以及一个时间戳来跟踪当前时间窗口的开始时间。

import time 
from threading import Lock 
class CounterRateLimiter: 
    def __init__(self, max_requests, window_size_in_seconds): 
        """ 
        初始化计数器限流器 
        :param max_requests: 时间窗口内允许的最大请求数 
        :param window_size_in_seconds: 时间窗口的大小（秒） 
        """ 
        self.max_requests = max_requests 
        self.window_size_in_seconds = window_size_in_seconds 
        self.request_count = 0 
        self.window_start_time = time.time() 
        self.lock = Lock() 
    def allow_request(self): 
        """ 
        检查是否允许请求通过 
        :return: 如果允许请求通过则返回True，否则返回False 
        """ 
        with self.lock: 
            current_time = time.time() 
# 检查是否到了新的时间窗口 
            if current_time - self.window_start_time >= self.window_size_in_seconds: 
# 重置计数器和时间窗口 
                self.request_count = 0 
                self.window_start_time = current_time 
# 检查请求数量是否超过阈值 
            if self.request_count < self.max_requests: 
                self.request_count += 1 
                return True 
            else: 
                return False 
# 使用示例 
if __name__ == "__main__": 
# 允许每分钟最多10个请求 
    rate_limiter = CounterRateLimiter(max_requests=10, window_size_in_seconds=60) 
# 模拟请求 
    for i in range(15): 
        if rate_limiter.allow_request(): 
            print(f"请求 {i + 1} 被允许") 
        else: 
            print(f"请求 {i + 1} 被拒绝") 
# 模拟请求间隔（例如，每秒一个请求） 
        time.sleep(1)

计数器限流是一种常用的限流策略，具有简单易用、直观易懂等优点，但也存在一些明显的问题，具体包括：

时间临界点突发流量问题：计数器限流存在“时间临界点”缺陷，即在时间区间的临界点附近，如果请求数突然增加，可能会导致短时间内大量请求通过限流检查，从而对系统造成压力。

假设我们要求系统在1分钟之内最多通过100个请求，系统以1分钟作为时间区间，因此 [0s, 59s] 是一个时间区间，[1min, 1min 59s]是一个时间区间。

假设有一个恶意用户，他在0:59时，瞬间发送了100个请求，并且1:00又瞬间发送了100个请求。我们希望的是1分钟内平均最多只能接受200个请求，但其实这个用户在 1秒里面，瞬间发送了200个请求。

要知道，1分钟均匀请求100次和1分钟内的短短几秒里突发请求100次所带给系统的压力是完全不同的。用户有可能通过算法的这个漏洞，瞬间压垮我们的应用。

这个问题，本质其实就在于这样固定周期清空计数器的方式精度太低，不能区分 1 分钟内均匀请求 100 次，以及只在 1 分钟内很小的一个时间区间里请求 100 次，这两种情况。

面试官：那么有没有什么更好的算法可以解决这个问题呢？

解决这个问题本质就是要解决时间区间的精度问题，而滑动窗口限流算法可以通过将一个大的时间区间进一步划分为多个小的区间来解决窗口精度问题。

其基本原理可以归纳如下：

一、定义

滑动窗口限流算法通过将大的时间区间划分为若干个固定大小的小区间（或称为时间片段），并且以这个大的时间区间作为一个时间窗口，随着时间的推移这个大的时间窗口可以在小的时间分片上移动，并动态地维护这个大窗口内的请求数量和小时间片内的请求数量，从而实现对流量的精确控制。

二、基本原理

1. 时间窗口划分：

• 将大的时间区间（如上面例子中的1分钟）划分为多个固定大小的小区间（例如10秒）。

45. 窗口滑动：
46. 在第一个一分钟里，窗口并不会滑动。而之后，每隔一个时间片（也就是每隔10秒），窗口就会往前移动一个时间片。以此类推，随着时间的推移，窗口会不断向前滑动。
47. 请求计数：
    在每个小的时间片段内，都维护一个计数器来记录该时间片内的请求数量。而大的时间窗口内的请求数量等于它里面所有小时间片段的请求数量之和。

当新的请求到达时，根据其到达时间确定其所属的小时间片段，并在该时间片的计数器上递增（也就是图中黄色的时间片），与此同时大窗口内的请求数量也同时增加。

而第一个时间片中的请求数量会被大窗口内丢弃（减法扣去），同时开始加上新的时间片内的请求计数。

49. 限流判断：
    • 每次请求到达的时候，都会检查大窗口内的总请求数量是否超过了预设的阈值。
    • 如果请求数量超过了阈值，则在窗口继续滑动到下一个时间片之前的请求都会被拒绝，等到10s过去了，窗口滑动到下一个时间片并丢弃掉窗口内第一个时间片的请求量之后，请求才会被允许通过并继续处理。

    • 三、优点

50. 灵活性：

• 滑动窗口限流算法可以根据业务需求动态地调整窗口内时间片的大小（可以调成10秒，也可以调成1秒）。
• 这使得算法能够更好地适应不同的流量场景和业务需求。
  54. 精确性：
• 如果将窗口内的时间片调节的足够小，滑动窗口限流算法就能够更有效地应对突发流量。
• 还是上面那个例子，依旧要求1分钟内最多处理100个请求，如果我将时间片调整为1秒，窗口就会每秒滑动一次。恶意用户依旧是在第59秒并发100个请求，再在第60秒并发另外100个请求。
• 那么当窗口滑动到第60秒的时候，第60秒的100个请求都会被拒绝。
  

  四、缺点

滑动窗口算法在将窗口划分的很细的情况下，可能会带来以下缺点：

1. 算法复杂度增加：

• 想要进一步提高被控制流量的平滑性，就需要不断增加窗口的精度，也就是缩小每个区间的大小。当窗口被划分得非常细时，需要维护的计数器数量会显著增加。每个子窗口都需要一个独立的计数器来跟踪请求数量，这会导致算法在计算和更新计数器时的复杂度增加。
• 窗口细化意味着需要处理更多的数据点和时间戳。这会增加算法的存储需求和计算开销。
• 60. 限流效果依旧不稳定：
• 滑动窗口限流算法的另一个缺点是限流仍然不够平滑。例如，如果在某个小窗口的开始阶段就达到了限流阈值，那么在这个小窗口剩余的时间内，所有新的请求都会被拒绝，这也会影响到用户体验。
  

  面试官：能不能具体实现一下滑动窗口限流算法呢？

Redis的有序集合（zset）非常适合实现这种算法，因为zset能够存储带时间戳的分数，并根据分数自动排序。

下面是一个使用Redis的zset实现滑动窗口限流算法的代码：

import redis 
import time 
class SlidingWindowRateLimiter: 
    def __init__(self, redis_client, key_prefix, max_requests, window_size_in_seconds): 
        self.redis = redis_client 
        self.key_prefix = key_prefix 
        self.max_requests = max_requests 
        self.window_size_in_seconds = window_size_in_seconds 
    def _get_key(self, user_id): 
        return f"{self.key_prefix}:{user_id}" 
    def _get_current_window_start_time(self): 
        return int(time.time()) // self.window_size_in_seconds * self.window_size_in_seconds 
    def _add_request(self, user_id): 
        key = self._get_key(user_id) 
        window_start_time = self._get_current_window_start_time() 
        score = time.time() 
# 使用管道来确保原子性 
        with self.redis.pipeline() as pipe: 
            pipe.zremrangebyscore(key, 0, window_start_time - self.window_size_in_seconds) 
            pipe.zadd(key, {score: score}) 
            pipe.zcard(key) 
            result = pipe.execute() 
# 移除旧窗口的数据，并添加新请求 
        _, _, current_count = result 
        return current_count 
    def allow_request(self, user_id): 
        current_count = self._add_request(user_id) 
        return current_count <= self.max_requests 
# 示例使用 
if __name__ == "__main__": 
# 连接到Redis服务器 
    redis_client = redis.StrictRedis(host='localhost', port=6379, db=0) 
# 创建限流器实例 
    rate_limiter = SlidingWindowRateLimiter( 
        redis_client, 
        key_prefix="rate_limiter", 
        max_requests=10,  # 每窗口最多10个请求 
        window_size_in_seconds=60  # 窗口大小为60秒 
    ) 
# 示例用户ID 
    user_id = "user_123" 
# 模拟请求 
    for i in range(15): 
        if rate_limiter.allow_request(user_id): 
            print(f"Request {i + 1} allowed.") 
        else: 
            print(f"Request {i + 1} denied.") 
        time.sleep(2)  # 每2秒发送一个请求

1. 初始化：

• __init__ 方法中初始化Redis客户端、key前缀、最大请求数和窗口大小。
  68. 获取Redis键：
• _get_key 方法根据用户ID生成Redis键。
  70. 获取当前窗口开始时间：
• _get_current_window_start_time 方法计算当前时间所属窗口的开始时间（向下取整到窗口大小的倍数）。
  72. 添加请求：
• _add_request 方法使用Redis管道（pipeline）来确保原子性操作。
• 先移除旧窗口的数据（分数小于当前窗口开始时间减去窗口大小的数据）。
• 添加新请求的时间戳到zset中。
• 获取当前窗口内的请求数量，判断请求是否允许。

在不使用中间件的情况下我们可以使用纯内存数据结构（队列）结合时间戳来实现滑动窗口限流算法。这种方法适用于单机环境，或者当你不希望/不能使用外部存储（如Redis）时。

以下是一个使用Python的 deque 实现滑动窗口限流的示例代码：

from collections import deque 
import time 
class SlidingWindowRateLimiter: 
    def __init__(self, max_requests, window_size_in_seconds): 
        self.max_requests = max_requests 
        self.window_size_in_seconds = window_size_in_seconds 
        self.window = deque()  # 存储 (timestamp, request_id) 元组 
    def _is_within_window(self, timestamp): 
        window_start = time.time() - self.window_size_in_seconds 
        return timestamp >= window_start 
    def allow_request(self): 
        current_timestamp = time.time() 
# 移除窗口外的请求 
        while self.window and not self._is_within_window(self.window[0][0]): 
            self.window.popleft() 
# 添加当前请求到窗口 
        self.window.append((current_timestamp, id(self)))  # 使用id(self)作为示例请求ID，实际中应替换为唯一请求标识 
# 检查窗口内的请求数量 
        if len(self.window) > self.max_requests: 
            return False 
        return True 
# 示例使用 
if __name__ == "__main__": 
# 创建限流器实例 
    rate_limiter = SlidingWindowRateLimiter(max_requests=5, window_size_in_seconds=10) 
# 模拟请求 
    for i in range(10): 
        if rate_limiter.allow_request(): 
            print(f"Request {i + 1} allowed.") 
        else: 
            print(f"Request {i + 1} denied.") 
        time.sleep(2)  # 每2秒发送一个请求

1. 初始化：

• __init__ 方法中初始化最大请求数、窗口大小和用于存储请求记录的 deque 。
  79. 检查时间戳是否在窗口内：
• _is_within_window 方法用于检查给定的时间戳是否在当前窗口内。
  81. 允许请求：
• allow_request 方法首先获取当前时间戳，然后移除窗口外（即过期）的请求记录。
• 接着，将当前请求（带有时间戳和请求ID）添加到窗口中。
• 最后，检查窗口内的请求数量是否超过最大请求数，并返回相应的结果。
  

  面试官：就像你刚刚说的，滑动窗口限流在面对突发流量的情况下依旧不够平滑，还有没有什么限流算法能够让系统在面对突发流量的时候更均匀的接收请求呢？

可以使用漏桶算法，当请求速率大于桶的流出速率时，漏桶只会按照桶的流出速率均匀的放出请求给系统处理，而超出速率的请求会被储蓄在桶中等待处理。

漏桶算法的原理可以简单描述为：数据（或请求）流入漏桶，桶内的数据以固定的速率流出，控制了系统能够处理请求的速度。如果桶满了（即桶的容量到达上限），后续的流量（或请求）会被丢弃，防止系统超载。

具体来说，漏桶算法维持一个具有固定容量的“桶”，该桶接受以任意速率流入的请求，并以固定的速率处理（即“漏出”）这些请求。当桶满时，任何新进入的请求都将被丢弃。这样，无论请求到达的速率如何变化，系统处理请求的速率都保持恒定，从而实现了流量整形和限流控制。

漏桶算法的重要参数主要包括以下2个：

1. 桶的容量：

• 代表了系统能够处理的请求的最大数量，也代表了漏桶（包缓存）可以容纳的数据包的最大字节数。这个尺寸受限于有效的系统内存。
• 91. 漏出速率：
• 这个速率代表了系统处理数据的速率。它决定了数据从漏桶中注入网络的速率，从而平滑了突发流量。

在实际应用中，漏桶算法的工作流程大致如下：

• 初始化：设置桶的容量和漏出速率。
• 接收请求：每当有数据（或请求）到达时，先检查桶的状态。
• 桶未满：如果桶未满，允许数据（或请求）进入桶中，然后等待漏出。
• 桶已满：如果桶已满，新到达的数据（或请求）会被丢弃或排队等待。
• 数据（或请求）处理：数据（或请求）在桶中等待直到之前的请求被处理完，然后以漏出速率再被处理。
  

  优缺点

优点：

1. 稳定性好：漏桶算法能够以固定的速率去控制流量，使得系统处理请求的速率保持恒定，从而有效防止了流量激增导致的服务不稳定。
2. 保护后端服务：通过限制请求的处理速率，漏桶算法能够保护后端服务免受大流量冲击，避免服务崩溃。

缺点：

1. 请求延迟处理：当系统在短时间内有突发的大流量时，由于漏桶的流出速率是固定的，因此无法及时处理这些突发流量，突发流量会被阻塞一段时间才得到处理，还可能导致部分请求被丢弃。

下面是一个使用Python实现的简单漏桶算法示例实现。

import time 
import threading 
from collections import deque 
class LeakyBucket: 
    def __init__(self, capacity, leak_rate): 
        self.capacity = capacity  # 桶的容量 
        self.leak_rate = leak_rate  # 漏出速率（每秒漏出的请求数） 
        self.bucket = deque()  # 使用双端队列来模拟桶，存储请求的时间戳 
        self.lock = threading.Lock()  # 线程锁，用于线程安全 
    def _leak(self): 
        """内部方法，用于模拟漏出请求""" 
        current_time = time.time() 
        while self.bucket and (current_time - self.bucket[0]) >= 1.0 / self.leak_rate: 
            self.bucket.popleft()  # 移除最旧的请求 
    def allow_request(self): 
        """判断是否可以允许一个请求进入桶中""" 
        with self.lock: 
            self._leak()  # 先尝试漏出一些请求 
            current_time = time.time() 
            if len(self.bucket) < self.capacity: 
                self.bucket.append(current_time)  # 允许请求进入桶中 
                return True 
            else: 
# 桶已满，拒绝请求 
                return False 
# 示例使用 
if __name__ == "__main__": 
    bucket = LeakyBucket(capacity=10, leak_rate=5)  # 创建一个漏桶，容量为10，漏出速率为5个请求/秒 
# 模拟请求到达 
    for i in range(20): 
        if bucket.allow_request(): 
            print(f"Request {i} allowed at {time.time():.2f}") 
        else: 
            print(f"Request {i} rejected at {time.time():.2f}") 
# 模拟请求到达的间隔 
        time.sleep(0.2)

在这个示例中，LeakyBucket 类实现了漏桶算法。

init 方法初始化了桶的容量、漏出速率以及用于存储请求时间戳的双端队列。

_leak 方法是一个内部方法，用于模拟漏出请求的过程，它会检查桶中最旧的请求是否已经超过了漏出速率所允许的时间，并移除这些请求。

allow_request 方法用于判断是否可以允许一个新的请求进入桶中，它首先尝试漏出一些请求，然后检查桶是否已满，如果未满则允许请求进入并返回 True，否则返回 False 表示请求被拒绝。

面试官：能否说说看与漏桶算法相似的令牌桶算法以及它们的差异呢？

令牌桶算法的基本算法是这样的：

如果我们需要在一秒内限制访问次数为 N 次，那么就每隔 1/N 的时间，往桶内放入一个令牌；

在处理请求之前先要从桶中获得一个令牌，如果桶中已经没有了令牌，那么就需要等待新的令牌或者直接拒绝服务；

桶中的令牌总数也要有一个限制，如果超过了限制就不能向桶中再增加新的令牌了。这样可以限制或增加令牌的总数，一定程度上可以避免瞬时流量高峰的问题。

令牌桶算法与漏桶算法的差异点是，漏桶算法的目的是“精确平滑控制速率”，它会完全避免系统受到突发流量的冲击，让系统一直以均匀的速度处理请求。

而令牌桶的桶本身具备一定的容量，可以允许一次把桶里的令牌全都取出，因此，令牌桶算法在限制请求的平均速率的同时，还允许一定程度的突发流量。

正常来说，系统并非是受不了一点冲击的温室花朵，也还是具有一定的处理突发流量的能力的，令牌桶算法也符合现实业务中90%的时间流量处于均值或者低谷，而偶尔存在并发高峰的场景。

下面是一个令牌桶算法的简单Python实现。

import time 
import threading 
class TokenBucket: 
    def __init__(self, rate, capacity): 
        """ 
        初始化令牌桶 
        :param rate: 令牌生成速率（每秒生成的令牌数） 
        :param capacity: 桶的容量（最大令牌数） 
        """ 
        self.rate = rate 
        self.capacity = capacity 
        self.tokens = capacity  # 初始令牌数 
        self.lock = threading.Lock() 
        self.last_refill_timestamp = time.time()  # 上次添加令牌的时间戳 
    def _add_tokens(self): 
        """ 
        根据时间差计算并添加令牌 
        """ 
        now = time.time() 
        elapsed = now - self.last_refill_timestamp 
        added_tokens = elapsed * self.rate 
        self.last_refill_timestamp = now 
# 添加令牌，但不能超过桶的容量 
        self.tokens = min(self.capacity, self.tokens + added_tokens) 
    def consume(self, num_tokens): 
        """ 
        尝试消耗指定数量的令牌 
        :param num_tokens: 需要消耗的令牌数 
        :return: 如果成功消耗则返回 True，否则返回 False 
        """ 
        with self.lock: 
            self._add_tokens() 
            if self.tokens >= num_tokens: 
                self.tokens -= num_tokens 
                return True 
            else: 
                return False 
# 示例使用 
if __name__ == "__main__": 
# 令牌生成速率 5 个/秒，桶容量 10 个 
    bucket = TokenBucket(rate=5, capacity=10) 
    for i in range(15): 
        if bucket.consume(1): 
            print(f"Successfully consumed 1 token. Current tokens: {bucket.tokens}") 
        else: 
            print("Failed to consume token. Not enough tokens in the bucket.") 
        time.sleep(0.5)  # 每 0.5 秒尝试消耗一次

此外还需要注意的是使用令牌桶算法就需要存储令牌的数量以及申请获取令牌，如果是单机上实现限流的话，可以在进程中使用一个变量来存储；但是如果在分布式环境下，不同的机器之间无法共享进程中的变量，我们就一般会使用 Redis 来存储这个令牌的数量。

这样的话，每次请求的时候都需要请求一次 Redis 来获取一个令牌，会增加几毫秒的延迟，性能上会有一些损耗。因此，一个折中的思路是：我们可以在每次取令牌的时候，不再只获取一个令牌，而是获取一批令牌，这样可以尽量减少请求 Redis 的次数。

面试官：如何实现不仅对总流量限流，也对单个用户的在指定长度的时间段限流？

举个例子，如果我需要对总流量限流为每分钟1万个请求，对单个用户限流为每分钟60个请求。这就意味着对于用户A来说，如果A请求时已经触发总流量1万个请求的限流，那么A无法访问；或者如果A请求时已经在1分钟内请求超过了60次，那么A也还是无法访问。

此时我们可以设置了两个令牌桶：一个是全局的令牌桶，一个是以 userid 为 key，按照用户来划分的令牌桶。

这里用两个令牌桶做了一个组合，就可以即做到对总流量限流，也对用户身份限流：

在全局令牌桶还有令牌的情况下，判断全局的令牌桶，如果全局令牌桶没有令牌了，那么触发限流拒绝访问。如果全局令牌桶还有令牌，就再去检查用户令牌桶；

在用户令牌桶有令牌的情况下，用户A的请求可以通过。用户令牌桶没有令牌的情况下，用户A的请求会被拒绝掉。

这样一来，就可以保证其他用户的请求不会受到影响。

---