如何“取消”CountDownLatch？

Question

我有多个使用者线程在使用await()的大小为1的CountDownLatch上等待。我有一个生产者线程，它在成功完成时调用countDown()。

这在没有错误的情况下工作得很好。

但是，如果生产者检测到错误，我希望它能够向使用者线程发出错误信号。理想情况下，我可以让生产者调用像abortCountDown()这样的东西，让所有的消费者接收一个InterruptedException或其他一些异常。我不想调用countDown()，因为这需要我的所有使用者线程在调用await()之后执行额外的手动检查以确定是否成功。我更希望他们只收到一个异常，他们已经知道如何处理。

我知道在CountDownLatch中没有中止功能。有没有另一种同步原语可以让我轻松地创建一个支持中止倒计时的CountDownLatch？

Answer 1

JB Nizet有一个很好的答案。我拿了他的，稍微打磨了一下。结果是CountDownLatch的一个名为AbortableCountDownLatch的子类，它向该类添加了一个"abort()“方法，该方法将导致等待锁存器的所有线程接收一个AbortException (InterruptedException的子类)。

此外，与JB的类不同，AbortableCountDownLatch将在中止时立即中止所有阻塞线程，而不是等待倒计时达到零(对于使用count>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);
        }
    }
}

Answer 2

在内部使用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();
    }
}

Answer 3

您可以在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();
    }
}