《Java 并发编程·性能诊断与事故》5.1 Little's law 在池化中的应用

Σ W_i

∫₀ᵀ L(t) dt = Σ W_i

(1/T) ∫₀ᵀ L(t) dt = (N/T) × (Σ W_i / N)
       平均 L      =    λ    ×    平均 W

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.concurrent.locks.LockSupport;

public class LittleLawProbe {
    static final int POOL = 8;
    static final long SERVICE_NANOS = 20_000_000L;   // 名义 20 ms 服务
    static final long RUN_NANOS = 8_000_000_000L;

    public static void main(String[] args) throws Exception {
        for (int rate : new int[]{100, 200, 300}) {
            run(rate);
        }
    }

    static void run(int arrivalPerSec) throws Exception {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                POOL, POOL, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
        pool.prestartAllCoreThreads();

        LongAdder completed = new LongAdder();
        LongAdder sumNanos = new LongAdder();
        AtomicBoolean running = new AtomicBoolean(true);
        LongAdder lSum = new LongAdder();
        LongAdder lCount = new LongAdder();

        long t0 = System.nanoTime();
        Thread sampler = Thread.ofPlatform().start(() -> {
            while (running.get()) {
                lSum.add(pool.getQueue().size() + pool.getActiveCount());
                lCount.increment();
                LockSupport.parkNanos(5_000_000L);
            }
        });

        long interval = 1_000_000_000L / arrivalPerSec;
        Thread pacer = Thread.ofPlatform().start(() -> {
            long start = System.nanoTime();
            long deadline = start + RUN_NANOS;
            for (long k = 0; ; k++) {
                long target = start + k * interval;
                if (target >= deadline) {
                    break;
                }
                if (target - System.nanoTime() > 200_000L) {
                    LockSupport.parkNanos(target - System.nanoTime() - 100_000L);
                }
                while (System.nanoTime() < target) {
                    Thread.onSpinWait();
                }
                long submitAt = System.nanoTime();
                pool.execute(() -> {
                    LockSupport.parkNanos(SERVICE_NANOS);
                    sumNanos.add(System.nanoTime() - submitAt);
                    completed.increment();
                });
            }
        });

        pacer.join();
        while ((pool.getQueue().size() > 0 || pool.getActiveCount() > 0)
                && System.nanoTime() - t0 < 13_000_000_000L) {
            Thread.sleep(50);
        }
        long elapsed = System.nanoTime() - t0;
        running.set(false);
        sampler.join();
        pool.shutdownNow();

        long done = completed.sum();
        double lambda = done / (elapsed / 1e9);
        double wMillis = sumNanos.sum() / (double) done / 1e6;
        double lSample = lSum.sum() / (double) lCount.sum();
        System.out.printf("target=%d/s actual=%.1f/s W=%.2fms L_sample=%.2f L_calc=%.2f%n",
                arrivalPerSec, lambda, wMillis, lSample, lambda * wMillis / 1000.0);
    }
}

target=100/s actual=99.3/s W=27.05ms L_sample=2.69 L_calc=2.69
target=200/s actual=198.7/s W=26.91ms L_sample=5.45 L_calc=5.35
target=300/s actual=289.7/s W=165.49ms L_sample=48.25 L_calc=47.95

--- 第二次运行 ---
target=100/s actual=99.2/s W=27.33ms L_sample=2.75 L_calc=2.71
target=200/s actual=198.6/s W=27.40ms L_sample=5.59 L_calc=5.44
target=300/s actual=291.7/s W=90.36ms L_sample=26.66 L_calc=26.36

L = 800 × 0.045 = 36

package com.plume.conc.ops;

/** 基于 Little's law 的池容量估算。所有入参必须来自同一段稳态观测窗口。 */
public record PoolPlan(double qps, double meanLatencyMillis, double targetUtilization) {

    public PoolPlan {
        if (qps <= 0 || meanLatencyMillis <= 0) {
            throw new IllegalArgumentException("qps 与延迟必须为正");
        }
        if (targetUtilization <= 0 || targetUtilization >= 1) {
            throw new IllegalArgumentException("利用率目标必须在 (0,1) 开区间");
        }
    }

    /** 稳态下同时在途的平均请求数：L = lambda * W。 */
    public double concurrency() {
        return qps * meanLatencyMillis / 1000.0;
    }

    /** 给定并发下的吞吐上限：lambda_max = L / W。 */
    public static double capacity(double concurrency, double meanLatencyMillis) {
        return concurrency / (meanLatencyMillis / 1000.0);
    }

    /** 为达到目标利用率所需的并发，向上取整。 */
    public long requiredConcurrency() {
        return (long) Math.ceil(concurrency() / targetUtilization);
    }

    /** 当前配置的利用率；大于 1 表示已经过载。 */
    public double utilization(double concurrency) {
        return concurrency() / concurrency;
    }
}