松林AI说头像
Java 并发工具 Semaphore 源码剖析:信号量在高并发限流场景中的使用陷阱封面图

Java 并发工具 Semaphore 源码剖析:信号量在高并发限流场景中的使用陷阱

封面信息图

在微服务架构或网关层的高并发场景中,保护下游弱依赖(如遗留数据库连接池、外部风控接口、高耗时计算服务)最常用的并发工具之一就是 java.util.concurrent.Semaphore。许多开发者认为信号量就是一把“容量可配的计数器”,只要在调用前 acquire(),调用后 release() 就能高枕无忧。

然而在真实的生产压测与大促流量洪峰中,看似轻量级的 Semaphore 却频频引发严重的事故:服务吞吐断崖式下跌、系统线程池被打满耗尽、许可证无限“膨胀”导致下游直接被压垮。深入剖析其底层 AQS 机制,才能认清它在限流场景中的适用边界与致命陷阱。


源码基石:基于 AQS 共享模式的状态流转

Semaphore 的核心依赖于 AbstractQueuedSynchronizer(AQS)的共享模式(Shared Mode)。与独占锁 ReentrantLock 仅允许单一线程持有 state(标记持有次数)不同,Semaphore 将 AQS 的核心同步状态 state 抽象为“当前可用的许可证数量”。

1. 非公平模式与公平模式获取许可

在默认的非公平模式(NonfairSync)下,线程调用 acquire() 最终触发的是 nonfairTryAcquireShared:

// java.util.concurrent.Semaphore.NonfairSync
final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

这里是一个标准的自旋 CAS 操作:

  • 如果 remaining < 0,说明当前许可不足,直接返回负数,外层 AQS 会将当前线程包装为 Node.SHARED 节点加入 CLH 同步等待队列,并调用 LockSupport.park() 挂起。
  • 如果 remaining >= 0,则尝试通过 CAS 将 state 修改为 remaining。只要 CAS 成功,当前线程即成功夺得许可。

而在公平模式(FairSync)下,仅多了一道检查:

// java.util.concurrent.Semaphore.FairSync
protected int tryAcquireShared(int acquires) {
    for (;;) {
        if (hasQueuedPredecessors())
            return -1;
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

hasQueuedPredecessors() 保证了只要队列中有等待更久的排队线程,后来的线程即使发现当前系统有剩余凭证,也必须进入队列尾部排队,严守先来后到原则。

2. 共享唤醒的向后传播:setHeadAndPropagate

当某个持有许可的线程调用 release() 时,AQS 的 tryReleaseShared 会通过 CAS 将 state 加回去。一旦释放成功,AQS 会触发 doReleaseShared。

在共享模式下,被唤醒的头节点线程在成功夺得许可脱离阻塞后,会调用 setHeadAndPropagate。如果此时发现仍然有剩余许可(propagate > 0),或者等待队列后继节点也是共享模式,它会顺次唤醒后继节点的线程,形成“级联唤醒(Propagation)”。


生产环境四大血泪使用陷阱

陷阱一:许可证凭空虚增(Permit Inflation)

这是生产中最容易因为粗心而引发的重大事故。很多开发者习惯写出如下代码:

// 错误示范:致命的许可证泄漏与膨胀
public void executeBad() {
    try {
        semaphore.acquire();
        doBusiness();
    } finally {
        semaphore.release();
    }
}

这段代码看似严谨地使用了 finally,但潜藏着致命漏洞:
如果当前线程在执行 semaphore.acquire() 时遭遇了线程中断(InterruptedException),acquire() 抛出异常,此时当前线程根本没有成功获得任何许可证!然而执行流直接跌入 finally 块,执行了 semaphore.release()。

翻阅 Semaphore 的 release() 实现可以发现:

protected final boolean tryReleaseShared(int releases) {
    for (;;) {
        int current = getState();
        int next = current + releases;
        if (next < current) // 溢出检测
            throw new Error("Maximum permit count exceeded");
        if (compareAndSetState(current, next))
            return true;
    }
}

Semaphore 从未记录“这个许可是谁借走的”,也没有对许可上限做与初始值相同的约束校验!
每次多余的 release() 都会直接使 state 加 1。在大规模服务异常或者线程池中断取消的高频冲击下,信号量的可用许可从原本限制的 50 个一路膨胀到几千个,下游数据库瞬间被洪峰冲垮,限流彻底失效。

陷阱二:公平信号量导致的严重惊群与上下文切换风暴

很多团队为了追求请求的绝对顺序与防饥饿,将信号量初始化为公平模式:new Semaphore(100, true)。

在高并发压测(如 5 万 QPS)中,CPU 的 sy(系统内核态态开销)急剧飙升至 80% 以上,而 us(用户态业务开销)不足 15%。这是因为公平信号量严格遵循 CLH 队列的节点顺序:
每当一个线程释放许可,队列头节点被唤醒并重新竞争;如果队列中有大量因为超时尝试退出的取消节点,AQS 内部需要不断遍历调整前驱和后继指针,频繁引发线程的 park 与 unpark。大量的上下文切换把 CPU 时钟周期全部消耗在系统调用上。

陷阱三:非受控的无界排队与级联超时

直接使用阻塞的 acquire() 意味着无限制等待。在依赖服务响应变慢(如网络抖动从 5ms 增加到 2s)时,前端进来的请求源源不断涌入,全部阻塞在 semaphore.acquire()。

这会导致容器线程池(如 Tomcat、Undertow、Netty EventLoop)工作线程迅速耗尽,前端客户端超时断开,但后端仍在徒劳等待获取许可,最终引发服务雪崩。必须采用带超时控制的 tryAcquire(timeout, unit)。

陷阱四:将 Semaphore 误当作漏桶或平滑限流器

Semaphore 只能限制瞬时并发数(Concurrency Limit),不能限制速率(Rate Limit)。
如果初始化 Semaphore(1000),理论上允许 1000 个请求在 1 毫秒内全部涌入系统。如果下游承受不住瞬时突发,这种“并发限流”无法起到削峰填谷的作用。需要严格区分速率限制(如 Guava RateLimiter 的令牌桶)与并发限制(Semaphore 的资源隔离)。


工业级健壮封装实现

针对上述陷阱,以下给出一套生产标准的信号量限流防护模式,包含防虚增保护、纳秒级超时、指标打点与快速失败。

package com.jinxin.concurrent.limiter;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 健壮的并发限流守卫
 */
public class RobustConcurrencyLimiter {

    private final Semaphore semaphore;
    private final int maxPermits;
    private final AtomicLong rejectedCount = new AtomicLong(0);

    public RobustConcurrencyLimiter(int maxPermits) {
        this.maxPermits = maxPermits;
        // 生产环境限流压测务必优先选择非公平模式以换取极限吞吐
        this.semaphore = new Semaphore(maxPermits, false);
    }

    public <T> T execute(long timeoutMs, BusinessTask<T> task) throws Exception {
        boolean acquired = false;
        try {
            // 采用带超时控制的非阻塞/限时尝试
            acquired = semaphore.tryAcquire(timeoutMs, TimeUnit.MILLISECONDS);
            if (!acquired) {
                rejectedCount.incrementAndGet();
                throw new RateLimitExceededException("Concurrency limit exceeded. Threshold: " + maxPermits);
            }

            // 成功夺取凭据后,执行真实业务逻辑
            return task.run();

        } finally {
            // 严禁未成功 acquire 就执行 release,防止 Permit 虚增
            if (acquired) {
                semaphore.release();
            }
        }
    }

    public int getAvailablePermits() {
        return semaphore.availablePermits();
    }

    public long getRejectedCount() {
        return rejectedCount.get();
    }

    @FunctionalInterface
    public interface BusinessTask<T> {
        T run() throws Exception;
    }

    public static class RateLimitExceededException extends RuntimeException {
        public RateLimitExceededException(String message) {
            super(message);
        }
    }
}

思考总结

理解 Semaphore,核心在于把握两点:

  1. 它是一个无状态归属的计数器,release() 必须与成功成功的 acquire() 严格配对,绝不能让中断异常打破平衡。
  2. 它是并发数的硬闸门,而非流量整形工具。在复杂分布式微服务治理中,应当将 Semaphore 用于进程内关键资源的物理隔离(如特定线程池、有限内存块),而将网关削峰留给令牌桶或分布式限流中间件。

转载自 CSDN-专业IT技术社区

原文链接:https://blog.csdn.net/cannonjinx/article/details/168250140

文章来源转载

评论

评论列表

微信小程序
QQ小程序

关于作者

点赞数:0
关注数:0
粉丝:0
文章:0
关注标签:0
加入于:--