Java中信号量(Semaphore):从本地到分布式

Java中信号量(Semaphore):从本地到分布式

Java中信号量(Semaphore):从本地到分布式

一、信号量是什么

信号量是一个计数器,控制同时访问某个资源的线程/进程数量。

锁(Lock):同一时刻只允许 1 个线程进入 → 互斥(二元信号量) 信号量(Semaphore):同一时刻允许 N 个线程进入 → 限流/资源池

生活中的类比:

场景信号量值含义
停车场100最多 100 辆车同时停
餐厅座位50最多 50 人同时就餐
卫生间隔间3最多 3 人同时使用
电梯载重10最多 10 人同时乘坐

注:

博客:

https://blog.csdn.net/badao_liumang_qizhi

二、核心操作

信号量只有两个基本操作:

acquire() — 获取一个许可(计数器 -1) release() — 释放一个许可(计数器 +1) 初始许可数 = 3 线程A acquire → 剩余许可: 2 线程B acquire → 剩余许可: 1 线程C acquire → 剩余许可: 0 线程D acquire → 阻塞等待(许可为0,没有空位了) ... 线程A release → 剩余许可: 1 线程D 被唤醒 → 获取许可成功,剩余许可: 0

三、JDK 中的 Semaphore

基本用法

importjava.util.concurrent.Semaphore;// 创建信号量:最多允许 3 个线程同时执行Semaphoresemaphore=newSemaphore(3);publicvoidaccessResource(){try{semaphore.acquire();// 获取许可(阻塞等待)// 临界区:最多 3 个线程同时在这里doWork();}catch(InterruptedExceptione){Thread.currentThread().interrupt();}finally{semaphore.release();// 释放许可}}

构造函数

// 非公平信号量(默认):不保证等待顺序Semaphoresemaphore=newSemaphore(3);// 公平信号量:按请求顺序获取许可(FIFO)SemaphorefairSemaphore=newSemaphore(3,true);

常用方法

// 阻塞获取 1 个许可semaphore.acquire();// 阻塞获取多个许可semaphore.acquire(2);// 一次获取 2 个// 尝试获取,获取不到立即返回 false(不阻塞)booleanacquired=semaphore.tryAcquire();// 尝试获取,最多等待指定时间booleanacquired=semaphore.tryAcquire(5,TimeUnit.SECONDS);// 释放许可semaphore.release();// 查看当前可用许可数intavailable=semaphore.availablePermits();// 获取正在等待的线程数intwaiting=semaphore.getQueueLength();

示例:数据库连接池

publicclassSimpleConnectionPool{privatefinalSemaphoresemaphore;privatefinalQueue<Connection>pool;publicSimpleConnectionPool(intmaxSize){this.semaphore=newSemaphore(maxSize);this.pool=newConcurrentLinkedQueue<>();// 预创建连接for(inti=0;i<maxSize;i++){pool.offer(createConnection());}}publicConnectiongetConnection()throwsInterruptedException{semaphore.acquire();// 获取许可(控制并发数)returnpool.poll();// 取出连接}publicvoidreleaseConnection(Connectionconn){pool.offer(conn);// 归还连接semaphore.release();// 释放许可}}

示例:接口限流(本地)

@RestControllerpublicclassOrderController{// 最多允许 10 个请求同时处理下单privatefinalSemaphoreorderSemaphore=newSemaphore(10);@PostMapping("/order/create")publicResultcreateOrder(@RequestBodyOrderDtodto){if(!orderSemaphore.tryAcquire()){returnResult.fail("系统繁忙,请稍后重试");}try{returnorderService.create(dto);}finally{orderSemaphore.release();}}}

四、信号量 vs 锁 vs 线程池

工具并发数用途区别
Lock/synchronized1互斥访问信号量(1) 的特例
SemaphoreN控制并发度不关心是哪个线程释放
线程池N控制执行线程数管理线程生命周期

关键区别:

// 锁:谁加的锁谁释放lock.lock();// ... 只能当前线程 unlocklock.unlock();// 信号量:任何线程都可以释放semaphore.acquire();// 线程A 获取// ...semaphore.release();// 线程B 也可以释放(不要求同一线程)

这个特性使得信号量适合"生产者-消费者"场景:一个线程 acquire,另一个线程 release。

五、信号量的变体

1. 二元信号量(Binary Semaphore)

Semaphoremutex=newSemaphore(1);// 许可数=1,等效于互斥锁

与 Lock 的区别:

  • Lock 有所有权(只能由持有者释放)
  • 二元信号量无所有权(任何线程可释放)

2. 计数信号量(Counting Semaphore)

Semaphorepool=newSemaphore(10);// 标准用法

3. 带超时的信号量

// 超时未获取则放弃booleanacquired=semaphore.tryAcquire(3,TimeUnit.SECONDS);if(!acquired){// 超时处理}

4. 可增减的信号量

// 动态增加许可(如动态扩容连接池)semaphore.release(5);// 增加 5 个许可// 动态减少许可semaphore.acquire(3);// 消耗 3 个许可(不释放 = 永久减少)

六、分布式信号量

为什么需要分布式信号量

JDK Semaphore 只在单个 JVM 内有效:

实例A: Semaphore(10) → 允许 10 个 实例B: Semaphore(10) → 允许 10 个 实际并发:可能 20 个同时访问(每个实例各 10 个)

分布式信号量通过 Redis 等中间件共享计数器,所有实例共享同一个许可池。

Redisson 分布式信号量

基本用法
@ResourceprivateRedissonClientredissonClient;publicvoidaccessExternalApi(){// 获取分布式信号量(所有实例共享)RSemaphoresemaphore=redissonClient.getSemaphore("semaphore:external-api");// 首次需要设置许可数(只需执行一次)semaphore.trySetPermits(10);try{// 获取许可(跨实例控制并发总数为 10)semaphore.acquire();callExternalApi();}catch(InterruptedExceptione){Thread.currentThread().interrupt();}finally{semaphore.release();}}
带超时的获取
RSemaphoresemaphore=redissonClient.getSemaphore("semaphore:db-connection");semaphore.trySetPermits(20);// 最多等待 5 秒booleanacquired=semaphore.tryAcquire(5,TimeUnit.SECONDS);if(acquired){try{queryDatabase();}finally{semaphore.release();}}else{thrownewBusinessException("系统繁忙");}
批量获取
// 一次获取 3 个许可(批量操作场景)semaphore.acquire(3);try{batchProcess();}finally{semaphore.release(3);}

Redisson 过期信号量(PermitExpirableSemaphore)

普通信号量的问题:如果获取许可后进程崩溃,许可永远不会被释放(许可泄漏)。

// 过期信号量:许可有 TTL,超时自动归还RPermitExpirableSemaphoresemaphore=redissonClient.getPermitExpirableSemaphore("semaphore:task-runner");semaphore.trySetPermits(5);// 获取许可,10秒后自动释放(返回许可ID)StringpermitId=semaphore.acquire(10,TimeUnit.SECONDS);try{runTask();// 手动提前释放semaphore.release(permitId);}catch(Exceptione){// 即使不释放,10秒后也会自动归还semaphore.release(permitId);}

对比:

类型崩溃后许可用法
RSemaphore永久丢失(需人工恢复)稳定进程
RPermitExpirableSemaphore超时自动归还不可靠进程

七、分布式信号量的 Redis 实现原理

数据结构

Key: semaphore:external-api Type: String Value: 10(当前可用许可数)

acquire 操作(Lua 脚本)

-- KEYS[1] = 信号量 key-- ARGV[1] = 要获取的许可数localpermits=tonumber(redis.call('get',KEYS[1]))ifpermits~=nilandpermits>=tonumber(ARGV[1])then-- 许可足够,扣减redis.call('decrby',KEYS[1],ARGV[1])return1end-- 许可不足return0

release 操作(Lua 脚本)

-- 归还许可redis.call('incrby',KEYS[1],ARGV[1])-- 通知等待者redis.call('publish',KEYS[2],ARGV[1])return1

等待机制

获取失败时不轮询,使用 Redis Pub/Sub 等待通知:

线程A acquire 失败 │ ├─ 订阅 Channel: redisson_sc:{semaphore:external-api} │ ├─ 阻塞等待通知 │ 线程B release → publish 消息到 Channel │ └─ 线程A 收到通知 → 再次尝试 acquire

八、实战场景

场景1:控制第三方 API 调用并发数

@ServicepublicclassThirdPartyApiService{@ResourceprivateRedissonClientredissonClient;/** * 第三方限制最多 5 个并发请求. */publicApiResponsecallThirdPartyApi(ApiRequestrequest){RSemaphoresemaphore=redissonClient.getSemaphore("semaphore:third-party-api");semaphore.trySetPermits(5);booleanacquired=false;try{acquired=semaphore.tryAcquire(10,TimeUnit.SECONDS);if(!acquired){thrownewBusinessException("第三方接口繁忙,请稍后重试");}returnhttpClient.post(request);}catch(InterruptedExceptione){Thread.currentThread().interrupt();thrownewBusinessException("操作被中断");}finally{if(acquired){semaphore.release();}}}}

场景2:分布式限流(令牌桶简化版)

@ComponentpublicclassDistributedRateLimiter{@ResourceprivateRedissonClientredissonClient;/** * 每秒最多处理 100 个请求(所有实例合计). */publicbooleantryAcquire(Stringresource){RSemaphoresemaphore=redissonClient.getSemaphore("rate:"+resource);returnsemaphore.tryAcquire();}/** * 每秒补充许可(定时任务). */@Scheduled(fixedRate=1000)publicvoidrefillPermits(){RSemaphoresemaphore=redissonClient.getSemaphore("rate:order-api");intcurrent=semaphore.availablePermits();if(current<100){semaphore.release(100-current);// 补充到 100}}}

场景3:数据库连接池保护

@ServicepublicclassDatabaseService{@ResourceprivateRedissonClientredissonClient;// 数据库最大连接 50,预留 10 给管理操作// 业务最多使用 40 个连接privatestaticfinalintMAX_BIZ_CONNECTIONS=40;public<T>TexecuteQuery(Supplier<T>query){RSemaphoresemaphore=redissonClient.getSemaphore("semaphore:db-biz-conn");semaphore.trySetPermits(MAX_BIZ_CONNECTIONS);try{semaphore.acquire();returnquery.get();}catch(InterruptedExceptione){Thread.currentThread().interrupt();thrownewRuntimeException(e);}finally{semaphore.release();}}}

场景4:并行任务控制

/** * 导出报表:允许系统同时最多处理 3 个导出任务(防止 OOM). */@ServicepublicclassReportExportService{@ResourceprivateRedissonClientredissonClient;publicvoidexportReport(IntegerreportId){RPermitExpirableSemaphoresemaphore=redissonClient.getPermitExpirableSemaphore("semaphore:report-export");semaphore.trySetPermits(3);StringpermitId=null;try{// 获取许可,最多持有 5 分钟(防止任务卡死占用许可)permitId=semaphore.tryAcquire(30,300,TimeUnit.SECONDS);if(permitId==null){thrownewBusinessException("当前导出任务过多,请稍后再试");}doExport(reportId);}finally{if(permitId!=null){semaphore.release(permitId);}}}}

九、信号量 vs 其他并发控制工具

工具控制维度适用场景
Semaphore并发数量控制同时执行的操作数
RateLimiter速率(每秒N个)控制请求频率
Lock互斥(0或1)独占资源
CountDownLatch等待计数归零等待多个任务完成
CyclicBarrier等待N个线程到达多线程同步汇合
线程池工作线程数管理执行资源

信号量 vs 线程池

// 线程池方式:控制执行线程数ExecutorServicepool=Executors.newFixedThreadPool(10);pool.submit(()->callApi());// 超过 10 个则排队// 信号量方式:控制并发数(不管你用什么线程)Semaphoresem=newSemaphore(10);sem.acquire();try{callApi();// 可以在任何线程中执行}finally{sem.release();}

区别:

  • 线程池管理线程的创建和销毁
  • 信号量只管"允许多少个同时执行",不管线程来自哪里

两者常配合使用:线程池控制线程总量,信号量控制某类操作的并发量。

信号量 vs RateLimiter

// 信号量:同一时刻最多 10 个并发// 如果每个请求处理 1 秒,则吞吐约 10/sSemaphoresem=newSemaphore(10);// RateLimiter:每秒最多 10 个请求// 不管并发多少,严格控制速率RateLimiterlimiter=RateLimiter.create(10.0);limiter.acquire();// 平滑限流
信号量RateLimiter
控制的是同时进行的数量单位时间的数量
短时间突发允许(并发数内)平滑(不允许突发)
请求处理时间影响处理越慢,吞吐越低不受处理时间影响

十、注意事项与陷阱

陷阱1:许可泄漏

// 错误:异常时不释放许可semaphore.acquire();doWork();// 如果这里抛异常semaphore.release();// 这行不会执行 → 许可永久丢失// 正确:finally 中释放semaphore.acquire();try{doWork();}finally{semaphore.release();}

陷阱2:释放多于获取

Semaphoresem=newSemaphore(3);// 没有 acquire 就 release → 许可数变成 4!sem.release();// 现在许可数 = 4,超过了设计的 3

信号量不会校验"是否之前获取过",多余的 release 会增加许可总数。

陷阱3:分布式环境下的初始化竞争

// 多个实例同时启动,都执行 trySetPermitssemaphore.trySetPermits(10);// 实例Asemaphore.trySetPermits(10);// 实例B(如果 key 已存在则不生效)

trySetPermits是"不存在才设置"(类似 SETNX),所以多实例并发调用是安全的。但如果要修改许可数,需要用addPermits

// 扩容:增加 5 个许可semaphore.addPermits(5);// 缩容:减少 3 个许可(当前可用 >= 3 才能成功)semaphore.addPermits(-3);

陷阱4:try-finally 中的 acquire 返回值

// 错误:tryAcquire 返回 false 但 finally 仍然 releasebooleanacquired=semaphore.tryAcquire();try{if(acquired)doWork();}finally{semaphore.release();// acquired=false 时多释放了!}// 正确:条件释放booleanacquired=semaphore.tryAcquire();try{if(!acquired){thrownewBusinessException("繁忙");}doWork();}finally{if(acquired){semaphore.release();}}

十一、总结

概念一句话
信号量一个计数器,控制"同时有多少个"能执行
acquire获取许可(计数-1),许可为0时阻塞
release释放许可(计数+1),唤醒等待者
与锁的区别锁是二元的(0或1),信号量是N元的
分布式信号量用 Redis 存储计数,所有实例共享许可池
过期信号量许可有 TTL,进程崩溃后自动归还
核心价值保护有限资源不被过度并发访问
常见用途API 并发控制、连接池保护、任务并行度限制