18

CountDownLatchを使用して、サイズ 1の を待機している複数のコンシューマ スレッドがありawait()ます。countDown()正常に終了したときに呼び出す単一のプロデューサー スレッドがあります。

これは、エラーがない場合にうまく機能します。

ただし、プロデューサーがエラーを検出した場合は、エラーをコンシューマー スレッドに通知できるようにしたいと考えています。理想的には、プロデューサーに次のようabortCountDown()に呼び出して、すべてのコンシューマーに InterruptedException またはその他の例外を受け取るようにさせることができます。を呼び出したくありませんcountDown()。これには、すべてのコンシューマー スレッドが への呼び出しの後に、成功を確認するために手動で追加のチェックを行う必要があるためawait()です。私はむしろ、彼らがすでに処理方法を知っている例外を受け取ることを望んでいます。

でアボート機能が利用できないことを知っていますCountDownLatch。CountDownLatchカウントダウンの中止をサポートするを効果的に作成するために簡単に適応できる別の同期プリミティブはありますか?

4

5 に答える 5

17

JB Nizet の素晴らしい回答がありました。私は彼を取り、それを少し磨きました。その結果、AbortableCountDownLatch と呼ばれる CountDownLatch のサブクラスが作成され、このクラスに「abort()」メソッドが追加され、ラッチを待機しているすべてのスレッドが AbortException (InterruptedException のサブクラス) を受け取るようになります。

また、JB のクラスとは異なり、AbortableCountDownLatch は、カウントダウンがゼロになるのを待つのではなく、アボート時にすべてのブロッキング スレッドを直ちにアボートします (カウント > 1 を使用する状況の場合)。

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class AbortableCountDownLatch extends CountDownLatch {
    protected boolean aborted = false;

    public AbortableCountDownLatch(int count) {
        super(count);
    }


   /**
     * Unblocks all threads waiting on this latch and cause them to receive an
     * AbortedException.  If the latch has already counted all the way down,
     * this method does nothing.
     */
    public void abort() {
        if( getCount()==0 )
            return;

        this.aborted = true;
        while(getCount()>0)
            countDown();
    }


    @Override
    public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
        final boolean rtrn = super.await(timeout,unit);
        if (aborted)
            throw new AbortedException();
        return rtrn;
    }

    @Override
    public void await() throws InterruptedException {
        super.await();
        if (aborted)
            throw new AbortedException();
    }


    public static class AbortedException extends InterruptedException {
        public AbortedException() {
        }

        public AbortedException(String detailMessage) {
            super(detailMessage);
        }
    }
}
于 2012-05-04T20:37:01.913 に答える
13

内部で CountDownLatch を使用して、特定の上位クラス内にこの動作をカプセル化します。

public class MyLatch {
    private CountDownLatch latch;
    private boolean aborted;
    ...

    // called by consumers
    public void await() throws AbortedException {
        latch.await();
        if (aborted) {
            throw new AbortedException();
        }
    }

    // called by producer
    public void abort() {
        this.aborted = true;
        latch.countDown();
    }

    // called by producer
    public void succeed() {
        latch.countDown();
    }
}
于 2012-05-04T18:08:33.507 に答える
4

CountDownLatchウェイターをキャンセルする機能を提供するラッパーを作成できます。待機中のスレッドを追跡し、タイムアウトしたときにそれらを解放する必要があります。また、ラッチがキャンセルされたことを記憶して、今後の呼び出しがawaitすぐに中断されるようにする必要があります。

public class CancellableCountDownLatch
{
    final CountDownLatch latch;
    final List<Thread> waiters;
    boolean cancelled = false;

    public CancellableCountDownLatch(int count) {
        latch = new CountDownLatch(count);
        waiters = new ArrayList<Thread>();
    }

    public void await() throws InterruptedException {
        try {
            addWaiter();
            latch.await();
        }
        finally {
            removeWaiter();
        }
    }

    public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
        try {
            addWaiter();
            return latch.await(timeout, unit);
        }
        finally {
            removeWaiter();
        }
    }

    private synchronized void addWaiter() throws InterruptedException {
        if (cancelled) {
            Thread.currentThread().interrupt();
            throw new InterruptedException("Latch has already been cancelled");
        }
        waiters.add(Thread.currentThread());
    }

    private synchronized void removeWaiter() {
        waiters.remove(Thread.currentThread());
    }

    public void countDown() {
        latch.countDown();
    }

    public synchronized void cancel() {
        if (!cancelled) {
            cancelled = true;
            for (Thread waiter : waiters) {
                waiter.interrupt();
            }
            waiters.clear();
        }
    }

    public long getCount() {
        return latch.getCount();
    }

    @Override
    public String toString() {
        return latch.toString();
    }
}
于 2012-05-04T18:10:52.383 に答える
0

保護されたメソッドへのアクセスを許可する を使用して、独自のロールCountDownLatchアウトを行うことができます。ReentrantLockgetWaitingThreads

例：

public class FailableCountDownLatch {
    private static class ConditionReentrantLock extends ReentrantLock {
        private static final long serialVersionUID = 2974195457854549498L;

        @Override
        public Collection<Thread> getWaitingThreads(Condition c) {
            return super.getWaitingThreads(c);
        }
    }

    private final ConditionReentrantLock lock = new ConditionReentrantLock();
    private final Condition countIsZero = lock.newCondition();
    private long count;

    public FailableCountDownLatch(long count) {
        this.count = count;
    }

    public void await() throws InterruptedException {
        lock.lock();
        try {
            if (getCount() > 0) {
                countIsZero.await();
            }
        } finally {
            lock.unlock();
        }
    }

    public boolean await(long time, TimeUnit unit) throws InterruptedException {
        lock.lock();
        try {
            if (getCount() > 0) {
                return countIsZero.await(time, unit);
            }
        } finally {
            lock.unlock();
        }
        return true;
    }

    public long getCount() {
        lock.lock();
        try {
            return count;
        } finally {
            lock.unlock();
        }
    }

    public void countDown() {
        lock.lock();
        try {
            if (count > 0) {
                count--;

                if (count == 0) {
                    countIsZero.signalAll();
                }
            }
        } finally {
            lock.unlock();
        }
    }

    public void abortCountDown() {
        lock.lock();
        try {
            for (Thread t : lock.getWaitingThreads(countIsZero)) {
                t.interrupt();
            }
        } finally {
            lock.unlock();
        }
    }
}

このクラスを変更して、キャンセル後にInterruptedExceptionon new 呼び出しをスローすることができます。その機能が必要な場合awaitは、このクラスを拡張することもできます。CountDownLatch

于 2012-05-04T18:19:39.200 に答える