Java中如何用semaphore信号量实现线程指定点位同步启停

需求说明 #

现有场景固定共3个工作线程，需实现线程同步屏障逻辑：

• 所有抵达指定函数的线程先阻塞等待
• 直到全部3个线程都到达该点位后，再同步唤醒所有线程继续执行
• 需将原有基于管程的实现替换为信号量实现

原有管程实现代码 #

public void checkThreadsAreToChangeColor() {
    synchronized (lock) {
        while (waitingThreads <= 1) { // Threads are going
            waitingThreads++;
            try {
                System.out.println(this.id + " sleeping");
                lock.wait();
                return;
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        }
        lock.notifyAll(); // Last thread to change color
        waitingThreads = 0;
        limit = 99999999;
        startTimeOnType = System.currentTimeMillis();
    }
}

信号量实现方案 #

该实现方案参考pveentjer的公开实现逻辑完成适配。

核心逻辑 #

基于JDK提供的java.util.concurrent.Semaphore实现，适配3个固定线程的场景：

1. 预先定义2个信号量和1个原子计数器：
   • arrivalSemaphore：初始许可数为0，用于阻塞先到达屏障的线程
   • resetSemaphore：初始许可数为1，用于屏障重置逻辑的互斥控制，避免计数错乱
   • arrivedCount：AtomicInteger类型，初始值为0，统计已到达屏障的线程数量
2. 每个线程到达屏障后先对arrivedCount做原子自增：
   • 若自增后的值小于3，说明还未集齐所有线程，调用arrivalSemaphore.acquire()阻塞等待
   • 若自增后的值等于3，说明是最后一个到达的线程，先获取resetSemaphore许可，连续调用2次arrivalSemaphore.release()释放2个许可给前面阻塞的2个线程，随后重置arrivedCount为0，释放resetSemaphore许可，当前线程直接继续执行

示例代码 #

// 类成员变量定义
private final Semaphore arrivalSemaphore = new Semaphore(0);
private final Semaphore resetSemaphore = new Semaphore(1);
private final AtomicInteger arrivedCount = new AtomicInteger(0);

public void checkThreadsAreToChangeColor() {
    int currentCount = arrivedCount.incrementAndGet();
    if (currentCount < 3) {
        try {
            System.out.println(this.id + " sleeping");
            arrivalSemaphore.acquire();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            e.printStackTrace();
        }
    } else {
        try {
            resetSemaphore.acquire();
            // 最后一个到达的线程，唤醒所有阻塞线程
            arrivalSemaphore.release(2);
            arrivedCount.set(0);
            // 原有业务逻辑
            limit = 99999999;
            startTimeOnType = System.currentTimeMillis();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            resetSemaphore.release();
        }
    }
}

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