异步迭代器

Question

我有以下代码:

while(slowIterator.hasNext()) {
  performLengthTask(slowIterator.next());
}

因为迭代器和任务都很慢,所以将它们放入单独的线程是有意义的.以下是Iterator包装器的快速而脏的尝试:

class AsyncIterator<T> implements Iterator<T> {
    private final BlockingQueue<T> queue = new ArrayBlockingQueue<T>(100);

    private AsyncIterator(final Iterator<T> delegate) {
      new Thread() {
        @Override
        public void run() {
          while(delegate.hasNext()) {
            queue.put(delegate.next()); // try/catch removed for brevity
          }
        }
      }.start();
    }

    @Override
    public boolean hasNext() {
      return true;
    }

    @Override
    public T next() {
        return queue.take(); // try/catch removed for brevity
    }
    // ... remove() throws UnsupportedOperationException
  }

但是,这种实现缺乏对"hasNext()"的支持.当然可以阻止hasNext()方法阻塞,直到它知道是否返回true.我可以在我的AsyncIterator中有一个peek对象,我可以更改hasNext()从队列中获取一个对象并让next()返回此窥视.但是如果已达到委托迭代器的结尾,这将导致hasNext()无限期地阻塞.

我自己可以自己进行线程通信,而不是使用ArrayBlockingQueue:

private static class AsyncIterator<T> implements Iterator<T> {

  private final Queue<T> queue = new LinkedList<T>();
  private boolean delegateDone = false;

  private AsyncIterator(final Iterator<T> delegate) {
    new Thread() {
      @Override
      public void run() {
        while (delegate.hasNext()) {
          final T next = delegate.next();
          synchronized (AsyncIterator.this) {
            queue.add(next);
            AsyncIterator.this.notify();
          }
        }
        synchronized (AsyncIterator.this) {
          delegateDone = true;
          AsyncIterator.this.notify();
        }
      }
    }.start();
  }

  @Override
  public boolean hasNext() {
    synchronized (this) {
      while (queue.size() == 0 && !delegateDone) {
        try {
          wait();
        } catch (InterruptedException e) {
          throw new Error(e);
        }
      }
    }
    return queue.size() > 0;
  }

  @Override
  public T next() {
    return queue.remove();
  }

  @Override
  public void remove() {
    throw new UnsupportedOperationException();
  }
}

然而,所有额外的同步,等待和通知并没有真正使代码更具可读性,并且很容易在某处隐藏竞争条件.

有更好的想法吗？

更新

是的我知道常见的观察者/可观察的模式.但是,通常的实现并不预见数据流的结束,它们不是迭代器.

我特别想要一个迭代器,因为实际上上面提到的循环存在于一个外部库中,它需要一个迭代器.

Answer 1

这是一个棘手的问题,但我想这次我得到了正确答案.(我删除了我的第一个答案.)

答案是使用哨兵.我没有测试过这段代码,为了清楚起见,我删除了try/catches:

public class AsyncIterator<T> implements Iterator<T> {

    private BlockingQueue<T> queue = new ArrayBlockingQueue<T>(100);
    private T sentinel = (T) new Object();
    private T next;

    private AsyncIterator(final Iterator<T> delegate) {
        new Thread() {
            @Override
            public void run() {
                while (delegate.hasNext()) {
                    queue.put(delegate.next());
                }
                queue.put(sentinel);
            }
        }.start();
    }

    @Override
    public boolean hasNext() {
        if (next != null) {
            return true;
        }
        next = queue.take(); // blocks if necessary
        if (next == sentinel) {
            return false;
        }
        return true;
    }

    @Override
    public T next() {
        T tmp = next;
        next = null;
        return tmp;
    }

}

这里的见解是hasNext()需要阻塞,直到下一个项目准备好.它还需要某种退出条件,并且由于线程问题,它不能使用空队列或布尔标志.哨兵在没有任何锁定或同步的情况下解决问题.

编辑:缓存"下一步",因此可以多次调用hasNext().