3

我有一些代码，我使用执行器和阻塞队列执行多个任务。结果必须作为迭代器返回，因为这是我工作的应用程序所期望的。但是，任务和添加到队列的结果之间存在 1:N 的关系，因此我不能使用ExecutorCompletionService。在调用 hasNext() 时，我需要知道所有任务何时完成并将所有结果添加到队列中，以便我可以停止从队列中检索结果。请注意，一旦项目被放入队列，另一个线程应该准备好使用（Executor.invokeAll(), 阻塞直到所有任务完成，这不是我想要的，也不是超时）。这是我的第一次尝试，我使用 AtomicInteger 只是为了证明这一点，即使它不起作用。有人可以帮助我了解如何解决这个问题吗？

public class ResultExecutor<T> implements Iterable<T> {
    private BlockingQueue<T> queue;
    private Executor executor;
    private AtomicInteger count;

    public ResultExecutor(Executor executor) {
        this.queue = new LinkedBlockingQueue<T>();
        this.executor = executor;
        count = new AtomicInteger();            
    }

    public void execute(ExecutorTask task) {
        executor.execute(task);
    }

    public Iterator<T> iterator() {
        return new MyIterator();
    }

    public class MyIterator implements Iterator<T> {
        private T current;          
        public boolean hasNext() {
            if (count.get() > 0 && current == null)
            {
                try {
                    current = queue.take();
                    count.decrementAndGet();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            return current != null;
        }

        public T next() {
            final T ret = current;
            current = null;
            return ret;
        }

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

    }

    public class ExecutorTask implements Runnable{
        private String name;

        public ExecutorTask(String name) {
            this.name = name;

        }

         private int random(int n)
         {
           return (int) Math.round(n * Math.random());
         }


        @SuppressWarnings("unchecked")
        public void run() {
            try {
                int random = random(500);
                Thread.sleep(random);
                queue.put((T) (name + ":" + random + ":1"));
                queue.put((T) (name + ":" + random + ":2"));
                queue.put((T) (name + ":" + random + ":3"));
                queue.put((T) (name + ":" + random + ":4"));
                queue.put((T) (name + ":" + random + ":5"));

                count.addAndGet(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }                   
        }           
    }       

}

调用代码如下所示：

    Executor e = Executors.newFixedThreadPool(2);
    ResultExecutor<Result> resultExecutor = new ResultExecutor<Result>(e);

    resultExecutor.execute(resultExecutor.new ExecutorTask("A"));
    resultExecutor.execute(resultExecutor.new ExecutorTask("B"));

    Iterator<Result> iter = resultExecutor.iterator();
    while (iter.hasNext()) {
        System.out.println(iter.next());
    }
4

5 回答 5

3

在 中使用“毒药”对象Queue表示任务将不再提供结果。

class Client
{

  public static void main(String... argv)
    throws Exception
  {
    BlockingQueue<String> queue = new LinkedBlockingQueue<String>();
    ExecutorService workers = Executors.newFixedThreadPool(2);
    workers.execute(new ExecutorTask("A", queue));
    workers.execute(new ExecutorTask("B", queue));
    Iterator<String> results = 
      new QueueMarkersIterator<String>(queue, ExecutorTask.MARKER, 2);
    while (results.hasNext())
      System.out.println(results.next());
  }

}

class QueueMarkersIterator<T>
  implements Iterator<T>
{

  private final BlockingQueue<? extends T> queue;

  private final T marker;

  private int count;

  private T next;

  QueueMarkersIterator(BlockingQueue<? extends T> queue, T marker, int count)
  {
    this.queue = queue;
    this.marker = marker;
    this.count = count;
    this.next = marker;
  }

  public boolean hasNext()
  {
    if (next == marker)
      next = nextImpl();
    return (next != marker);
  }

  public T next()
  {
    if (next == marker)
      next = nextImpl();
    if (next == marker)
      throw new NoSuchElementException();
    T tmp = next;
    next = marker;
    return tmp;
  }

  /*
   * Block until the status is known. Interrupting the current thread 
   * will cause iteration to cease prematurely, even if elements are 
   * subsequently queued.
   */
  private T nextImpl()
  {
    while (count > 0) {
      T o;
      try {
        o = queue.take();
      }
      catch (InterruptedException ex) {
        count = 0;
        Thread.currentThread().interrupt();
        break;
      }
      if (o == marker) {
        --count;
      }
      else {
        return o;
      }
    }
    return marker;
  }

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

}

class ExecutorTask
  implements Runnable
{

  static final String MARKER = new String();

  private static final Random random = new Random();

  private final String name;

  private final BlockingQueue<String> results;

  public ExecutorTask(String name, BlockingQueue<String> results)
  {
    this.name = name;
    this.results = results;
  }

  public void run()
  {
    int random = ExecutorTask.random.nextInt(500);
    try {
      Thread.sleep(random);
    }
    catch (InterruptedException ignore) {
    }
    final int COUNT = 5;
    for (int idx = 0; idx < COUNT; ++idx)
      results.add(name + ':' + random + ':' + (idx + 1));
    results.add(MARKER);
  }

}
于 2010-07-09T16:47:40.673 回答
1

我相信未来是你正在寻找的。它允许您将异步任务与结果对象相关联，并查询该结果的状态。对于您开始的每个任务，保留对其的引用Future并使用它来确定它是否已完成。

于 2010-07-09T16:33:16.310 回答
1

如果我正确理解了您的问题（我不确定是否正确），您可以通过使用 [BlockingQueue.poll][1] 而不是 take() 来防止在空队列上无限等待。这使您可以指定超时，null如果队列为空，则将返回超时时间。

如果您将其直接放入您的hasNext实现中（具有适当的短超时），则逻辑将是正确的。空队列将返回false，而剩余实体的队列将返回true。

[1]： http: //java.sun.com/javase/6/docs/api/java/util/concurrent/BlockingQueue.html#poll (long , java.util.concurrent.TimeUnit)

于 2010-07-09T16:42:02.717 回答
0

这是一个替代解决方案，它使用带有等待/通知、 AtomicInteger和回调的非阻塞队列。

public class QueueExecutor implements CallbackInterface<String> {

    public static final int NO_THREADS = 26;

    private Object syncObject = new Object();
    private AtomicInteger count;
    Queue<String> queue = new LinkedList<String>();

    public void execute() {
        count = new AtomicInteger(NO_THREADS);
        ExecutorService executor = Executors.newFixedThreadPool(NO_THREADS/2);
        for(int i=0;i<NO_THREADS;i++)
            executor.execute(new ExecutorTask<String>("" + (char) ('A'+i), queue, this));

        Iterator<String> iter = new QueueIterator<String>(queue, count);
        int count = 0;
        while (iter.hasNext()) {

            System.out.println(iter.next());
            count++;
        }

        System.out.println("Handled " + count + " items");
    }

    public void callback(String result) {
        System.out.println(result);
        count.decrementAndGet();
        synchronized (syncObject) {
            syncObject.notify();
        }
    }

    public class QueueIterator<T> implements Iterator<T> {
        private Queue<T> queue;
        private AtomicInteger count;

        public QueueIterator(Queue<T> queue, AtomicInteger count) {
            this.queue = queue;
            this.count = count;
        }

        public boolean hasNext() {          
            while(true) {                   
                synchronized (syncObject) {
                    if(queue.size() > 0)
                        return true;

                    if(count.get() == 0)
                        return false;

                    try {
                        syncObject.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        public T next() {

            synchronized (syncObject) {
                if(hasNext())
                    return queue.remove();
                else
                    return null;
            }
        }

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

    }

    class ExecutorTask<T> implements Runnable {
        private String name;
        private Queue<T> queue;
        private CallbackInterface<T> callback;

        public ExecutorTask(String name, Queue<T> queue,
                CallbackInterface<T> callback) {
            this.name = name;
            this.queue = queue;
            this.callback = callback;
        }

        @SuppressWarnings("unchecked")
        public void run() {
            try {
                Thread.sleep(1000);
                                    Random randomX = new Random();
                for (int i = 0; i < 5; i++) {
                    synchronized (syncObject) {
                        Thread.sleep(randomX.nextInt(10)+1);

                        queue.add((T) (name + ":" + ":" + i));
                        syncObject.notify();
                    }
                }

                callback.callback((T) (name + ": Done"));

            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

public interface CallbackInterface<T> {
    void callback(T result);
}

调用代码很简单：

    QueueExecutor exec = new QueueExecutor();
    exec.execute();
于 2010-07-12T15:59:52.477 回答
0

我不确定我是否理解你，但为什么工作线程不能将自己的列表放入队列中。然后，您可以创建一个自定义迭代器，该迭代器在外循环中遍历队列并通过子迭代器。所有这些都没有并发魔法。

于 2013-03-07T23:17:26.670 回答