5大常见高并发限流算法选型浅析

01

class FixedWindowRateLimiter {public:    FixedWindowRateLimiter(int max_requests_per_win, std::chrono::seconds window_size)        : max_requests_per_win_(max_requests_per_win), window_size_(window_size), request_count_(0) {        window_start_time_ = std::chrono::steady_clock::now();    }
    bool TryAcquire(int request_size) {        std::lock_guard<std::mutex> lock(mtx_);        auto now = std::chrono::steady_clock::now();
        // 如果当前时间在窗口内        if (now - window_start_time_ < window_size_) {            // 检查请求数量是否超过限制            if (request_count_ + request_size <= max_requests_per_win_) {                request_count_ += request_size; // 增加请求计数                return true; // 允许请求            } else {                return false; // 超过最大请求数            }        } else {            // 重置窗口            window_start_time_ = now;            request_count_ = request_size; // 重置请求计数为当前请求数量            return true; // 允许请求        }    }
private:    int max_requests_per_win_; // 最大请求数    std::chrono::seconds window_size_; // 窗口大小    int request_count_; // 当前请求计数    std::chrono::steady_clock::time_point window_start_time_; // 窗口开始时间    std::mutex mtx_; // 互斥锁};

02

class SlidingWindowRateLimiter {public:    SlidingWindowRateLimiter(int max_requests_per_win, std::chrono::seconds bucket_size, int num_buckets)        : max_requests_per_win_(max_requests_per_win), bucket_size_(bucket_size), num_buckets_(num_buckets) {        request_counts_.resize(num_buckets, 0);        last_bucket_time_ = std::chrono::steady_clock::now();    }
    bool TryAcquire(int request_size) {        std::lock_guard<std::mutex> lock(mtx_);        auto now = std::chrono::steady_clock::now();        UpdateBuckets_(now);
        int total_requests = 0;        for (int count : request_counts_) {            total_requests += count;        }
        if (total_requests + request_size <= max_requests_per_win_) {            request_counts_[current_bucket_index_] += request_size; // 增加当前桶的请求计数            return true; // 允许请求        } else {            return false; // 超过最大请求数        }    }
private:    void UpdateBuckets_(std::chrono::steady_clock::time_point now) {        auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(now - last_bucket_time_);        int buckets_to_update = static_cast<int>(elapsed / bucket_size_);
        if (buckets_to_update > 0) {            for (int i = 0; i < std::min(num_buckets_, buckets_to_update); ++i) {                auto next_bucket_index = (current_bucket_index_ + 1) % num_buckets_;                request_counts_[next_bucket_index] = 0; // 移除最旧的桶                current_bucket_index_ = next_bucket_index; // 移动到下一个桶            }            last_bucket_time_ += buckets_to_update * bucket_size_; // 更新最后桶的时间        }    }
    int max_requests_per_win_; // 最大请求数    std::chrono::seconds bucket_size_; // 每个小桶的大小    int num_buckets_; // 小桶的数量    std::vector<int> request_counts_; // 每个小桶的请求计数    std::mutex mtx_; // 互斥锁    std::chrono::steady_clock::time_point last_bucket_time_; // 上一个桶的时间    int current_bucket_index_ = 0; // 当前桶的索引};

03

class SlidingLogRateLimiter {public:    SlidingLogRateLimiter(int max_requests_per_win, std::chrono::seconds window_size)        : max_requests_per_win_(max_requests_per_win), window_size_(window_size) {}
    bool TryAcquire(int request_size) {        std::lock_guard<std::mutex> lock(mtx_);        auto now = std::chrono::steady_clock::now();        CleanUp_(now);
        int total_requests = 0;        for (const auto& timestamp : request_log_) {            total_requests += timestamp.second;        }
        if (total_requests + request_size <= max_requests_per_win_) {            request_log_.emplace_back(now, request_size); // 记录当前请求            return true; // 允许请求        } else {            return false; // 超过最大请求数        }    }
private:    void CleanUp_(std::chrono::steady_clock::time_point now) {        auto expiration_time = now - window_size_;        while (!request_log_.empty() && request_log_.front().first < expiration_time) {            request_log_.pop_front(); // 移除过期的请求记录        }    }
    int max_requests_per_win_; // 最大请求数    std::chrono::seconds window_size_; // 窗口大小    std::deque<std::pair<std::chrono::steady_clock::time_point, int>> request_log_; // 请求日志    std::mutex mtx_; // 互斥锁};

04

class Task {public:    virtual int GetLoad() = 0;    virtual void Run() = 0;}
class LeakyBucketRateLimiter {public:    LeakyBucketRateLimiter(int capacity, int leak_rate)        : capacity_(capacity), leak_rate_(leak_rate), current_water_(0) {        last_leak_time_ = std::chrono::steady_clock::now();    }
    bool TryRun(const TaskPtr& task) {        std::lock_guard<std::mutex> lock(mtx_);        Leak();
        if (current_water_ + task.GetLoad() <= capacity_) {            current_water_ += task.GetLoad(); // 将请求放入漏桶            heap_timer_.Schedule(task, current_water_ / leak_rate); // 定时器在该任务的水漏完后将会执行task的Run方法            return true; // 允许请求        } else {            return false; // 漏桶已满，请求被拒绝        }    }
private:    void Leak() {        auto now = std::chrono::steady_clock::now();        auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(now - last_leak_time_);        int leaked_water = static_cast<int>(elapsed.count()) * leak_rate_;
        if (leaked_water > 0) {            current_water_ = std::max(0, current_water_ - leaked_water); // 减少水量            last_leak_time_ = now; // 更新最后漏水时间        }    }
    int capacity_; // 漏桶的容量    int leak_rate_; // 漏水速率（每秒漏掉的水量）    int current_water_; // 当前水量    HeapTimer heap_timer_; // 一个任务执行的定时器    std::mutex mtx_; // 互斥锁    std::chrono::steady_clock::time_point last_leak_time_; // 上次漏水的时间};

05

class TokenBucketRateLimiter {public:    TokenBucketRateLimiter(int capacity, int refill_rate)        : capacity_(capacity), refill_rate_(refill_rate), current_tokens_(0) {        last_refill_time_ = std::chrono::steady_clock::now();    }
    bool TryAcquire(int request_size) {        std::lock_guard<std::mutex> lock(mtx_);        Refill_();
        if (current_tokens_ >= request_size) {            current_tokens_ -= request_size; // 消耗令牌            return true; // 允许请求        } else {            return false; // 令牌不足，请求被拒绝        }    }
private:    void Refill_() {        auto now = std::chrono::steady_clock::now();        auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(now - last_refill_time_);        current_tokens_ = std::min(capacity_, current_tokens_ + static_cast<int>(elapsed.count()) * refill_rate_); // 更新令牌数量        last_refill_time_ = now; // 更新最后补充时间    }
    int capacity_; // 令牌桶的容量    int refill_rate_; // 令牌补充速率（每秒补充的令牌数量）    int current_tokens_; // 当前令牌数量    std::mutex mtx_; // 互斥锁    std::chrono::steady_clock::time_point last_refill_time_; // 上次补充令牌的时间};

06