详解Java 漏斗算法及应用场景

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class LeakyBucketLimiter {

    // 漏斗容量（队列最大大小）
    private final int capacity;
    // 漏出速率（单位：毫秒/个）
    private final long leakIntervalMs;
    // 阻塞队列存放请求ID或任务
    private final BlockingQueue<String> queue;
    // 记录被丢弃的请求数（监控用）
    private final AtomicInteger discardedCount = new AtomicInteger(0);

    public LeakyBucketLimiter(int capacity, int leakRatePerSecond) {
        this.capacity = capacity;
        this.leakIntervalMs = 1000 / leakRatePerSecond;
        this.queue = new LinkedBlockingQueue<>(capacity);
        // 启动后台线程以固定速率消费队列
        startLeaking();
    }

    // 后台漏出线程
    private void startLeaking() {
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(() -> {
            String request = queue.poll();
            if (request != null) {
                // 模拟处理请求（调用真实业务逻辑）
                System.out.println("Processing: " + request + " at " + System.currentTimeMillis());
            }
        }, 0, leakIntervalMs, TimeUnit.MILLISECONDS);
    }

    // 尝试放入请求（外部调用）
    public boolean tryAcquire(String requestId) {
        boolean offered = queue.offer(requestId); // 非阻塞，队列满则返回false
        if (!offered) {
            discardedCount.incrementAndGet();
            System.out.println("Request " + requestId + " discarded. Queue full.");
        }
        return offered;
    }

    // 获取当前丢弃总数
    public int getDiscardedCount() {
        return discardedCount.get();
    }

    // 测试示例
    public static void main(String[] args) throws InterruptedException {
        LeakyBucketLimiter limiter = new LeakyBucketLimiter(10, 2); // 容量10，每秒漏出2个

        // 模拟瞬间20个请求涌入
        for (int i = 0; i < 20; i++) {
            String reqId = "Req-" + i;
            limiter.tryAcquire(reqId);
            Thread.sleep(50); // 间隔50ms发送
        }

        Thread.sleep(5000); // 等待处理完成
        System.out.println("Total discarded: " + limiter.getDiscardedCount());
    }
}

// 简单计数器（固定窗口），不能平滑流量，仅做演示
public class SimpleCounterLimiter {
    private final int maxRequestsPerSecond;
    private AtomicLong counter = new AtomicLong(0);
    private volatile long lastResetTime = System.currentTimeMillis();

    public boolean tryAcquire() {
        long now = System.currentTimeMillis();
        if (now - lastResetTime > 1000) {
            counter.set(0);
            lastResetTime = now;
        }
        return counter.incrementAndGet() <= maxRequestsPerSecond;
    }
}

import com.google.common.util.concurrent.RateLimiter;

// 每秒发出 2 个许可（稳定速率）
RateLimiter limiter = RateLimiter.create(2.0); 
// 预热期平滑（可选）
// RateLimiter.create(2.0, 1, TimeUnit.SECONDS);

if (limiter.tryAcquire()) {
    // 处理请求
} else {
    // 拒绝请求
}

-- 漏斗 key: user_id, capacity: 100, leak_rate: 10/s
local key = KEYS[1]
local now = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local leak_rate = tonumber(ARGV[3])

-- 获取上次访问时间和剩余水量
local last_time = redis.call('hget', key, 'last_time') or now
local water = redis.call('hget', key, 'water') or 0

-- 计算漏掉的水量
local leak = (now - last_time) * leak_rate / 1000
water = math.max(0, water - leak)

-- 判断容量
if water < capacity then
    redis.call('hset', key, 'water', water + 1)
    redis.call('hset', key, 'last_time', now)
    return 1  -- 允许
else
    return 0  -- 拒绝
end