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

在java中，漏斗算法（leaky bucket algorithm）通常指流量整形或限流策略。它的核心思想是将请求视为水流，倒入一个底部有洞的漏斗中。

如果漏斗未满：请求正常流入并排队，以恒定的速率（漏孔大小）流出并被处理。
如果漏斗已满：新请求将被丢弃（溢出），从而保护下游系统。

1. 核心原理（与令牌桶的区别）

特性	漏斗算法（leaky bucket）	令牌桶算法（token bucket）
输出速率	绝对恒定，无论输入流量多大。	允许突发流量（只要有令牌，就可瞬时高速输出）。
处理方式	请求在桶内排队，以固定速率消费。	请求直接取令牌，取到即处理。
适用场景	需要平滑突发流量，对输出速率有严格要求（如数据库写入）。	允许一定突发，追求高吞吐和低延迟（如网关api限流）。

2. java 代码实现（基于队列 + 定时任务）

最可靠的实现是使用阻塞队列 + scheduledexecutorservice 模拟匀速消费。

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());
    }
}

输出效果：即使前10个请求立即占满队列，后续请求被丢弃。队列中的请求会以 500ms/个 的间隔匀速被消费。

3. 简化实现（计数器版本）—— 不推荐

如果你仅仅追求固定时间窗口内的总请求数控制，可以用 atomiclong 配合时间戳重置。但这不是严格意义上的漏斗，因为它允许窗口末期的突发流量。

// 简单计数器（固定窗口），不能平滑流量，仅做演示
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;
    }
}

4. 工业级实现推荐（guava ratelimiter）

guava 的 ratelimiter 实现的是令牌桶（允许突发），但如果你将 burst 设置为 0，它也能模拟近乎漏斗的效果。

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 {
    // 拒绝请求
}

注意：tryacquire() 默认是非阻塞的，acquire() 是阻塞的。guava 是生产环境的首选。

5. 真实应用场景详解

场景一：保护数据库写入（削峰填谷）

问题：双11大促，订单数据瞬时写入mysql，每秒5万qps，但数据库连接池最大只能承受2000 tps。
方案：在service层前加漏斗，容量设为2000，漏出速率设为2000/s。所有请求在漏斗内排队，以安全速率写入db，避免数据库连接池耗尽或cpu打满。

场景二：外部api调用（第三方限流）

问题：调用微信支付接口，对方限制每分钟最多300次。
方案：漏斗容量设为300，漏出速率设为5/s（300/60）。严格匀速调用，杜绝因突发超限被拉黑。

场景三：消息队列消费平滑

问题：kafka突然积压100万条消息，消费者若全部拉取会导致下游服务oom。
方案：消费者端用漏斗，将拉取速率限制为1000条/s，给下游留有处理喘息空间。

场景四：防止恶意刷票/点击

问题：用户投票接口，恶意脚本瞬间发起10万次请求。
方案：按用户ip维度使用漏斗，容量20，漏出速率2/s。即使瞬间刷爆，也只会处理20个，其余全部丢弃，保证投票公正性。

6. 漏斗算法的缺陷与变种

缺陷	解决方案
面对突发流量，大量请求被丢弃，体验差（如秒杀）。	改为令牌桶允许一定突发，或使用滑动窗口兼顾公平。
队列积压导致请求延迟过高（超过客户端超时）。	设置队列容量不宜过大（如容量 = 漏出速率 * 最大容忍延迟秒数）。
单机漏斗无法应对分布式集群。	使用 redis + lua 实现分布式漏斗，或采用 sentinel / hystrix 的集群限流。

7. 分布式漏斗实现（redis + lua 伪代码）

-- 漏斗 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

在java中通过 jedis 或 lettuce 调用此脚本，实现集群下的精确限流。