选择一个用于屏障动作执行的线程 - Java CyclicBarrier

Question

查看CyclicBarrier的javadocs,我在类文档中发现了以下语句,我并不完全理解.来自javadoc:

如果屏障操作不依赖于各方在执行时被暂停,那么该方中的任何线程都可以在释放时执行该操作.为了促进这一点,每次调用await()都会返回该线程在屏障处的到达索引.然后,您可以选择应执行屏障操作的线程,例如:

if (barrier.await() == 0) {
  // log the completion of this iteration
} 

有人可以解释如何在所有各方调用.await()之后指定一个特定线程来执行屏障操作并提供一个示例吗？

Answer 1

CyclicBarrier 允许按 ORDER 指定线程：

如果如您所说，将屏障完成逻辑包含在特定于线程索引的条件中，则可以指定按特定顺序返回的线程。因此，您的上述实现将根据您引用的文档进行工作。

然而，这里的混淆点是文档正在讨论返回屏障的顺序的线程标识，而不是线程对象标识。因此，线程0指的是第0个线程完成。

替代方案：使用其他机制指定线程。

如果您希望在其他工作完成后让特定线程执行特定操作，您可以使用不同的机制 - 例如信号量。如果您想要这种行为，您可能并不真正需要循环屏障。

要检查文档的含义，请运行下面的类（修改自http://programmingexamples.wikidot.com/cyclobarrier），我在其中合并了您的代码片段。

CyclicBarrier 文档含义的示例

封装线程；导入 java.util.concurrent.BrokenBarrierException; 导入 java.util.concurrent.CyclicBarrier;

public class CyclicBarrierExample
{
    private static int matrix[][] = 
    { 
        { 1 }, 
        { 2, 2 }, 
        { 3, 3, 3 },
        { 4, 4, 4, 4 }, 
        { 5, 5, 5, 5, 5 } };

    static final int rows = matrix.length;
    private static int results[]=new int[rows];


    static int threadId=0;
    private static class Summer extends Thread
    {
        int row;

        CyclicBarrier barrier;

        Summer(CyclicBarrier barrier, int row)
        {
            this.barrier = barrier;
            this.row = row;
        }

        public void run()
        {
            int columns = matrix[row].length;
            int sum = 0;
            for (int i = 0; i < columns; i++)
            {
                sum += matrix[row][i];
            }
            results[row] = sum;
            System.out.println("Results for row " + row + " are : " + sum);
            // wait for the others 
            // Try commenting the below block, and watch what happens. 
            try
            {
                int w = barrier.await();
                if(w==0)
                {
                    System.out.println("merging now !");
                    int fullSum = 0;
                    for (int i = 0; i < rows; i++)
                    {

                        fullSum += results[i];
                    }
                    System.out.println("Results are: " + fullSum);
                }
            }
            catch(Exception e)
            {
                e.printStackTrace();
            }
        }
    }
    public static void main(String args[])
    {
        /*
         * public CyclicBarrier(int parties,Runnable barrierAction)
         * Creates a new CyclicBarrier that will trip when the given number
         * of parties (threads) are waiting upon it, and which will execute 
         * the merger task when the barrier is tripped, performed 
         * by the last thread entering the barrier.
         */
        CyclicBarrier barrier = new CyclicBarrier(rows );
        for (int i = 0; i < rows; i++)
        {
            System.out.println("Creating summer " + i);
            new Summer(barrier, i).start();

        }
        System.out.println("Waiting...");
    }
}