如何在Spring Boot中实现带等待重试的用户线程数限流？

我有一个处理不同用户请求的Spring Boot应用，部分请求耗时较长最终会超时。若某用户发起大量此类请求，服务器可能耗尽可用HTTP线程，进而影响所有用户。为避免该情况并保障其他用户正常使用，我需要一个限流组件来防止线程耗尽：当某用户占用的线程数达到N时，新请求需触发限制。

我当前通过Filter类结合内部用户-线程数映射实现限流，达到限制后会直接拒绝请求并返回HTTP 429错误，但该方案无法应对偶尔的请求峰值。理想情况是Filter等待数秒并重试，若仍无空闲线程再抛出错误。请问是否存在无需复杂Java并发代码的简单实现方式？

当前实现代码如下：

public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {

    String userName = null;
    Integer counter = 0;
    try {
        HttpServletRequest httpRequest = (HttpServletRequest)request;
        userName = getUsername(httpRequest);

        boolean ok;
        synchronized (lock) {
            counter = counters.get(userName);
            if (counter==null) {
                counter = new Integer(0);
            }
            counter++;
            counters.put(userName, counter);
            ok = counter <= limit;
        }
        if (ok) {
            chain.doFilter(request, response);
        } else {
            throw new RuntimeException("Too many calls");
        }
    } finally {
        synchronized (lock) {
            counter--;
            counters.put(userName,counter);
        }
    }
}

可以利用Java自带的Semaphore（信号量）简化并发控制，同时实现等待重试逻辑，无需手动维护计数和复杂同步代码。核心思路是为每个用户分配固定许可数的Semaphore，新请求尝试获取许可时指定超时时间，超时未获取则返回429。

具体实现步骤 #

1. 用ConcurrentHashMap存储每个用户对应的Semaphore，保证线程安全且无需手动加锁
2. Filter中获取当前用户的Semaphore，不存在则自动创建（许可数设为你需要的N）
3. 调用semaphore.tryAcquire(long timeout, TimeUnit unit)尝试获取许可，指定等待时长
4. 成功获取则继续执行请求，完成后释放许可；超时未获取则返回429错误

优化后的Filter代码 #

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import javax.servlet.Filter;
import javax.servlet.FilterChain;
import javax.servlet.FilterConfig;
import javax.servlet.ServletException;
import javax.servlet.ServletRequest;
import javax.servlet.ServletResponse;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

public class UserThreadLimitFilter implements Filter {

    // 每个用户允许同时占用的线程数
    private int userThreadLimit = 5;
    // 请求等待超时时间（秒）
    private int waitTimeout = 3;
    // 存储用户与对应信号量的映射
    private final ConcurrentHashMap<String, Semaphore> userSemaphores = new ConcurrentHashMap<>();

    @Override
    public void init(FilterConfig filterConfig) throws ServletException {
        // 可从配置文件读取参数
        String limitParam = filterConfig.getInitParameter("userThreadLimit");
        if (limitParam != null) {
            userThreadLimit = Integer.parseInt(limitParam);
        }
        String timeoutParam = filterConfig.getInitParameter("waitTimeout");
        if (timeoutParam != null) {
            waitTimeout = Integer.parseInt(timeoutParam);
        }
    }

    @Override
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws ServletException {
        HttpServletRequest httpRequest = (HttpServletRequest) request;
        HttpServletResponse httpResponse = (HttpServletResponse) response;
        String userName = getUsername(httpRequest);

        // 获取或创建用户对应的Semaphore
        Semaphore semaphore = userSemaphores.computeIfAbsent(userName, k -> new Semaphore(userThreadLimit));

        boolean acquired = false;
        try {
            // 尝试获取许可，等待指定时长
            acquired = semaphore.tryAcquire(waitTimeout, TimeUnit.SECONDS);
            if (acquired) {
                chain.doFilter(request, response);
            } else {
                // 超时未获取到许可，返回429
                httpResponse.setStatus(HttpServletResponse.SC_TOO_MANY_REQUESTS);
                httpResponse.getWriter().write("Too many concurrent requests, please try again later.");
            }
        } catch (InterruptedException e) {
            // 等待被中断，恢复中断状态并返回500
            Thread.currentThread().interrupt();
            httpResponse.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
        } finally {
            // 只有成功获取到许可才需要释放
            if (acquired) {
                semaphore.release();
            }
        }
    }

    private String getUsername(HttpServletRequest request) {
        // 此处实现获取当前用户名的逻辑，比如从请求头、Session或Token中解析
        return request.getHeader("X-User-Name");
    }

    @Override
    public void destroy() {
        // 销毁时清空资源
        userSemaphores.clear();
    }
}

方案优势 #

• 无需手动计数：Semaphore自动维护许可数量，避免原代码中Integer包装类的线程安全问题（原代码counter++实际创建新Integer对象，并发场景下可能计数错误）
• 内置等待超时：tryAcquire直接支持超时等待，无需自己实现重试逻辑
• 线程安全：ConcurrentHashMap和Semaphore都是Java并发包提供的线程安全组件，无需手动加synchronized块，性能更优
• 可配置：通过Filter初始化参数可灵活调整每个用户的线程限制和等待超时时间

注意事项 #

• 若用户量极大，userSemaphores可能占用较多内存，可添加过期清理逻辑（比如定时清理长时间未使用的Semaphore）
• getUsername方法需确保正确识别用户，避免不同用户误判为同一用户导致限流错误

内容的提问来源于stack exchange，提问作者Ilia Stepanov