4

我在使用共享 a 的 n Runnable 类的 Java 并发解决方案中遇到问题CyclicBarrier，并且Runnable由 a 处理ExecutorService，这是代码：

public class Worker implements Runnable {
   private CyclicBarrier writeBarrier;
   private int index;
   private int valuetocalculate;

   public Worker(int i,CyclicBarrier writeBarrier)
   {
      this.writeBarrier = writeBarrier;
      this.index = i;
      this.valuetocalculate = 0;
   }

   public void run() {

      //calculations with valuetocalculate
      writeBarrier.await();
      //write new valuetocalculate value
   }
}

public class Context {

    private ArrayList<Worker> workers;
    private Chief chief;

    public Context()
    {
       workers = new ArrayList<Worker>();
       chief = new Chief();
    }

    public void generateRandomWorkers(nworkers)
    {
       writeBarrier = newWriteBarrier(workers);
       chief.setBarrier(writeBarrier);
       //generate random woker
       for (int i = 0; i<nworkers;i++)
       {
           Worker worker = new Worker(i,writeBarrier);
           workers.add(worker);
       }
       chief.setWorkersArray(workers);
       chief.start();
     }
}

public class Chief extend Thread {

    private CyclicBarrier writeBarrier;
    private ArrayList<Worker> workers;
    private ExecutorService executor;
    private int cores;

    public Chief ()
    {
       cores = Runtime.getRuntime().availableProcessors()+1;
    }

    public void setBarrier (CyclicBarrier writeBarrier)
    {
       this.writeBarrier = writeBarrier;
    }

    public setWorkersArray(ArrayList<Worker> workers)
    {
       this.workers = workers;
    }

    public ArrayList<Integer> getvaluetocalculate()
    {
        ArrayList<Integer> values = new ArrayList<Integer> ();
        for (int i = 0; i<workers.size();i++)
        {
           values.add(workers.get(i).valuetocalculate);
        }

         return values;
    }

    public void run(){
       while (!stop) //always true for testing
       {
          getvaluetocalculate();
          //make calculations
          writeBarrier.reset();
          executor = Executors.newFixedThreadPool(cores);
          for (int i = 0;i<workers.size();i++)
          {
             Runnable runnable = workers.get(i);
             executor.execute(runnable);
           }
           executor.shutdown();
           while (!executor.isTerminated())
           {
           }
        }
     }
}

一切从主要开始：

Context = new Context();
context.generateRandomWorkers();

问题是Runnable没有超过运行中的第一个“迭代” Chief，所以问题似乎是工人没有超过writerBarrier.await();，而是如果我初始化了这个：

executor = Executors.newFixedThreadPool(cores);

与workers.size(), 有效，但似乎不同步...我该如何解决？

4

1 回答 1

1

好的，看起来您正在尝试执行以下操作

  • 一名控制工人的主管/调度员
  • 一名或多名工人进行计算
  • 酋长处决所有工人
  • 酋长等待所有工人完成
  • Chief 从每个 Worker 获取结果以计算结果

假设以上是您的问题。

  • 在worker run() 方法中执行barrier.await() 可防止该线程被释放回池以运行后续worker。因此，当池大小 < 工作者大小时，第一个“池大小”工作者消耗线程，然后停止等待其他无法运行的线程。这就是为什么除非您将池大小更改为workers.size()，否则您的所有worker 都不会运行。
  • 该valuetocalculate变量在设置结果的工作人员和读取它的负责人之间不同步，因此您可能会看到陈旧的结果。

实现这种系统的正确方法是让工作人员实现 Callable，一旦工作人员计算完毕，可调用对象就会返回结果。这负责将结果发布给您的主管（您将在下面看到这是如何工作的）。

移除循环障碍，你不需要那个。

像现在一样创建执行程序，并invokeAll()使用 Callables（您的工作人员）列表进行调用。此方法使用执行程序调用 Callables 并等待它们完成。它会一直阻塞，直到所有工作人员都完成，此时它将返回一个List<Future<T>>. 每个 Future 对应于您传入的其中一个工人/可调用对象。迭代列表以提取结果。如果工作人员未能尝试从中get()获取结果，Future则会引发异常。

希望有帮助。

于 2014-05-09T11:45:37.107 回答