《Java 并发编程·性能诊断与事故》5.3 队列长度、拒绝率与降级

队列长度 ≈ 可接受排队时间 × 服务速率

Abort      rejected=3738 rejectRate=37.4% done=6262 p50=56.6ms p99=65.0ms 提交线程阻塞=0.10s
CallerRuns rejected=0    rejectRate= 0.0% done=10000 p50=56.4ms p99=70.3ms 提交线程阻塞=7.31s
Discard    rejected=0    rejectRate= 0.0% done=6453 p50=54.7ms p99=63.1ms 提交线程阻塞=0.01s

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

/** 定速投递，统计拒绝率与被接受请求的分位数。 */
public class RejectCurveProbe {
    static final long SERVICE_NANOS = 5_000_000L;
    static final long RUN_NANOS = 5_000_000_000L;
    static final int ARRIVAL_PER_SEC = 2000;

    public static void main(String[] args) throws Exception {
        for (int poolSize : new int[]{4, 16}) {
            for (int queueSize : new int[]{0, 16, 64, 256}) {
                run(poolSize, queueSize);
            }
        }
    }

    static void run(int poolSize, int queueSize) throws Exception {
        BlockingQueue<Runnable> queue = queueSize == 0
                ? new SynchronousQueue<>()
                : new ArrayBlockingQueue<>(queueSize);
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                poolSize, poolSize, 0, TimeUnit.MILLISECONDS, queue,
                new ThreadPoolExecutor.AbortPolicy());

        LongAdder accepted = new LongAdder();
        LongAdder rejected = new LongAdder();
        long[] latency = new long[1 << 19];
        AtomicInteger idx = new AtomicInteger();

        long start = System.nanoTime();
        long interval = 1_000_000_000L / ARRIVAL_PER_SEC;
        for (long k = 0; start + k * interval < start + RUN_NANOS; k++) {
            long target = start + k * interval;
            if (target - System.nanoTime() > 200_000L) {
                LockSupport.parkNanos(target - System.nanoTime() - 100_000L);
            }
            while (System.nanoTime() < target) {
                Thread.onSpinWait();
            }
            long submitAt = System.nanoTime();
            try {
                pool.execute(() -> {
                    LockSupport.parkNanos(SERVICE_NANOS);
                    int i = idx.getAndIncrement();
                    if (i < latency.length) {
                        latency[i] = System.nanoTime() - submitAt;
                    }
                });
                accepted.increment();
            } catch (RejectedExecutionException e) {
                rejected.increment();
            }
        }

        // 排空剩余任务，最多等 3 秒
        long deadline = System.nanoTime() + 3_000_000_000L;
        while ((queue.size() > 0 || pool.getActiveCount() > 0)
                && System.nanoTime() < deadline) {
            Thread.sleep(20);
        }
        pool.shutdownNow();

        long sub = accepted.sum() + rejected.sum();
        long[] s = Arrays.copyOf(latency, Math.min(idx.get(), latency.length));
        Arrays.sort(s);
        System.out.printf("pool=%2d queue=%3d rejection=%5.1f%% p50=%6.1fms p99=%7.1fms%n",
                poolSize, queueSize, rejected.sum() * 100.0 / sub,
                pct(s, 0.50), pct(s, 0.99));
    }

    static double pct(long[] sorted, double p) {
        if (sorted.length == 0) {
            return -1;
        }
        return sorted[(int) Math.min(sorted.length - 1L,
                Math.round(p * (sorted.length - 1)))] / 1e6;
    }
}

pool= 4 queue=  0 rejection= 71.1%  p50=  6.1ms p99=   8.0ms
pool= 4 queue= 16 rejection= 68.4%  p50= 30.8ms p99=  38.3ms
pool= 4 queue= 64 rejection= 68.3%  p50=108.1ms p99= 125.0ms
pool= 4 queue=256 rejection= 67.3%  p50=428.0ms p99= 457.4ms
pool=16 queue=  0 rejection=  2.7%  p50=  6.1ms p99=   9.6ms
pool=16 queue= 16 rejection=  0.5%  p50=  6.1ms p99=  11.8ms
pool=16 queue= 64 rejection=  0.0%  p50=  6.1ms p99=  10.0ms
pool=16 queue=256 rejection=  0.0%  p50=  6.3ms p99=  14.0ms

--- 第二次运行（拒绝率 / p99）---
pool= 4: 71.4%/8.5ms  68.9%/37.5ms  68.3%/121.0ms  66.8%/448.1ms
pool=16:  3.5%/10.9ms  0.2%/8.6ms    0.0%/15.2ms    0.0%/14.7ms

拒绝率 = 1 - 容量 / 到达率

package com.plume.conc.ops;

import java.util.concurrent.*;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Supplier;

/** 借阅查询入口：有界队列、快速失败、降级返回值、拒绝与降级可观测。 */
public final class BorrowGateway {
    private final ThreadPoolExecutor pool;
    private final long timeoutMillis;
    private final LongAdder rejected = new LongAdder();
    private final LongAdder degraded = new LongAdder();

    public BorrowGateway(int threads, int queueCapacity, long timeoutMillis) {
        this.pool = new ThreadPoolExecutor(
                threads, threads, 60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(queueCapacity),
                r -> new Thread(r, "borrow-gw"),
                new ThreadPoolExecutor.AbortPolicy());
        this.timeoutMillis = timeoutMillis;
    }

    /** fallback 只在拒绝、超时、异常时调用，必须自己保证不抛异常。 */
    public String query(String isbn, Supplier<String> fallback) {
        Future<String> future;
        try {
            future = pool.submit(() -> load(isbn));
        } catch (RejectedExecutionException e) {
            rejected.increment();
            degraded.increment();
            return fallback.get();
        }
        try {
            return future.get(timeoutMillis, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            future.cancel(true);              // 中断下游调用，避免线程继续被占用
            degraded.increment();
            return fallback.get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            future.cancel(true);
            return fallback.get();
        } catch (ExecutionException e) {
            degraded.increment();
            return fallback.get();
        }
    }

    public long rejectedCount() {
        return rejected.sum();
    }

    public long degradedCount() {
        return degraded.sum();
    }

    public int queued() {
        return pool.getQueue().size();
    }

    private String load(String isbn) {
        return "book:" + isbn;
    }
}