Java中如何通过Semaphore实现同方向列车线程并发、反方向互斥执行

原有代码的问题 #

原实现使用单个许可为1的信号量，所有方向列车都竞争同一个通行权限，导致同方向列车也无法并发通行，不符合“同方向可并行、反方向互斥”的需求。

核心设计思路 #

完全基于Semaphore实现同步，通过共享状态计数 + 方向等待信号量的组合，实现规则：

• 用1许可的公平信号量做互斥锁，保护所有共享状态的修改，避免线程安全问题
• 记录两个方向的在途列车数，同方向列车直接累加计数即可进入
• 反方向有在途列车时，当前方向列车进入对应等待队列阻塞
• 当前方向所有列车驶离（计数归0）时，批量唤醒反方向所有等待的列车

备注：当前实现为同方向优先的调度规则，若需要避免反方向列车饥饿（比如持续有同方向列车到达导致反方向永远等待），可在进入逻辑的循环判断中增加反方向等待计数的检查，即当反方向有等待列车时，新到达的同方向列车也需进入等待，实现先来先服务的公平调度。

改造后代码实现 #

共享状态与线程类 #

class MyThread extends Thread {
    // 保护所有共享状态的互斥信号量，公平模式避免饥饿
    private static final Semaphore stateLock = new Semaphore(1, true);
    // 两个方向当前在铁路上的列车计数
    private static int nsCount = 0; // nord-sud方向在途数
    private static int snCount = 0; // sud-nord方向在途数
    // 等待进入的方向信号量（初始无许可）与等待列车计数
    private static final Semaphore nsWait = new Semaphore(0, true);
    private static int nsWaitCount = 0;
    private static final Semaphore snWait = new Semaphore(0, true);
    private static int snWaitCount = 0;

    // 原有测试用共享计数
    static class Shared {
        static int count = 0;
    }

    String threadName;

    public MyThread(String threadName) {
        super(threadName);
        this.threadName = threadName;
    }

    @Override
    public void run() {
        try {
            System.out.println("Starting " + threadName);
            System.out.println(threadName + " 等待进入铁路");
            boolean isNs = threadName.equals("nord-sud");

            // 执行进入铁路的校验逻辑
            enterRailway(isNs);

            System.out.println(threadName + " 已获得铁路通行权");
            // 原有铁路通行逻辑保持不变
            for (int i = 0; i < 5; i++) {
                Shared.count = isNs ? Shared.count + 1 : Shared.count - 1;
                System.out.println(threadName + ": " + i + " | 当前共享计数: " + Shared.count);
                Thread.sleep(1000);
            }

            // 执行离开铁路的释放逻辑
            leaveRailway(isNs);
            System.out.println(threadName + " 已离开铁路");

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 进入铁路的校验逻辑
     */
    private void enterRailway(boolean isNs) throws InterruptedException {
        stateLock.acquire();
        try {
            // 循环检查反方向是否有在途列车，避免误唤醒
            while (isNs ? snCount > 0 : nsCount > 0) {
                // 反方向有列车，当前方向进入等待队列
                if (isNs) {
                    nsWaitCount++;
                } else {
                    snWaitCount++;
                }
                // 释放状态锁，允许其他线程修改状态
                stateLock.release();
                // 阻塞等待被唤醒
                if (isNs) {
                    nsWait.acquire();
                } else {
                    snWait.acquire();
                }
                // 被唤醒后重新获取状态锁，再次检查反方向状态
                stateLock.acquire();
                // 等待计数减1
                if (isNs) {
                    nsWaitCount--;
                } else {
                    snWaitCount--;
                }
            }
            // 无反方向列车，当前方向在途计数加1
            if (isNs) {
                nsCount++;
            } else {
                snCount++;
            }
        } finally {
            stateLock.release();
        }
    }

    /**
     * 离开铁路的释放逻辑
     */
    private void leaveRailway(boolean isNs) throws InterruptedException {
        stateLock.acquire();
        try {
            // 当前方向在途计数减1
            if (isNs) {
                nsCount--;
            } else {
                snCount--;
            }
            // 当前方向无在途列车时，批量唤醒反方向所有等待的列车
            if (isNs ? nsCount == 0 : snCount == 0) {
                if (isNs) {
                    // nord-sud方向清空，唤醒所有sud-nord方向等待的列车
                    for (int i = 0; i < snWaitCount; i++) {
                        snWait.release();
                    }
                } else {
                    // sud-nord方向清空，唤醒所有nord-sud方向等待的列车
                    for (int i = 0; i < nsWaitCount; i++) {
                        nsWait.release();
                    }
                }
            }
        } finally {
            stateLock.release();
        }
    }
}

测试主类 #

验证同方向并行、反方向互斥的规则：

public class RailwayTest {
    public static void main(String[] args) throws InterruptedException {
        // 构造两列同方向、一列反方向的测试线程
        MyThread nsTrain1 = new MyThread("nord-sud");
        MyThread nsTrain2 = new MyThread("nord-sud");
        MyThread snTrain1 = new MyThread("sud-nord");

        nsTrain1.start();
        nsTrain2.start();
        // 延迟100ms确保前两列ns方向列车已经进入铁路
        Thread.sleep(100);
        snTrain1.start();

        nsTrain1.join();
        nsTrain2.join();
        snTrain1.join();
        System.out.println("所有列车通行完毕");
    }
}

规则验证逻辑 #

当测试运行时：

• nsTrain1和nsTrain2启动时，反方向snCount为0，直接累加nsCount后并行进入铁路
• snTrain1启动时，检测到nsCount=2>0，进入等待队列阻塞
• 两列ns方向列车全部驶离后，nsCount归0，批量唤醒所有等待的sn方向列车，snTrain1才开始通行

内容的提问来源于stack exchange，提问作者Seif Khédija