Semaphore与Executor Service锁机制解析及可用许可异常疑问

问题描述 #

我正在学习并行处理中的Semaphore机制，实例化了一个拥有3个许可的Semaphore，同时准备了包含5个元素的字符串列表，通过Executor Service创建了5个基于Callable接口的线程。

按照我的理解，Semaphore应该允许3个线程同时执行，当任意线程完成后，下一个线程可获取许可继续执行。但执行代码后，输出显示最终可用许可为2，执行过程中可用许可依次为0、1、0、1、2。我无法通过输出理解其内部运行逻辑，希望有人解释其中原理，以及为何可用许可未恢复为3。同时请推荐一些Semaphore常用的API。

代码示例 #

public class SemaphoreDemo {
    
    Semaphore semaphore= new Semaphore(3);

    public void doit() {
        List<String> ll =List.of("1","2","3","4","5");
        
        List<Callable<String>> parallelprocess= new ArrayList<Callable<String>>(ll.size());
        
        for(String data:ll) {
        
            parallelprocess.add(() -> {
                semaphore.acquire();
                String res= data+Thread.currentThread().getName();
                Thread.sleep(1000);
                System.out.println(res);
                System.out.println(semaphore.availablePermits());
                semaphore.release();
                return res;
            });
            
        }
        
        ExecutorService executor=Executors.newFixedThreadPool(parallelprocess.size());
        
        try {
            List<Future<String>> reponse=executor.invokeAll(parallelprocess);
            
            for(Future<String> future:reponse) {
                
                try {
                    System.out.println(future.get().toString());
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        executor.shutdown();
        
    }
    
    public static void main(String[] args) {
        SemaphoreDemo ss=new SemaphoreDemo();
        
        ss.doit();
    }
}

执行输出 #

3pool-1-thread-3
0
2pool-1-thread-2
1
1pool-1-thread-1
0
4pool-1-thread-4
1
5pool-1-thread-5
2
1pool-1-thread-1
2pool-1-thread-2
3pool-1-thread-3
4pool-1-thread-4
5pool-1-thread-5

逻辑解析 #

核心执行流程 #

1. 初始化后Semaphore有3个许可，ExecutorService启动5个线程，前3个线程（线程1、2、3）会立即抢占许可，此时可用许可变为0。
2. 这3个线程进入Thread.sleep(1000)，等待1秒后开始执行后续代码：
   • 线程3率先完成sleep：此时它仍持有许可，线程1、2也在持有许可，所以可用许可为0，输出0；随后调用release()释放许可，可用许可变为1，线程4立刻抢占该许可。
   • 线程2完成sleep：此时线程3已释放许可，线程4的抢占操作可能还未完成，当前只有线程1和线程2持有许可，可用许可为3-2=1，输出1；随后线程2释放许可，可用许可变为2，线程5立刻抢占一个许可，可用许可变为1。
   • 线程1完成sleep：此时线程4、5已各抢占一个许可，当前持有许可的是线程1、4、5，可用许可为3-3=0，输出0；线程1释放许可后，可用许可变为1。
   • 线程4完成sleep：此时线程1已释放许可，当前持有许可的是线程4、5，可用许可为3-2=1，输出1；线程4释放许可后，可用许可变为2。
   • 线程5完成sleep：此时线程4已释放许可，只有它自己持有许可，可用许可为3-1=2，输出2；线程5释放许可后，可用许可恢复为初始的3。

为何看起来许可未恢复为3 #

你看到的最终输出2是线程5调用release()前的可用许可数，线程5执行完release()后，Semaphore的许可会完全恢复为3，只是代码中没有输出这个最终状态。

Semaphore常用API推荐 #

• acquire()：获取1个许可，若无可用许可则阻塞等待。
• acquire(int permits)：一次性获取指定数量的许可，阻塞等待直到获取成功。
• release()：释放1个许可，归还到Semaphore中。
• release(int permits)：一次性释放指定数量的许可。
• availablePermits()：返回当前剩余的可用许可数。
• tryAcquire()：尝试获取1个许可，立即返回结果（成功返回true，失败返回false），不会阻塞。
• tryAcquire(long timeout, TimeUnit unit)：在指定时间内尝试获取许可，超时仍未获取则返回false。
• drainPermits()：一次性获取所有可用的许可，并返回获取到的许可数量。

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