java中的简单屏障同步

Posted 2023-03-17

【中文标题】java中的简单屏障同步

【英文标题】：simple barrier synchronisation in java

【发布时间】：2017-10-19 00:34:14

【问题描述】：

我试图理解障碍问题。我对编程仍然很陌生，但在课堂上遇到了这个问题要解决。

“我必须使用计数信号量来解决屏障问题。你可以假设有一个共享变量 N 表示系统中的并发线程数。当前 N -1 个线程到达屏障时，它们应该阻塞直到第 N 个线程到达，此时所有线程都可能继续。

共享计数器变量可用于跟踪已到达的线程数，信号量互斥锁和屏障可用于解决同步问题。”

import java.util.concurrent.Semaphore;
public class BarrierSynchronization extends Thread 

int N;
int count;
Semaphore mutex;
Semaphore barrier;

public BarrierSynchronization ()

    this.N = 5;
    this.count = 0;
    this.mutex = new Semaphore(1);
    this.barrier = new Semaphore(0);


public void run()

    try 
        mutex.acquire();  
        count = count + 1;
        System.out.println(Thread.currentThread().getName() + ": " + count);
        mutex.release();

        if (count == N)
        
            barrier.release();
            System.out.println("All " + count + " threads have reached the barrier. The barrier is now open" );
         // unblock one thread

            barrier.acquire();
            barrier.release();  
            System.out.println(Thread.currentThread().getName() + " has passed the barrier");


     catch (InterruptedException e) 
        e.printStackTrace();
    
   

我尝试实现信号量小书中的伪代码。我在主类中调用了这个线程并运行了它，但是由于某种原因它给出了关于 mutex.wait() 的错误。当我删除它运行但什么也不显示的代码时。我究竟应该为这个问题做些什么？

public class Main 

public static void main(String[] args) throws InterruptedException 
           
    BarrierSynchronization barrier = new BarrierSynchronization();

    Thread bs1 = new Thread(barrier);
    Thread bs2 = new Thread(barrier);
    Thread bs3 = new Thread(barrier);
    Thread bs4 = new Thread(barrier);
    Thread bs5 = new Thread(barrier);

    bs1.start();
    bs2.start();        
    bs3.start();        
    bs4.start();
    bs5.start();

why does it output the rare one before the barrier is unlocked for all the threads? i think im close enough to solving this problem. is a race condition or something? CLICK TO SEE IMAGE

【问题讨论】：

使用acquire() 等待信号量。

是的，我已经解决了这个问题。我想知道这段代码如何向您展示任务的目的？我到底应该在这里输出什么？

不应该为所有线程共享BarrierSynchronization实例吗？

共享计数器变量：你没有任何共享计数器变量。每个 BarrierSynchronization 实例都有自己的计数字段。

是的，我试图共享同一个类，只是不知道语法。但我修好了。只是不知道为什么会出现图中罕见的输出。也谢谢你们，你们真的很有帮助

【参考方案1】：

可能有点晚了，但这里有一个驱动程序代码有效的实现。您必须保证互斥并跟踪到达屏障的线程数。

public class Barrier 
    private int capacity;
    private Semaphore s, exclusao, counter;

    public Barrier(int capacity) 
        this.capacity = capacity;
        counter = new Semaphore(0);
        s = new Semaphore(0);
        exclusao = new Semaphore(1);
    

    public void espera() throws InterruptedException 
        exclusao.acquire();
        if (counter.availablePermits() < capacity - 1) 
            counter.release();
            exclusao.release();

            s.acquire();
         else             
            exclusao.release();

            System.out.println("RELEASE ALL");
            for (int i = 0; i < capacity; i++) 
                s.release();
            
        
    



class TesteThread extends Thread 
    private Barrier b;
    private long waitPeriod;

    public TesteThread(long wait, Barrier b) 
        this.b = b;
        this.waitPeriod = wait;
        System.out.println("Thread started" + this.getName());

    

    public void espera() throws InterruptedException 
        b.espera();
    

    @Override
    public void run() 
        try 
            System.out.println("Thread a dormir " + this.getName());
            sleep(waitPeriod);
            System.out.println("Thread a esperar " + this.getName());
            espera();
         catch (InterruptedException e) 
            e.printStackTrace();
        
    


class BarrierExample 

    public static void main(String[] args) throws InterruptedException 
        Barrier BR = new Barrier(5);

        TesteThread[] teste = new TesteThread[5];
        for (int i = 0; i < teste.length; i++) 
            teste[i] = new TesteThread((long) (Math.random() * 1000), BR);
            teste[i].start();
        
        for (int i = 0; i < teste.length; i++) 
            teste[i].join();
        

    
`package examesFSO.exame2020_normal;

import java.util.concurrent.Semaphore;

public class Barrier 
    private int capacity;
    private Semaphore s, exclusao, counter;

    public Barrier(int capacity) 
        this.capacity = capacity;
        counter = new Semaphore(0);
        s = new Semaphore(0);
        exclusao = new Semaphore(1);
    

    public void espera() throws InterruptedException 
        exclusao.acquire();
        if (counter.availablePermits() < capacity - 1) 
            counter.release();
            exclusao.release();

            s.acquire();
         else 
            System.out.println("RELEASE ALL");
            for (int i = 0; i < capacity; i++) 
                s.release();
            
        
        exclusao.release();
    



class TesteThread extends Thread 
    private Barrier b;
    private long waitPeriod;

    public TesteThread(long wait, Barrier b) 
        this.b = b;
        this.waitPeriod = wait;
        System.out.println("Thread instanciada " + this.getName());

    

    public void espera() throws InterruptedException 
        b.espera();
    

    @Override
    public void run() 
        try 
            System.out.println("Thread a dormir " + this.getName());
            sleep(waitPeriod);
            System.out.println("Thread a esperar " + this.getName());
            espera();
         catch (InterruptedException e) 
            e.printStackTrace();
        
    


class BarrierExample 

    public static void main(String[] args) throws InterruptedException 
        Barrier BR = new Barrier(5);

        TesteThread[] teste = new TesteThread[5];
        for (int i = 0; i < teste.length; i++) 
            teste[i] = new TesteThread((long) (Math.random() * 1000), BR);
            teste[i].start();
        
        for (int i = 0; i < teste.length; i++) 
            teste[i].join();