为什么CyclicBarrier重置()方法要抛出BrokenBarrierException

Question

我的目的很简单。我想使用CyclicBarrier和它的reset()方法4次运行3个线程，使用下面的代码。研究了网络中的所有可能的资源，在实践中的并发性，以及Java中的思考。无法以我想要的方式解决它。在Java中，在HorseRace.Java下有一种这样的解决方案，但是已经使用了Executor服务。我只想用CyclicBarrier和reset()方法来做这件事。下面是我的代码和输出，它一直运行到最后，但是在reset()方法之后抛出BrokenBarrierException。

package com.apal.barrier;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierEx1 {
    CyclicBarrier cb;
    public static int count = 0;

    public static void main(String[] args) {
        new CyclicBarrierEx1().manageThread();
    }

    private void manageThread() {
        cb = new CyclicBarrier(3, new Runnable() {
            @Override
            public void run() {
                if (count == 3) {
                    System.out.println("Exit from system");
                    return;
                }
                System.out.println("Collating task");
                cb.reset();
                for (int i = 0; i < 3; i++) {
                    new Thread(new Worker(cb)).start();
                }
                count++;
            }
        });

        for (int i = 0; i < 3; i++) {
            new Thread(new Worker(cb)).start();
        }
    }
}

class Worker implements Runnable {
    CyclicBarrier cb;

    public Worker(CyclicBarrier cb) {
        this.cb = cb;
    }

    @Override
    public void run() {
        doSomeWork();
        try {
            cb.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    }

    private void doSomeWork() {
        System.out.println("Doing some work ");
    }
}

样本输出

Doing some work 
Doing some work 
Doing some work 
Collating task
Doing some work 
Doing some work 
Doing some work 
Collating task
Doing some work 
java.util.concurrent.BrokenBarrierExceptionDoing some work 
Doing some work 
Collating task
Doing some work 

    at java.util.concurrent.CyclicBarrier.dowait(Unknown Source)
    at java.util.concurrent.CyclicBarrier.await(Unknown Source)
    at com.apal.barrier.Worker.run(CyclicBarrierEx1.java:48)
    at java.lang.Thread.run(Unknown Source)
java.util.concurrent.BrokenBarrierException
    at java.util.concurrent.CyclicBarrier.dowait(Unknown Source)
    at java.util.concurrent.CyclicBarrier.await(Unknown Source)
Doing some work     at com.apal.barrier.Worker.run(CyclicBarrierEx1.java:48)

    at java.lang.Thread.run(Unknown Source)
Doing some work 
Exit from systemjava.util.concurrent.BrokenBarrierException

    at java.util.concurrent.CyclicBarrier.dowait(Unknown Source)
    at java.util.concurrent.CyclicBarrier.await(Unknown Source)
    at com.apal.barrier.Worker.run(CyclicBarrierEx1.java:48)
    at java.lang.Thread.run(Unknown Source)
java.util.concurrent.BrokenBarrierException
    at java.util.concurrent.CyclicBarrier.dowait(Unknown Source)
    at java.util.concurrent.CyclicBarrier.await(Unknown Source)
    at com.apal.barrier.Worker.run(CyclicBarrierEx1.java:48)
    at java.lang.Thread.run(Unknown Source)
java.util.concurrent.BrokenBarrierException
    at java.util.concurrent.CyclicBarrier.dowait(Unknown Source)
    at java.util.concurrent.CyclicBarrier.await(Unknown Source)
    at com.apal.barrier.Worker.run(CyclicBarrierEx1.java:48)
    at java.lang.Thread.run(Unknown Source)
java.util.concurrent.BrokenBarrierException
    at java.util.concurrent.CyclicBarrier.dowait(Unknown Source)
    at java.util.concurrent.CyclicBarrier.await(Unknown Source)
    at com.apal.barrier.Worker.run(CyclicBarrierEx1.java:48)
    at java.lang.Thread.run(Unknown Source)

Answer 1

在运行CyclicBarrier构造函数中传递的屏障操作时，您有一个争用条件。CyclicBarrier.await()谈到了操作方法是如何运行的(重点是添加的)：

如果当前线程是到达的最后一个线程，并且在构造函数中提供了一个非空屏障操作，那么在允许其他线程继续执行之前，当前线程运行该操作。

这意味着屏障操作方法对reset()的调用可以在其他线程仍在等待屏障时发生。这将导致BrokenBarrierException。

参见该段落的文档，其开头为“如果屏障行动不依赖执行时被暂停的当事方，则该方中的任何线程都可以在释放时执行该操作”。使用这种技术，您可以在工作程序中执行当前正在执行的工作，就在工作线程从await()中释放之后。下面是一次未经测试的尝试(注意--我还对count变量进行了重新调整，以避免在count开始增量之前工作线程可能完成的争用条件)：

package com.apal.barrier;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierEx1 {
    CyclicBarrier cb;
    public static int count = 0;

    public static void main(String[] args) {
        new CyclicBarrierEx1().manageThread();
    }

    public static void barrierComplete(CyclicBarrier cb) {
        System.out.println("Collating task");

        if (count == 3) {
            System.out.println("Exit from system");
            return;
        }
        count++;

        for (int i = 0; i < 3; i++) {
            new Thread(new Worker(cb)).start();
        }
    }

    private void manageThread() {
        cb = new CyclicBarrier(3);

        for (int i = 0; i < 3; i++) {
            new Thread(new Worker(cb)).start();
        }
    }
}


class Worker implements Runnable {
    CyclicBarrier cb;

    public Worker(CyclicBarrier cb) {
        this.cb = cb;
    }

    @Override
    public void run() {
        doSomeWork();
        try {
            if (cb.await() == 0) {
                CyclicBarrierEx1.barrierComplete(cb);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    }

    private void doSomeWork() {
        System.out.println("Doing some work ");
    }
}

Answer 2

根据MichaelBurr的提示，检查工作线程中是否有等待线程计数来重置CyclicBarrier同步器。我发布自己的答案是因为我真的很想使用CyclicBarrier和reset()方法来实现多次运行线程并整理它们的任务，比如矩阵操作。

package com.apal.barrier;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierEx1 {
    CyclicBarrier cb;
    public static int count = 0;

    public static void main(String[] args) {
        new CyclicBarrierEx1().manageThread();
    }

    private void manageThread() {
        cb = new CyclicBarrier(3, new Runnable() {
            @Override
            public void run() {
                if (count == 3) {
                    System.out.println("Exit from system");
                    return;
                }
                System.out.println("Collating task");
                count++;
                // cb.reset();         **Commented and replaced in Worker**
                for (int i = 0; i < 3; i++) {
                    new Thread(new Worker(cb)).start();
                }
            }
        });

        for (int i = 0; i < 3; i++) {
            new Thread(new Worker(cb)).start();
        }
    }
}

class Worker implements Runnable {
    CyclicBarrier cb;

    public Worker(CyclicBarrier cb) {
        this.cb = cb;
    }

    @Override
    public void run() {
        doSomeWork();
        try {
            cb.await();
            //if (cb.getNumberWaiting() == 0) // **if no one is waiting, then reset it.**
            //  cb.reset();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    }

    private void doSomeWork() {
        System.out.println("Doing some work ");
    }
}

样本输出

Doing some work 
Doing some work 
Doing some work 
Collating task
Doing some work 
Doing some work 
Doing some work 
Collating task
Doing some work 
Doing some work 
Doing some work 
Collating task
Doing some work 
Doing some work 
Doing some work 
Exit from system

来自Javadoc的，用于重置()

将势垒重置到初始状态。如果目前有任何一方在隔离墙前等待，他们将带着BrokenBarrierException返回。注意，由于其他原因，中断后的重置可能会很复杂；线程需要以其他方式重新同步，并选择一个来执行重置。相反，最好是为以后的使用建立一个新的屏障。

因此，重置会导致当前正在等待的线程抛出BrokenBarrierException并立即唤醒。当你想要“突破”障碍时，就会使用重置。

在正常情况下，不应该使用reset()。