ブロッキングキューから検索を終了する方法
https://stackoverflow.com/questions/3214597
-
13-09-2020 - |
italiano
english
français
española
中国
日本の
العربية
Deutsch
한국어
Português
Russian質問
エグゼキュオタとブロックキューを使用して複数のタスクを実行するコードがいくつかあります。その結果はイテレータとして返されなければなりませんので、私が予想しているアプリケーションのアプリケーションです。ただし、タスクとキューに追加された結果の間には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
これは、待機/通知、 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();
他のヒント
Queueの「Poison」オブジェクトを使用して、タスクがこれ以上結果を提供しないことを示します。
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);
}
}
私は未来あなたが探しているものです。それはあなたが結果オブジェクトと非同期タスクを関連付け、その結果のステータスを照会することを可能にします。始めるタスクごとに、Futureへの参照を保ち、それを使用して完了したかどうかを判断します。
あなたの問題を正しく理解した場合（私が私が確かないとわからない）、Take（）の代わりに[blockingqueue.poll] [1]を使用して、空のキューの無限の待機を防ぐことができます。これにより、タイムアウトを指定できます。その後、キューが空の場合はnullが返されます。
hasNextの実装（適切に短いタイムアウトを持つ）にまっすぐドロップした場合、ロジックは正しいです。空のキューはfalseを返します。
残りのエンティティはtrueを返します。
[1]：htht://java.sun.com/javase/6/docs/api/java/util/concurrent/blockingqueue.html#poll（long "rel=" nofollow noreferrer "> http://java.sun.com/javase/6/docs/api/java/util/concurrent/blockingqueue.html#poll（long 、java.util.concurrent.timeUnit）
私はあなたを理解しているわけではありませんが、ワーカースレッドがキューに一覧表示されていることができないのはなぜですか。その後、外部ループ内のキューを越えて超和子を介してカスタムイテレータを作成できます。並行性の魔法なし。
所属していません StackOverflow
