目錄

一、令牌桶限流機制原理

二、場景設計與目標

三、核心實現代碼（Java）

1. 完整代碼實現

四、運行效果分析

五、應用建議

---

在高吞吐數據處理場景中，如何限制數據處理速率、保護系統資源、防止下游服務過載是系統設計中重要的環節。本文將介紹一種簡單實用的限流方式 —— 基于 Google Guava 的令牌桶限流機制（Token Bucket），并通過實際代碼演示如何將其應用于數據處理任務中。

一、令牌桶限流機制原理

令牌桶算法（Token Bucket）是一種常見的流量控制算法。其基本原理如下：

• 系統按照固定速率（如每秒 5 個）往桶中放入令牌；

• 每次處理數據前，需要從桶中取出一個令牌；

• 若桶中有令牌，則允許處理數據；

• 若桶中無令牌，當前請求會被阻塞或丟棄（根據實現策略）；

• 桶容量可設定為最大突發請求數，支持短時間突發。

在 Google 的 Guava 庫中，RateLimiter 類就實現了一個基于令牌桶的限流器，它適用于：

• 接口請求限速；

• 后臺批處理速率控制；

• 防止后端系統過載等場景。

二、場景設計與目標

我們構建一個數據處理系統，包含以下三個核心組件：

1. 數據生成器線程：以較高頻率（每秒 2 條）不斷將數據放入隊列。

2. 數據處理器線程：通過 RateLimiter 每秒只允許處理 1 條數據。

3. 監控線程：每隔 2 秒打印隊列中積壓的數據數量，觀測限流效果。

三、核心實現代碼（Java）

使用 Guava 31+，Maven 依賴如下：

<dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>31.1-jre</version>
</dependency>

1. 完整代碼實現

package google;
?
import com.google.common.util.concurrent.RateLimiter;
?
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
?
public class TokenBucketDataProcessor {
?// 每秒最多處理5個數據private static final RateLimiter rateLimiter = RateLimiter.create(1);
?// 模擬數據管道private static final BlockingQueue<String> queue = new LinkedBlockingQueue<>();
?public static void main(String[] args) {
?// 線程1：數據生成線程，每秒生成2條數據Thread producer = new Thread(() -> {int i = 0;while (true) {try {Thread.sleep(500); // 每100ms生成1條數據 -> 每秒10條String data = "data-" + (i++);queue.put(data);System.out.println("[Producer] Generated: " + data);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}});
?// 線程2：數據處理線程，受RateLimiter限流影響Thread consumer = new Thread(() -> {while (true) {try {rateLimiter.acquire(1); // 阻塞直到拿到令牌String data = queue.take(); // 取數據System.out.println("[Consumer] Processed: " + data);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}});
?// 線程3：監控線程，每2秒輸出積壓情況Thread monitor = new Thread(() -> {while (true) {try {Thread.sleep(2000);System.out.println("[Monitor] Queue size: " + queue.size());} catch (InterruptedException e) {Thread.currentThread().interrupt();}}});
?producer.start();consumer.start();monitor.setDaemon(true); // 守護線程monitor.start();}
}

四、運行效果分析

預期輸出如下：

[Producer] Generated: data-0
[Consumer] Processed: data-0
[Producer] Generated: data-1
[Consumer] Processed: data-1
[Producer] Generated: data-2
[Consumer] Processed: data-2
[Monitor] Queue size: 0
[Producer] Generated: data-3
[Producer] Generated: data-4
[Consumer] Processed: data-3
[Producer] Generated: data-5
[Producer] Generated: data-6
[Consumer] Processed: data-4
[Monitor] Queue size: 2
[Producer] Generated: data-7
[Producer] Generated: data-8
[Consumer] Processed: data-5
[Producer] Generated: data-9
[Producer] Generated: data-10
[Consumer] Processed: data-6
[Monitor] Queue size: 4
[Producer] Generated: data-11
[Producer] Generated: data-12
[Consumer] Processed: data-7
[Producer] Generated: data-13
[Producer] Generated: data-14
[Consumer] Processed: data-8
[Monitor] Queue size: 6
... ...

從輸出中可以看到：

• 數據生產速度 > 數據處理速度；

• queue.size() 會逐步增長，表明數據被限流處理；

• RateLimiter.acquire() 自動阻塞了處理線程，使得處理速度不超過 2 qps；

• 限流處理過程對系統其他線程無影響，靈活、安全、無鎖。

五、應用建議

在需要處理數據流或消息流的系統中，控制處理速率是一項必要手段：

• Guava 的 RateLimiter 實現了高效、線程安全、非阻塞或阻塞可控的令牌桶限流機制；

• 利用其 acquire() 阻塞等待機制，我們可以方便地將限流邏輯嵌入處理線程中；

• 搭配 BlockingQueue 和監控線程，我們可以直觀觀察限流效果，并對系統做動態調優；

• 適用場景包括但不限于：

  • Kafka 消費處理限流

  • 多線程數據清洗任務控制速率

  • 日志寫入、數據庫插入頻率控制