限流器¶
在交易系统/风控/撮合/网关等高并发链路里，限流的目标是：把“不可控的突发流量”变成“系统可承受的稳定流量”，从而：
- 保护核心依赖（撮合、账户、风控、行情、DB/Cache、下游三方）
- 降低级联故障风险（雪崩、排队放大、超时重试风暴）
- 让延迟更可预测（尾延迟降低）
限流是为了在系统接近饱和时保持稳定和可恢复性。
常见使用场景（交易系统口径）
- 按账户/用户：
accountId维度限制下单/撤单频率，防止单账户刷爆系统 - 按 API Key/渠道：外部接入方 QPS 配额、突发能力（burst）
- 按接口/路由：如
/orders、/cancel、/query不同限额 - 按连接/会话：单连接突发保护（尤其是长连接网关）
- 按策略/机器人：每个策略独立限流，避免“策略异常”拖垮全局
- 按下游依赖：对某个依赖（如风控服务）做反压保护（可叠加熔断/降级）
限流算法选型
A) ApiRateLimiter（分片锁 + Token Bucket，通用生产版）
适用：key 数量大、需要自动过期清理、需要硬上限防 key 洪泛、并发中等~高。
原理（Token Bucket）：
- 桶里有“令牌”，每秒按
rate_per_sec生成，最多存burst（允许突发） - 请求需要消耗
tokens个令牌：有则通过，无则拒绝/返回 retry-after
实现要点：
- 为了高并发，使用 shards（分片）把
unordered_map拆开，降低锁竞争 unordered_map支持string_view异构查找：命中路径不再分配字符串TokenBucket用整数定点（token * 1e9 units），避免浮点带来的不确定性与额外开销idle_timeout/prune_interval/max_entries控制 key 生命周期与内存上限
B) LockFreeApiRateLimiter（无锁 + GCRA + 固定容量表）
适用：极致吞吐/低尾延迟、key 空间可控且可预估、愿意用固定容量换性能。
原理（GCRA / Virtual Scheduling）：
- 每个 key 维护一个“理论到达时间”
tat（Theoretical Arrival Time） - 对于每次请求，计算该请求在“理想均匀速率”下的下一次
tat - 判定条件：若
now >= tat - tolerance则允许，并把tat原子更新为max(tat, now) + increment increment约等于tokens / rate_per_sec对应的时间tolerance对应 burst 能力（允许短时超前）
实现要点与取舍：
- 热点路径主要是原子 CAS + 少量整数运算
- key 存储在固定容量的 lock-free open addressing 表里
- 容量固定：表满/探测超过
max_probe会拒绝插入新 key - key 用 64-bit hash 表示：冲突概率极低但非 0
#include <algorithm>
#include <atomic>
#include <cassert>
#include <chrono>
#include <cmath>
#include <cstdint>
#include <functional>
#include <iostream>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <string_view>
#include <thread>
#include <unordered_map>
#include <utility>
#include <vector>
namespace rate_limiter {
static inline uint64_t steady_now_ns() noexcept {
using clock = std::chrono::steady_clock;
return static_cast<uint64_t>(
std::chrono::duration_cast<std::chrono::nanoseconds>(
clock::now().time_since_epoch())
.count());
}
// Token Bucket（整数定点版本）:
// - 单位：token_units = token * 1e9（1 token = 1_000_000_000 units）
// - refill：每 ns 增加 rate_per_sec units（delta_ns * rate_per_sec / 1e9 tokens
// -> *1e9 => delta_ns*rate_per_sec units）
// - 好处：无浮点、更可预测，热点路径只有整数运算
class TokenBucket {
public:
TokenBucket() = default;
TokenBucket(uint64_t rate_per_sec, uint64_t burst_tokens, uint64_t now_ns)
: rate_per_sec_(rate_per_sec),
capacity_units_(mul_1e9_sat(burst_tokens)),
tokens_units_(capacity_units_),
last_ns_(now_ns) {}
bool try_acquire(uint64_t need, uint64_t now_ns) noexcept {
refill(now_ns);
const uint64_t need_units = mul_1e9_sat(need);
if (tokens_units_ >= need_units) {
tokens_units_ -= need_units;
return true;
}
return false;
}
// 返回距离下一次可获取 need 所需等待的时间；nullopt 表示永远等不到（rate=0 且
// tokens 不够）。
std::optional<uint64_t> retry_after_ns(uint64_t need,
uint64_t now_ns) noexcept {
refill(now_ns);
const uint64_t need_units = mul_1e9_sat(need);
if (tokens_units_ >= need_units) return 0ULL;
if (rate_per_sec_ == 0) return std::nullopt;
const uint64_t deficit = need_units - tokens_units_;
// ceil(deficit / rate_per_sec_) since add_units per ns == rate_per_sec_
return (deficit + rate_per_sec_ - 1) / rate_per_sec_;
}
private:
static inline uint64_t mul_1e9_sat(uint64_t x) noexcept {
constexpr uint64_t k = 1'000'000'000ULL;
if (x > UINT64_MAX / k) return UINT64_MAX;
return x * k;
}
void refill(uint64_t now_ns) noexcept {
if (now_ns <= last_ns_) return;
const uint64_t delta_ns = now_ns - last_ns_;
last_ns_ = now_ns;
if (rate_per_sec_ == 0) return;
using u128 = unsigned __int128;
const u128 add128 =
static_cast<u128>(delta_ns) * static_cast<u128>(rate_per_sec_);
uint64_t add_units = (add128 > static_cast<u128>(UINT64_MAX))
? UINT64_MAX
: static_cast<uint64_t>(add128);
if (add_units == 0) return;
const uint64_t space = (tokens_units_ >= capacity_units_)
? 0ULL
: (capacity_units_ - tokens_units_);
if (add_units >= space) {
tokens_units_ = capacity_units_;
} else {
tokens_units_ += add_units;
}
}
uint64_t rate_per_sec_ = 0;
uint64_t capacity_units_ = 0;
uint64_t tokens_units_ = 0;
uint64_t last_ns_ = 0;
};
// 无锁的限流器：
// - per-key 只维护一个原子量 `tat`（GCRA / virtual scheduling）
// - key 存储在固定容量的 lock-free open addressing 表中（避免动态分配与锁）
// 约束：key 容量固定，满了会拒绝；key 用 64-bit hash 表示（冲突概率极低但非
// 0）。
class LockFreeApiRateLimiter {
public:
using NowFn = std::function<uint64_t()>;
struct Options {
uint64_t rate_per_sec = 0; // >0
uint64_t burst = 1; // >=1
size_t capacity = 1 << 20; // 表容量（建议 2^n）
size_t max_probe = 64; // 最大探测步数（控制尾延迟）
};
explicit LockFreeApiRateLimiter(Options opt, NowFn now_fn = steady_now_ns)
: opt_(normalize(std::move(opt))),
now_fn_(std::move(now_fn)),
mask_(opt_.capacity - 1),
tau_ns_(ceil_div_u128(1'000'000'000ULL, opt_.rate_per_sec)),
tolerance_ns_((opt_.burst > 0 ? (opt_.burst - 1) : 0) * tau_ns_),
entries_(opt_.capacity) {}
static uint64_t hash_key(std::string_view key) noexcept {
// FNV-1a 64 + splitmix64 finalizer
uint64_t h = 14695981039346656037ULL;
for (unsigned char c : key) {
h ^= static_cast<uint64_t>(c);
h *= 1099511628211ULL;
}
// splitmix64
h += 0x9e3779b97f4a7c15ULL;
h = (h ^ (h >> 30)) * 0xbf58476d1ce4e5b9ULL;
h = (h ^ (h >> 27)) * 0x94d049bb133111ebULL;
h = h ^ (h >> 31);
if (h <= kReservedMax) h += (kReservedMax + 1);
return h;
}
bool try_acquire(std::string_view key, uint64_t tokens = 1) noexcept {
return try_acquire_hashed(hash_key(key), tokens);
}
bool try_acquire_hashed(uint64_t key_hash, uint64_t tokens = 1) noexcept {
if (tokens == 0) return true;
if (opt_.rate_per_sec == 0) return false;
if (tokens > opt_.burst) return false;
Entry* e = get_or_insert_entry_(normalize_key_(key_hash));
if (!e) return false;
const uint64_t now = now_fn_();
const uint64_t inc = per_request_ns_(tokens);
const uint64_t tol = tolerance_ns_;
// GCRA: accept iff now >= tat - tol, then tat := max(tat, now) + inc
while (true) {
uint64_t tat = e->tat_ns.load(std::memory_order_relaxed);
const uint64_t threshold = (tat > tol) ? (tat - tol) : 0ULL;
if (now < threshold) return false;
const uint64_t base = (tat > now) ? tat : now;
const uint64_t new_tat = add_sat_(base, inc);
if (e->tat_ns.compare_exchange_weak(tat, new_tat,
std::memory_order_acq_rel,
std::memory_order_relaxed)) {
return true;
}
}
}
std::optional<uint64_t> retry_after_ns(std::string_view key,
uint64_t tokens = 1) noexcept {
return retry_after_ns_hashed(hash_key(key), tokens);
}
std::optional<uint64_t> retry_after_ns_hashed(uint64_t key_hash,
uint64_t tokens = 1) noexcept {
if (tokens == 0) return 0ULL;
if (opt_.rate_per_sec == 0) return std::nullopt;
if (tokens > opt_.burst) return std::nullopt;
Entry* e = get_or_insert_entry_(normalize_key_(key_hash));
if (!e) return std::nullopt;
const uint64_t now = now_fn_();
const uint64_t tat = e->tat_ns.load(std::memory_order_relaxed);
const uint64_t threshold =
(tat > tolerance_ns_) ? (tat - tolerance_ns_) : 0ULL;
if (now >= threshold) return 0ULL;
return threshold - now;
}
size_t capacity() const noexcept { return opt_.capacity; }
private:
static constexpr uint64_t kEmpty = 0;
static constexpr uint64_t kTombstone = 1;
static constexpr uint64_t kReserved = 2;
static constexpr uint64_t kReservedMax = kReserved;
struct alignas(64) Entry {
std::atomic<uint64_t> key{kEmpty};
std::atomic<uint64_t> tat_ns{0};
};
static Options normalize(Options o) {
if (o.burst == 0) o.burst = 1;
if (o.capacity < 8) o.capacity = 8;
// 需要 2^n
if ((o.capacity & (o.capacity - 1)) != 0) {
size_t p = 1;
while (p < o.capacity) p <<= 1;
o.capacity = p;
}
if (o.max_probe == 0) o.max_probe = 1;
return o;
}
static inline uint64_t add_sat_(uint64_t a, uint64_t b) noexcept {
const uint64_t r = a + b;
if (r < a) return UINT64_MAX;
return r;
}
static inline uint64_t ceil_div_u128(uint64_t numer,
uint64_t denom) noexcept {
// denom > 0
using u128 = unsigned __int128;
const u128 n = static_cast<u128>(numer);
const u128 d = static_cast<u128>(denom);
return static_cast<uint64_t>((n + d - 1) / d);
}
static inline uint64_t ceil_div_mul_u128(uint64_t a, uint64_t b,
uint64_t denom) noexcept {
using u128 = unsigned __int128;
const u128 n = static_cast<u128>(a) * static_cast<u128>(b);
const u128 d = static_cast<u128>(denom);
return static_cast<uint64_t>((n + d - 1) / d);
}
inline uint64_t per_request_ns_(uint64_t tokens) const noexcept {
// ceil(tokens * 1e9 / rate)
return ceil_div_mul_u128(tokens, 1'000'000'000ULL, opt_.rate_per_sec);
}
static inline uint64_t normalize_key_(uint64_t h) noexcept {
if (h <= kReservedMax) h += (kReservedMax + 1);
return h;
}
Entry* get_or_insert_entry_(uint64_t key_hash) noexcept {
const size_t start = static_cast<size_t>(key_hash) & mask_;
size_t idx = start;
size_t first_tomb = static_cast<size_t>(-1);
for (size_t probe = 0; probe < opt_.max_probe; ++probe) {
Entry& e = entries_[idx];
uint64_t k = e.key.load(std::memory_order_acquire);
if (k == key_hash) return &e;
if (k == kReserved) {
idx = (idx + 1) & mask_;
continue;
}
if (k == kTombstone && first_tomb == static_cast<size_t>(-1))
first_tomb = idx;
if (k == kEmpty) {
const size_t target =
(first_tomb != static_cast<size_t>(-1)) ? first_tomb : idx;
Entry& t = entries_[target];
uint64_t expected = (target == idx) ? kEmpty : kTombstone;
if (t.key.compare_exchange_strong(expected, kReserved,
std::memory_order_acq_rel,
std::memory_order_relaxed)) {
t.tat_ns.store(0, std::memory_order_relaxed);
t.key.store(key_hash, std::memory_order_release);
return &t;
}
// CAS 失败：继续探测
}
idx = (idx + 1) & mask_;
}
return nullptr;
}
Options opt_;
NowFn now_fn_;
size_t mask_;
uint64_t tau_ns_;
uint64_t tolerance_ns_;
std::vector<Entry> entries_;
};
// 生产通用 API 限流器：按 key 独立桶（适合 userId/apiKey/accountId/endpoint
// 等维度）。
// - shard 分片减少锁竞争
// - idle_timeout + prune_interval 防止 key 无界增长
// - max_entries 做硬上限，避免被“key 洪泛”打爆内存
class ApiRateLimiter {
public:
using NowFn = std::function<uint64_t()>;
struct Options {
uint64_t rate_per_sec = 0; // 每秒令牌数
uint64_t burst = 0; // 最大突发
size_t shards = 64; // 分片数（并发越高可适当增大）
size_t max_entries = 100000; // key 上限
uint64_t idle_timeout_ns = 300'000'000'000ULL; // 5min
uint64_t prune_interval_ns = 1'000'000'000ULL; // 1s
};
explicit ApiRateLimiter(Options opt, NowFn now_fn = steady_now_ns)
: opt_(normalize(std::move(opt))),
now_fn_(std::move(now_fn)),
total_entries_(0) {
shards_.reserve(opt_.shards);
for (size_t i = 0; i < opt_.shards; ++i)
shards_.push_back(std::make_unique<Shard>());
}
bool try_acquire(std::string_view key, uint64_t tokens = 1) {
const uint64_t now = now_fn_();
Shard& s = *shards_[shard_index(key)];
std::unique_lock<std::mutex> lk(s.mu);
maybe_prune_locked(s, now);
auto it = s.map.find(key); // heterogeneous lookup: no allocation on hit
if (it == s.map.end()) {
if (!can_insert_locked(s, now)) return false;
auto [ins_it, ok] = s.map.try_emplace(
std::string(key),
Entry{TokenBucket(opt_.rate_per_sec, opt_.burst, now), now});
(void)ok;
it = ins_it;
total_entries_.fetch_add(1, std::memory_order_relaxed);
} else {
it->second.last_seen_ns = now;
}
return it->second.bucket.try_acquire(tokens, now);
}
std::optional<uint64_t> retry_after_ns(std::string_view key,
uint64_t tokens = 1) {
const uint64_t now = now_fn_();
Shard& s = *shards_[shard_index(key)];
std::unique_lock<std::mutex> lk(s.mu);
maybe_prune_locked(s, now);
auto it = s.map.find(key); // heterogeneous lookup
if (it == s.map.end()) {
if (!can_insert_locked(s, now)) return std::nullopt;
auto [ins_it, ok] = s.map.try_emplace(
std::string(key),
Entry{TokenBucket(opt_.rate_per_sec, opt_.burst, now), now});
(void)ok;
it = ins_it;
total_entries_.fetch_add(1, std::memory_order_relaxed);
} else {
it->second.last_seen_ns = now;
}
return it->second.bucket.retry_after_ns(tokens, now);
}
size_t size_approx() const noexcept {
return total_entries_.load(std::memory_order_relaxed);
}
private:
struct TransparentHash {
using is_transparent = void;
size_t operator()(std::string_view s) const noexcept {
return std::hash<std::string_view>{}(s);
}
size_t operator()(const std::string& s) const noexcept {
return std::hash<std::string_view>{}(s);
}
};
struct TransparentEq {
using is_transparent = void;
bool operator()(std::string_view a, std::string_view b) const noexcept {
return a == b;
}
bool operator()(const std::string& a, std::string_view b) const noexcept {
return std::string_view(a) == b;
}
bool operator()(std::string_view a, const std::string& b) const noexcept {
return a == std::string_view(b);
}
bool operator()(const std::string& a, const std::string& b) const noexcept {
return a == b;
}
};
struct Entry {
TokenBucket bucket;
uint64_t last_seen_ns = 0;
};
struct Shard {
std::mutex mu;
std::unordered_map<std::string, Entry, TransparentHash, TransparentEq> map;
uint64_t last_prune_ns = 0;
};
static Options normalize(Options o) {
if (o.shards == 0) o.shards = 1;
if (o.burst == 0) o.burst = 1;
if (o.max_entries == 0) o.max_entries = 1;
if (o.prune_interval_ns == 0) o.prune_interval_ns = 1'000'000'000ULL;
return o;
}
size_t shard_index(std::string_view key) const noexcept {
return std::hash<std::string_view>{}(key) % opt_.shards;
}
void maybe_prune_locked(Shard& s, uint64_t now) {
if (s.last_prune_ns != 0 && now - s.last_prune_ns < opt_.prune_interval_ns)
return;
s.last_prune_ns = now;
if (opt_.idle_timeout_ns == 0) return;
for (auto it = s.map.begin(); it != s.map.end();) {
const uint64_t last = it->second.last_seen_ns;
if (now > last && now - last > opt_.idle_timeout_ns) {
it = s.map.erase(it);
total_entries_.fetch_sub(1, std::memory_order_relaxed);
} else {
++it;
}
}
}
bool can_insert_locked(Shard& s, uint64_t now) {
(void)s;
if (total_entries_.load(std::memory_order_relaxed) < opt_.max_entries)
return true;
// 尝试通过当前 shard 的 prune 释放空间（跨 shard 的 prune
// 由调用路径自然触发）。 这里再 prune
// 一次提高“满了但其实已经过期”的成功率。
for (const auto& shard_ptr : shards_) {
Shard& shard = *shard_ptr;
std::unique_lock<std::mutex> lk(shard.mu, std::defer_lock);
if (!lk.try_lock()) continue;
maybe_prune_locked(shard, now);
if (total_entries_.load(std::memory_order_relaxed) < opt_.max_entries)
return true;
}
return total_entries_.load(std::memory_order_relaxed) < opt_.max_entries;
}
Options opt_;
NowFn now_fn_;
std::vector<std::unique_ptr<Shard>> shards_;
std::atomic<size_t> total_entries_;
};
} // namespace rate_limiter
// ------------------------- Tests -------------------------
namespace {
struct FakeClock {
std::atomic<uint64_t> now_ns{0};
uint64_t operator()() const noexcept {
return now_ns.load(std::memory_order_relaxed);
}
void advance_ns(uint64_t delta) noexcept {
now_ns.fetch_add(delta, std::memory_order_relaxed);
}
};
static void test_burst_and_refill() {
FakeClock clk;
rate_limiter::LockFreeApiRateLimiter::Options opt;
opt.rate_per_sec = 10;
opt.burst = 5;
opt.capacity = 1024;
opt.max_probe = 64;
rate_limiter::LockFreeApiRateLimiter rl(opt, [&] { return clk(); });
for (int i = 0; i < 5; ++i) assert(rl.try_acquire("u1", 1));
assert(!rl.try_acquire("u1", 1));
clk.advance_ns(500'000'000ULL); // 0.5s -> +5 tokens
for (int i = 0; i < 5; ++i) assert(rl.try_acquire("u1", 1));
assert(!rl.try_acquire("u1", 1));
}
static void test_per_key_isolation() {
FakeClock clk;
rate_limiter::LockFreeApiRateLimiter::Options opt;
opt.burst = 2;
opt.rate_per_sec = 1; // 用极低速率避免 rate=0 特殊情况
opt.capacity = 1024;
rate_limiter::LockFreeApiRateLimiter rl(opt, [&] { return clk(); });
assert(rl.try_acquire("A", 1));
assert(rl.try_acquire("A", 1));
assert(!rl.try_acquire("A", 1));
assert(rl.try_acquire("B", 2));
assert(!rl.try_acquire("B", 1));
}
static void test_retry_after() {
FakeClock clk;
rate_limiter::LockFreeApiRateLimiter::Options opt;
opt.rate_per_sec = 10;
opt.burst = 1;
opt.capacity = 1024;
rate_limiter::LockFreeApiRateLimiter rl(opt, [&] { return clk(); });
assert(rl.try_acquire("k", 1));
auto ra = rl.retry_after_ns("k", 1);
assert(ra.has_value());
assert(*ra >= 100'000'000ULL); // 1 token @ 10/s => ~0.1s
clk.advance_ns(100'000'000ULL);
assert(rl.try_acquire("k", 1));
}
static void test_max_entries_and_prune() {
FakeClock clk;
rate_limiter::LockFreeApiRateLimiter::Options opt;
opt.rate_per_sec = 10;
opt.burst = 1;
opt.capacity = 8;
opt.max_probe = 8;
rate_limiter::LockFreeApiRateLimiter rl(opt, [&] { return clk(); });
// 固定容量：确保不会无限插入；这里用不同 key 填满，之后应拒绝插入新 key。
assert(rl.try_acquire("A", 1));
assert(rl.try_acquire("B", 1));
assert(rl.try_acquire("C", 1));
assert(rl.try_acquire("D", 1));
assert(rl.try_acquire("E", 1));
assert(rl.try_acquire("F", 1));
assert(rl.try_acquire("G", 1));
assert(rl.try_acquire("H", 1));
// table 很可能已满（或接近满）；再来一个新 key 应该失败（尾部受 max_probe
// 控制）
(void)rl.try_acquire("I", 1);
}
static void test_multithread_smoke() {
// 不做严格计数断言（容易因调度抖动），只验证线程安全与基本行为。
rate_limiter::LockFreeApiRateLimiter::Options opt;
opt.rate_per_sec = 5000;
opt.burst = 1000;
opt.capacity = 4096;
opt.max_probe = 64;
rate_limiter::LockFreeApiRateLimiter rl(opt);
std::atomic<uint64_t> ok{0};
auto worker = [&](int tid) {
const std::string key = "acct-" + std::to_string(tid % 8);
const auto end =
std::chrono::steady_clock::now() + std::chrono::milliseconds(50);
while (std::chrono::steady_clock::now() < end) {
if (rl.try_acquire(key, 1)) ok.fetch_add(1, std::memory_order_relaxed);
}
};
std::vector<std::thread> threads;
for (int i = 0; i < 8; ++i) threads.emplace_back(worker, i);
for (auto& t : threads) t.join();
assert(ok.load(std::memory_order_relaxed) > 0);
}
} // namespace
int main() {
test_burst_and_refill();
test_per_key_isolation();
test_retry_after();
test_max_entries_and_prune();
test_multithread_smoke();
std::cout << "All tests passed.\n";
return 0;
}
/*
==================== 使用示例 1：ApiRateLimiter====================
// 订单/下单 API：按 accountId 限流：1000 req/s，允许突发 200
rate_limiter::ApiRateLimiter::Options opt;
opt.rate_per_sec = 1000;
opt.burst = 200;
opt.shards = 64; // 并发更高可调大
opt.max_entries = 200000; // 可控 key 上限（防 key 洪泛）
opt.idle_timeout_ns = 5ULL * 60 * 1000 * 1000 * 1000; // 5min
opt.prune_interval_ns = 1ULL * 1000 * 1000 * 1000; // 1s
rate_limiter::ApiRateLimiter limiter(opt);
std::string key = "acct:" + accountId;
if (!limiter.try_acquire(key, 1)) {
// 返回 429 / 触发降级
auto wait = limiter.retry_after_ns(key, 1);
// wait.has_value() 时可回填 Retry-After / 自适应退避
}
==================== 使用示例 2：LockFreeApiRateLimiter====================
// 连接级/账户级高频限流：50_000 req/s，突发 1000；容量固定（必须可控）
rate_limiter::LockFreeApiRateLimiter::Options lf;
lf.rate_per_sec = 50'000;
lf.burst = 1000;
lf.capacity = 1 << 20; // 可容纳约 1,048,576 个 key（负载越低尾延迟越好）
lf.max_probe = 64; // 控制最坏探测步数（尾延迟）
rate_limiter::LockFreeApiRateLimiter fast(lf);
// 直接用 string key
if (!fast.try_acquire("acct:123", 1)) {
auto wait = fast.retry_after_ns("acct:123", 1);
}
// 或者提前 hash（避免重复 hash 成本）
uint64_t h = rate_limiter::LockFreeApiRateLimiter::hash_key("acct:123");
if (!fast.try_acquire_hashed(h, 1)) {
auto wait = fast.retry_after_ns_hashed(h, 1);
}
*/