您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java 流控與熔斷降級：令牌桶算法實現與 DelayQueue 延遲隊列實戰

編輯：Java 相關知識

業務場景：高併發下的流量整形

在短信網關、API 閘道、消息隊列等場景中，需要對下游調用進行「每秒最大請求數」限制。本文實現一個基於半秒滑動窗口的流控器，並配合 JDK DelayQueue 實現超限請求的「延遲重試/異步處理」，完美替代 .NET 的 RegisterWaitForSingleObject。

核心組件一：半秒滑動窗口流控器

設計思路：將 1 秒拆分為兩個 500ms 槽位，動態計算當前槽位可用配額，避免「邊界突刺」問題。

public class OverflowController {
    private final int maxPerSecond;      // 秒級總閾值
    private long windowStart;            // 當前半秒窗口起始時間戳
    private int currentCount;            // 當前窗口已用配額

    public OverflowController(int maxPerSecond) {
        this.maxPerSecond = Math.max(maxPerSecond, 2); // 最小閾值 2
    }

    /** 獲取當前半秒窗口可用配額 */
    private int getCurrentQuota() {
        long nowMs = System.currentTimeMillis();
        long halfSec = nowMs / 500; // 當前半秒槽位編號
        if (halfSec != windowStart / 500) { // 跨窗口重置
            windowStart = halfSec * 500;
            currentCount = 0;
        }
        return maxPerSecond / 2; // 每半秒分配一半額度
    }

    /** 判斷是否超限，未超限則消耗配額 */
    public synchronized boolean isOverflow(int permits) {
        int quota = getCurrentQuota();
        if (currentCount + permits > quota) return true;
        currentCount += permits;
        return false;
    }
}

測試驗證：設定 50/s，每 10ms 發送 1 次，前 25 次成功，後續進入超限分支。

核心組件二：DelayQueue 延遲處理超限請求

超限請求不直接丟棄，而是封裝為 Delayed 任務投入 DelayQueue，由消費者線程定時拉取執行，實現「削峰填谷」。

1. 延遲任務實體

public class DelayEntry implements Delayed {
    private final long executeTime; // 到期絕對時間戳
    private final int payload;      // 業務負載

    public DelayEntry(int payload, long delayMs) {
        this.payload = payload;
        this.executeTime = System.currentTimeMillis() + delayMs;
    }

    @Override public long getDelay(TimeUnit unit) {
        return unit.convert(executeTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }
    @Override public int compareTo(Delayed o) {
        return Long.compare(this.executeTime, ((DelayEntry)o).executeTime);
    }
    public int getPayload() { return payload; }
}

2. 生產者-消費者服務

public class DelayService {
    private final DelayQueue queue = new DelayQueue<>();
    private final Thread consumer = new Thread(this::consumeLoop, "DelayConsumer");

    public void start() { consumer.start(); }

    public void submit(int payload, long delayMs) {
        queue.offer(new DelayEntry(payload, delayMs));
    }

    private void consumeLoop() {
        while (!Thread.interrupted()) {
            try {
                DelayEntry task = queue.take(); // 阻塞直到任務到期
                System.out.printf("執行延遲任務：payload=%d, queueSize=%d, now=%d%n",
                        task.getPayload(), queue.size(), System.currentTimeMillis());
                // 實際業務：重試發送、寫入DB、回調等
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); break;
            }
        }
    }
}