等待未知数量的异步任务

Question

我有一个任务队列和一个线程，在几秒钟内窥探一次队列，并且它有一个任务执行它。

我有另一个代码段（当然是另一个线程），它在循环中创建任务（我无法预先从循环外部知道任务数量）并将它们插入到队列中。任务包含一些“结果”对象，外部线程（创建这些任务）需要等待所有任务完成并最终从每个任务中获取结果。 问题是我无法将java Semaphore \ CountDownLatch等传递给结果对象，因为我不知道提前显示器的数量。 我也不能使用使用invokeAll的Executor，或者等待Future对象，因为任务是不同步的（外部线程只是将任务加入到队列中，而另一个线程在他有空时执行任务）。

我唯一想到的解决方案是创建一些“反转信号量”类，它包含一组结果和一个监视器计数器。该功能的getResult将检查计数器== 0，如果答案是肯定就会通知一些锁定的对象，以及功能的getResult会等待这个锁：

public class InvertedSemaphore<T> { 
    Set<T> resultSet; 
    int usages; 
    final Object c; 

    public InvertedSemaphore() { 
     resultSet = Collections.synchronizedSet(new HashSet<T>()); 
     usages = 0; 
     c = new Object(); 
    } 

    public void addResult(T result) { 
     resultSet.add(result); 
    } 

    public void addResults(Set<T> result) { 
     resultSet.addAll(result); 
    } 


    public void acquire() { 
     usages++; 
    } 

    public void release() { 
     synchronized (c) { 
      if (--usages == 0) { 
       c.notify(); 
      } 
     } 
    } 

    public Set<T> getResults() { 
     synchronized (c) { 
      try { 
       while (usages > 0) { 
        c.wait(); 
       } 
      } catch (InterruptedException e) { 
       e.printStackTrace(); 
      } 
     } 
     return resultSet; 
    } 

} 

每个addTask方法调用semaphore.acquire，和每个（未同步）任务将在任务结束时调用semaphore.release。

这听起来很复杂，我很确定在java并发库中有更好的解决方案。

任何想法将appriciated :)

Answer 1

如果任务不需要为了进行处理，使用ExecutorCompletionService

更一般地说，没有必要为了上ExecutorService使用invokeAll得到Future的结果。 ExecutorService#submit可以用于此目的，或者可选地，创建的任务本身可以实现Future，从而允许任务的创建者在稍后的时间点询问结果。

一些代码：

class MyTask { 
    AtomicReference<?> result = new AtomicReference<?>(); 

    void run() { 
     //do stuff here 
     result.set(/* the result of the calculation */); 
    } 

    boolean resultReady() { 
     return result.get()!=null; 
    } 

    ? get() { 
     return result.get(); 
    } 
} 

...别处代码

void createTasks() { 
    Collection<MyTask> c = new ...; 

    while(indeterminable condition) { 
     MyTask task = new MyTask(); 
     c.add(task); 
     mysteryQueue.add(task); 
    } 

    while(haven't received all results) { 
     MyTask task = c.get(...); //or iterate or whatever 
     ? result = task.get(); 
     if (result!=null) { 
      //do stuff, probably remove the task from the collection c would be smart 
     } 
    } 
} 

Answer 2

一个想法是使用一个单独的队列结果。
所以你将有一个阻塞队列，线程A为线程B放置任务，从而具有生产者 - 消费者的方法，并且当每个任务完成时，结果可以被放置在的第二个结果队列中，将消费者 - 生产者角色因为现在线程A最初创建的任务将消耗第二个队列的结果。

Answer 3

您可以执行以下操作： 每个生产者都将拥有自己的队列。生产者将通过一种手段将此队列报告给任务本身。当任务完成运行时，它将把结果排队到这个队列中。它是由一些代码描述的兽：

class Result{} 

interface IResultCallback{ 
    void resultReady(Result r); // this is an abstraction of the queue 
} 

class Producer implements IResultCallback{ 
    // the producer needs to pass itself to the constructor of the task, 
    // the task will only see its "resultReady" facade and will be able to report to it. 
    // the producer can aggragte the results at it will and execute its own computation as 
     // as soon it is ready 

    Queue<Result> results; // = init queue 

    @Override 
    public void resultReady(Result r) { 
     results.add(r); 

     if(results.size() == 9){ 
      operate(); 
     } 
     results.clear(); 
    } 

    public void operate(){ 
     // bla bla 
    } 
} 

public class Task { 
    IResultCallback callback; 

    public Task(IResultCallback callback){ 
     this.callback = callback; 
    } 
    public void execute(){ 
     // bla bla 

     Result r = null; // init result; 
     callback.resultReady(r); 
    } 
}