在不可变的Iterable上并发迭代

Question

我有一个包含大量元素的不可变Iterable<X>。(它恰好是一个List<>，但没关系。)

我想做的是启动一些并行/异步任务，用相同的迭代器和迭代Iterable<> ，我想知道我应该使用什么接口。

下面是一个要确定的接口QuasiIteratorInterface的示例实现:

public void process(Iterable<X> iterable)
{
   QuasiIteratorInterface<X> qit = ParallelIteratorWrapper.iterate(iterable);
   for (int i = 0; i < MAX_PARALLEL_COUNT; ++i)
   {
      SomeWorkerClass worker = new SomeWorkerClass(qit);
      worker.start();
   }
}
class ParallelIteratorWrapper<T> implements QuasiIteratorInterface<T>
{
   final private Iterator<T> iterator;
   final private Object lock = new Object();
   private ParallelIteratorWrapper(Iterator<T> iterator) { 
      this.iterator = iterator;
   }
   static public <T> ParallelIteratorWrapper<T> iterate(Iterable<T> iterable)
   {
      return new ParallelIteratorWrapper(iterable.iterator());
   }
   private T getNextItem()
   {
      synchronized(lock)
      {
         if (this.iterator.hasNext())
            return this.iterator.next();
         else
            return null;
      }
   }
   /* QuasiIteratorInterface methods here */
}

我的问题是:

• 直接使用Iterator是没有意义的，因为hasNext()和next()有一个同步问题，如果其他人在你之前调用next()， hasNext()是无用的。

• 我想使用Queue，但我需要的唯一方法是poll()

• 我想使用ConcurrentLinkedQueue来保存我的大量元素…但是我可能需要多次遍历元素，所以我不能使用

有什么建议吗?

Answer 1

使用poll()方法或等效方法(例如Guava的Supplier)创建自己的Producer接口。实现选项有很多，但是如果你有一个不可变的随机访问列表，那么你可以简单地维护一个线程安全的单调计数器(例如AtomicInteger)并调用list.get(int)例如:

class ListSupplier<T> implements Supplier<T> {
  private final AtomicInteger next = new AtomicInteger();
  private final List<T> elements; // ctor injected
  …
  public <T> get() {
    // real impl more complicated due to bounds checks
    // and what to do when exhausted
    return elements.get(next.getAndIncrement());
  }
}

这是线程安全的，但是你可能想要返回一个Option样式的东西或者在用尽时返回null。

Answer 2

有一个dispatcher线程，它在Iterable上迭代，并将元素分派给多个执行元素工作的worker线程。您可以使用ThreadPoolExecutor自动执行此操作。