与信号量同步的“阶段”序列无法正常工作

Question

我正在学习线程和同步，我正在尝试建立一个“阶段”系统，其中第一阶段生成一条消息（为方便起见是一个 int），然后将其传递到阶段 2，该阶段修改它（乘以 2），然后将其传递到最后一个阶段，该阶段进一步修改它并将其打印到控制台。 问题是第三阶段永远无法运行，即使它收到了消息。

在我的示例中，我设置了一个名为“resources”的信号量数组，其中信号量 A（索引 0）有 4 个许可，信号量 B 有 3 个，C 有 2 个。所有设置都是公平的。

我尝试将信号量设置为公平，但这并没有解决我的问题。我也尝试过更改睡眠时间，但没有成功。

class Fase1 extends Thread{
    private int i = 0;
    private Semaphore[] resources;
    private Fase2 recipient;

    public Fase1(Semaphore[] res, Fase2 fase2){
        recipient=fase2;
        resources=res;
    }

    @Override
    public void run(){
        try{
            while(true){
                resources[0].acquire(2);
                resources[1].acquire(2);
                recipient.receiveMessage(i);
                i++;
                sleep(200);
                resources[1].release(2);
                resources[0].release(2);
            }
        } catch (InterruptedException e){
        }
    }
}

class Fase2 extends Thread{
    private Semaphore[] resources;
    private Fase3 recipient;
    private boolean receivedMessage = false;
    private int message = 0;

    public Fase2(Semaphore[] res, Fase3 fase3){
        recipient=fase3;
        resources=res;
    }

    @Override
    public void run(){
        try{
            while(true){
                if(receivedMessage){
                    resources[0].acquire(2);
                    resources[1].acquire(2);
                    resources[2].acquire(2);
                    recipient.receiveMessage(message*2);
                    receivedMessage = false;
                    sleep(200);
                    resources[2].release(2);
                    resources[1].release(2);
                    resources[0].release(2);
                }
            }
        } catch (InterruptedException e){
        }
    }
    public void receiveMessage(int msg){
        message = msg;
        receivedMessage = true;
    }
}

class Fase3 extends Thread{
    private Semaphore[] resources;
    private boolean receivedMessage = false;
    private int message = 0;

    public Fase3(Semaphore[] res){
        resources=res;
    }

    @Override
    public void run(){
        try{
            while(true){
                if(receivedMessage){
                    resources[1].acquire(2);
                    resources[2].acquire(2);
                    System.out.println(message+1);
                    receivedMessage = false;
                    sleep(200);
                    resources[2].release(2);
                    resources[1].release(2);
                }
            }
        } catch (InterruptedException e){
        }
    }
    public void receiveMessage(int msg){
        message = msg;
        receivedMessage = true;
    }
}

我注意到许可证以某种方式被搞砸了，几乎就像一些线程没有正确释放它们一样，即使在我看来这是正确的。

Answer 1

您的设计存在一个根本缺陷：您正在同步对message 资源的访问，而不是对receivedMessage 标志的访问。当您在线程#2 中将标志设置为true 时，JVM 没有义务将该写入传播到线程#3，因为该线程在它位于if 块内之前不会执行同步，这很可能永远不会发生。线程#1 和#2 之间的通信也是如此。

acquire() 和release() 操作act as synchronization points 将使写入跨线程可见，因此您需要在检查标志之前调用它们。比如Fase3：

    @Override
    public void run(){
        try{
            while(true){
                resources[1].acquire(2); // All writes by thread #2 are now visible
                if(receivedMessage){
                    resources[2].acquire(2);
                    System.out.println(message+1);
                    sleep(200);
                    receivedMessage = false;
                    resources[2].release(2);
                }
                resources[1].release(2);
            }
        } catch (InterruptedException e){
        }
    }

还有另一种解决方案，就是将receivedMessage 标记为volatile，但正确使用单一锁定机制会更简洁。

附带说明，最好是use Runnables instead of extending Thread。