Libz's Blog

一、问题背景：当多个租户共享一个线程池#

想象你在一个微服务网关层工作。每秒有数十万请求涌入，它们来自不同的租户——有核心 API 服务（要求毫秒级响应）、有 CDN 合作伙伴（中等优先级）、有数据分析服务（可容忍延迟）、还有免费试用用户（尽力而为）。

所有这些请求共享同一个线程池。

问题来了：如何保证核心租户在系统过载时依然获得优先服务，同时不让低优先级租户完全饿死？

这不是一个简单的”设置线程优先级”就能解决的问题。Java 的 Thread.setPriority() 在现代操作系统上几乎没有实际效果，而且线程池中的线程是复用的，优先级会被继承和污染。

我们需要的是一套应用层的调度引擎——在线程池之上，自己实现公平性、限流、背压和熔断。

本文将完整拆解这样一个多租户自适应调度引擎的设计与实现。

二、整体架构总览#

先看全局视角，理解各组件之间的关系：

三、租户 SLA 模型：用枚举定义资源配额#

一切的起点是租户分级。不同等级的租户拥有不同的资源配额：

1
public enum TierLevel {
2
    PLATINUM(8, 10000, 500), // 权重8，令牌桶容量10000，队列深度500
3
    GOLD(4, 5000, 300),
4
    SILVER(2, 2000, 200),
5
    BRONZE(1, 1000, 100);
6

7
    final int weight;              // 调度权重：决定获得多少 CPU 时间片
8
    final int tokenBucketCapacity; // 令牌桶容量：决定突发流量的承受能力
9
    final int maxQueueDepth;       // 最大队列深度：决定背压触发阈值
10
}

三个维度的配额形成了一个立体的资源控制模型：

PLATINUM 租户的令牌补充速率是 BRONZE 的 8 倍，队列容量是 5 倍。这意味着在系统过载时，BRONZE 租户会最先被限流和背压拒绝，而 PLATINUM 租户几乎不受影响。

四、submit 方法：四层防御的分层漏斗#

submit 是整个引擎的核心入口，每个请求都要经过四道关卡。设计哲学是：成本最低的检查放最前面，成本最高的操作放最后面。

1
if (ctx.circuitState == CircuitState.OPEN) {
2
    if (System.currentTimeMillis() - ctx.lastFailureTimestamp > 5000) {
3
        ctx.circuitState = CircuitState.HALF_OPEN; // 冷却后试探
4
    } else {
5
        ctx.rejectedCount.incrementAndGet();
6
        return CompletableFuture.failedFuture(...);  // 秒拒
7
    }
8
}

1
boolean tryAcquire() {
2
    refill();  // 先补充令牌
3
    while (true) {
4
        long current = tokens.get();
5
        if (current <= 0) return false;           // 无令牌，拒绝
6
        if (tokens.compareAndSet(current, current - 1))
7
            return true;                           // CAS 成功，获取令牌
8
        // CAS 失败 → 自旋重试（比 synchronized 高效）
9
    }
10
}

1
private void refill() {
2
    long now = System.nanoTime();
3
    long elapsed = now - lastRefillTime;
4
    if (elapsed < 1_000_000) return; // 至少 1ms 才补充，减少无效计算
5

6
    long tokensToAdd = (long)(elapsed / 1_000_000_000.0 * refillRate * adaptiveFactor);
7
    if (tokensToAdd > 0) {
8
        lastRefillTime = now;
9
        tokens.getAndUpdate(current -> Math.min(capacity, current + tokensToAdd));
10
    }
11
}

1
if (ctx.isQueueFull()) {  // taskQueue.size() >= tier.maxQueueDepth
2
    ctx.rejectedCount.incrementAndGet();
3
    return CompletableFuture.failedFuture(
4
        new RejectedExecutionException("[背压] 队列已满"));
5
}

1
ScheduledTask<T> task = new ScheduledTask<>(tenantId, priority, work);
2
ctx.taskQueue.offer(task);  // PriorityBlockingQueue，按优先级排序
3
ctx.submittedCount.incrementAndGet();
4
return task.resultFuture;   // 返回 CompletableFuture

1
public int compareTo(ScheduledTask other) {
2
    int cmp = Integer.compare(this.priority, other.priority);
3
    return cmp != 0 ? cmp : Long.compare(this.enqueueTimeNano, other.enqueueTimeNano);
4
}

五、调度核心：Deficit Round Robin + Work-Stealing#

任务入队后，由 Worker 线程负责消费。这里用了两个经典算法的组合。

1
private void workerLoop() {
2
    Map<String, Integer> deficitCounters = new HashMap<>(); // 每个 Worker 独立维护
3

4
    while (running) {
5
        boolean didWork = false;
6

7
        // Phase 1: DRR 调度
8
        for (Map.Entry<String, TenantContext> entry : tenants.entrySet()) {
9
            String tenantId = entry.getKey();
10
            TenantContext ctx = entry.getValue();
11

12
            // 累加 deficit
13
            deficitCounters.merge(tenantId, ctx.tier.weight, Integer::sum);
14

15
            int deficit = deficitCounters.getOrDefault(tenantId, 0);
16
            while (deficit > 0) {
17
                ScheduledTask<?> task = ctx.taskQueue.poll();
18
                if (task == null) break;  // 队列空了，deficit 保留到下一轮
19

20
                executeTask(ctx, task);
21
                deficit--;
22
                didWork = true;
23
            }
24
            deficitCounters.put(tenantId, Math.max(0, deficit));
25
        }
26

27
        // Phase 2: Work-Stealing
28
        if (!didWork) {
29
            didWork = tryWorkStealing();
30
        }
31

32
        // Phase 3: 无任务时休眠 1ms
33
        if (!didWork) {
34
            Thread.sleep(1);
35
        }
36
    }
37
}

1
private boolean tryWorkStealing() {
2
    TenantContext busiest = null;
3
    int maxDepth = 0;
4

5
    // 找到队列最深的租户
6
    for (TenantContext ctx : tenants.values()) {
7
        int depth = ctx.currentQueueDepth();
8
        if (depth > maxDepth) {
9
            maxDepth = depth;
10
            busiest = ctx;
11
        }
12
    }
13

14
    // 从最繁忙的队列偷一个任务
15
    if (busiest != null) {
16
        ScheduledTask<?> stolen = busiest.taskQueue.poll();
17
        if (stolen != null) {
18
            executeTask(busiest, stolen);
19
            return true;
20
        }
21
    }
22
    return false;
23
}

六、熔断与自愈：executeTask 中的状态机驱动#

任务执行不仅仅是 task.run()，还承担着熔断器状态转换和指标采集的职责：

1
private void executeTask(TenantContext ctx, ScheduledTask<?> task) {
2
    // 采集调度延迟
3
    long scheduleLatencyMicros = (System.nanoTime() - task.enqueueTimeNano) / 1000;
4
    ctx.scheduleLatency.record(scheduleLatencyMicros);
5

6
    try {
7
        task.run();
8
        ctx.completedCount.incrementAndGet();
9

10
        // 熔断恢复：HALF_OPEN 状态下成功 → 回到 CLOSED
11
        if (ctx.circuitState == CircuitState.HALF_OPEN) {
12
            ctx.circuitState = CircuitState.CLOSED;
13
            ctx.consecutiveFailures.set(0);
14
        }
15
    } catch (Exception e) {
16
        int failures = ctx.consecutiveFailures.incrementAndGet();
17
        ctx.lastFailureTimestamp = System.currentTimeMillis();
18

19
        // 连续失败 ≥ 5 次 → 触发熔断
20
        if (failures >= 5) {
21
            ctx.circuitState = CircuitState.OPEN;
22
        }
23
    }
24
}

熔断器的完整生命周期：

1
volatile CircuitState circuitState = CircuitState.CLOSED;

七、自适应控制器：系统的”自主神经”#

每秒执行一次的自适应控制器，是整个引擎的”大脑”：

1
private void adaptiveControl() {
2
    // 计算系统负载 = 所有队列深度之和 / 总容量
3
    long totalDepth = tenants.values().stream()
4
            .mapToLong(TenantContext::currentQueueDepth).sum();
5
    long totalCapacity = tenants.values().stream()
6
            .mapToLong(ctx -> ctx.tier.maxQueueDepth).sum();
7

8
    currentSystemLoad = totalCapacity > 0 ? (double) totalDepth / totalCapacity : 0;
9

10
    // 通知所有令牌桶调整速率
11
    tenants.values().forEach(ctx -> ctx.tokenBucket.adjustFactor(currentSystemLoad));
12
}

这个设计形成了一个负反馈回路：

这就是控制论中的比例控制器（P-Controller）——偏差越大，修正力度越大。系统会自动趋向一个动态平衡点。

八、P50/P99 延迟追踪：无锁直方图#

监控是生产系统的眼睛。LatencyTracker 用 ConcurrentSkipListMap 实现了一个无锁的直方图：

1
public static class LatencyTracker {
2
    private final ConcurrentSkipListMap<Long, AtomicInteger> histogram = new ConcurrentSkipListMap<>();
3
    private final AtomicLong totalCount = new AtomicLong();
4

5
    void record(long latencyMicros) {
6
        long bucket = (latencyMicros / 100) * 100; // 100μs 粒度桶化
7
        histogram.computeIfAbsent(bucket, k -> new AtomicInteger()).incrementAndGet();
8
        totalCount.incrementAndGet();
9
    }
10

11
    long getPercentile(double percentile) {
12
        long target = (long) (totalCount.get() * percentile);
13
        long cumulative = 0;
14
        for (Map.Entry<Long, AtomicInteger> entry : histogram.entrySet()) {
15
            cumulative += entry.getValue().get();
16
            if (cumulative >= target) return entry.getKey();
17
        }
18
        return 0;
19
    }
20
}

为什么用 ConcurrentSkipListMap 而不是 ConcurrentHashMap？

因为计算百分位数需要有序遍历。ConcurrentSkipListMap 的 key 天然有序，遍历时从小到大累加，到达目标百分位就返回。ConcurrentHashMap 无序，需要额外排序。

100μs 的桶粒度是怎么选的？

• 太细（1μs）：桶太多，内存爆炸
• 太粗（10ms）：精度不够，P99 没有参考价值
• 100μs：对于微服务场景，足够区分”快”和”慢”，同时内存可控（即使延迟分布在 0-1s，也只有 10000 个桶）

九、流量模拟：真实场景的压力测试#

main 方法模拟了四种真实的流量模式：

其中 analytics 租户的突发流量（50 → 500 RPS，10 倍突增）是最有意思的测试场景。它会触发：

1. 令牌桶快速耗尽：SILVER 的令牌桶容量只有 2000，补充速率 1000/s，500 RPS 的突发会在约 4 秒内耗尽
2. 队列快速填满：maxQueueDepth = 200，如果消费速度跟不上，很快触发背压
3. 自适应控制器介入：全局负载上升，所有租户的令牌补充速率下降
4. Work-Stealing 启动：analytics 队列变深，空闲 Worker 自动过来帮忙

这个场景完美展示了各个组件如何协同工作。

1
private static void busyWaitMicros(long micros) {
2
    long waitUntil = System.nanoTime() + micros * 1000;
3
    while (System.nanoTime() < waitUntil) {
4
        Thread.onSpinWait(); // JDK 9+ CPU hint
5
    }
6
}

十、数据结构选型总结#

十一、并发安全性分析#

整个引擎的并发模型可以归纳为：

关键的并发安全保证：

1. 令牌桶：AtomicLong + CAS 保证原子性，volatile lastRefillTime 保证可见性
2. 任务队列：PriorityBlockingQueue 本身线程安全
3. 计数器：全部使用 AtomicLong/AtomicInteger
4. 熔断状态：volatile 保证可见性，状态转换是幂等的
5. deficit 计数器：线程局部变量，无需同步
6. adaptiveFactor：单线程写（自适应控制器）、多线程读（令牌桶），volatile 足够

没有一把 synchronized 锁，没有一个 ReentrantLock。 整个热路径（submit → 入队 → 出队 → 执行）全部是无锁或 lock-free 的。

十二、设计模式与架构思想总结#

十三、可以继续优化的方向#

这个引擎已经是一个相当完整的实现，但在生产环境中还可以考虑：

1. 令牌桶的 refill 竞态：多线程同时调用 refill() 时，lastRefillTime 的更新不是原子的，可能导致少量令牌多补或少补。可以用 CAS 保护 lastRefillTime 的更新。

2. 熔断器的并发状态转换：多个线程可能同时将 OPEN 转为 HALF_OPEN，导致多个”试探请求”同时放行。可以用 compareAndSet 保证只有一个线程成功转换。

3. 队列深度检查与入队的非原子性：isQueueFull() 和 taskQueue.offer() 之间有时间窗口，可能导致队列实际深度略超 maxQueueDepth。在实践中这个误差可以接受，但如果需要严格控制，可以用 Semaphore 做精确的容量控制。

4. 指标采集的内存增长：LatencyTracker 的直方图会持续增长，长时间运行后需要考虑滑动窗口或定期重置。

5. 更精细的 Work-Stealing 策略：当前是从最深的队列偷取，可以考虑加入租户权重因素——优先偷取高权重租户的任务。

十四、总结#

这个多租户调度引擎的核心设计哲学可以用一句话概括：

用最小代价、在最早的时机，把不该进来的请求挡在门外，同时保证高等级租户永远比低等级租户更难被拒绝。

它不是某一个算法的胜利，而是多个经典算法和设计模式的精心编排：

• 入口层用分层漏斗做快速失败
• 调度层用 DRR + Work-Stealing 做加权公平
• 控制层用负反馈回路做自适应调节
• 保护层用熔断器做优雅降级

每一层都在做自己最擅长的事，组合在一起就形成了一个能扛住十万级 QPS、在过载时优雅降级而不是雪崩的调度系统。