算法实战：限流与鉴权

一、问题背景

微服务架构中，两个核心安全问题：

• 鉴权：如何快速判断一个请求是否有权限访问某资源？
• 限流：如何控制接口调用频率，防止系统被打垮？

两者背后都依赖经典数据结构与算法。

二、鉴权：快速权限判断

2.1 权限模型

graph LR
  U[用户] --> R[角色]
  R --> P[权限]
  P --> API[接口资源]

典型 RBAC（基于角色的访问控制）：

• 用户 → 角色（多对多）
• 角色 → 权限（多对多）
• 权限 → 接口 URL

2.2 数据结构选择

鉴权是超高频读操作（每个请求都要判断），需要 O(1) 查找：

2.3 实现方案

public class AuthService {
    // 用户 -> 角色集合
    private Map<String, Set<String>> userRoles = new HashMap<>();
    // 角色 -> 权限URL集合
    private Map<String, Set<String>> rolePermissions = new HashMap<>();

    public boolean hasPermission(String userId, String requestUrl) {
        Set<String> roles = userRoles.get(userId);
        if (roles == null) return false;

        for (String role : roles) {
            Set<String> perms = rolePermissions.get(role);
            if (perms != null && matchUrl(perms, requestUrl)) {
                return true;
            }
        }
        return false;
    }

    private boolean matchUrl(Set<String> perms, String url) {
        // 精确匹配
        if (perms.contains(url)) return true;
        // 前缀通配: /api/admin/* 匹配 /api/admin/users
        for (String perm : perms) {
            if (perm.endsWith("/*")) {
                String prefix = perm.substring(0, perm.length() - 1);
                if (url.startsWith(prefix)) return true;
            }
        }
        return false;
    }
}

2.4 性能优化

三、限流：保护系统不被打垮

3.1 四种经典限流算法

3.2 固定窗口计数器

public class FixedWindowLimiter {
    private final int limit;
    private final long windowMs;
    private long windowStart;
    private int count;

    public FixedWindowLimiter(int limit, long windowMs) {
        this.limit = limit;
        this.windowMs = windowMs;
        this.windowStart = System.currentTimeMillis();
        this.count = 0;
    }

    public synchronized boolean tryAcquire() {
        long now = System.currentTimeMillis();
        if (now - windowStart >= windowMs) {
            windowStart = now;
            count = 0;
        }
        if (count < limit) {
            count++;
            return true;
        }
        return false;
    }
}

3.3 令牌桶（推荐）

import time

class TokenBucket:
    def __init__(self, rate, capacity):
        """
        rate: 每秒放入令牌数
        capacity: 桶容量（允许的最大突发）
        """
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_time = time.time()

    def try_acquire(self, tokens=1):
        now = time.time()
        # 补充令牌
        elapsed = now - self.last_time
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_time = now

        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

3.4 漏桶 vs 令牌桶

graph TD
  subgraph 漏桶
    A1[请求流入] --> B1[桶]
    B1 --> C1[固定速率流出]
  end
  subgraph 令牌桶
    A2[匀速放令牌] --> B2[桶]
    B2 --> C2[请求取令牌]
  end

四、分布式限流

单机限流用内存计数器即可，分布式场景需要：

-- Redis Lua 限流脚本
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local current = redis.call('INCR', key)
if current == 1 then
    redis.call('EXPIRE', key, 60)
end
if current > limit then
    return 0
end
return 1

五、总结

设计原则：

• 鉴权追求低延迟（每请求必走）→ 内存缓存 + 位运算
• 限流追求精确与平滑→ 令牌桶允许合理突发
• 分布式场景 → Redis 原子操作保证一致性