Bir CountDownLatch'ı nasıl "iptal ederim"?

Question

await() numarasını kullanarak CountDownLatch boyutunda 1'in üzerinde birden çok tüketici iletkeni var. Başarıyla tamamlandığında, countDown() numaralı telefonu arayarak tek bir üretici iş parçacığım var.

Bu, hata olmadığında mükemmel çalışır. Bununla birlikte, eğer üretici bir hata tespit ederse, hata mesajını tüketici ipliklerine bildirebilmek isterim. İdeal olarak, üreticinin abortCountDown() gibi bir şey çağırmasını ve tüm tüketicilerin bir InterruptedException veya başka bir istisna almasını sağlayabiliyordum. countDown() numaralı telefonu aramak istemiyorum çünkü bu, tüm tüketici iş parçacıklarımın await() numaralı çağrıdan sonra başarı için ek bir manuel kontrol yapmasını gerektiriyor. Nasıl başa çıkacaklarını bildikleri bir istisna almayı tercih ederim.

Bir iptal etme tesisinin CountDownLatch numaralı telefondan temin edilemediğini biliyorum. Geri sayımın iptal edilmesini destekleyen CountDownLatch'u etkili bir şekilde oluşturmak için kolayca uyarlayabileceğim bir başka senkronizasyon var mı?

Answer 1

JB Nizet büyük bir cevap aldı. Onu aldım ve biraz parlattım. Sonuç, AbortableCountDownLatch adlı bir alt sınıftır. Bu, bir abort() yöntemini AbortException (InterruptedException alt sınıfı) almak için mandalda bekleyen tüm iş parçacıklarına neden olacak bir "abort()" yöntemi ekler. Ayrıca, JB sınıfının aksine, AbortableCountDownLatch, geri sayımın sıfıra ulaşmasını beklemekten ziyade tüm engelleme iş parçacıklarını derhal iptal eder (bir sayımı> 1 kullandığınız durumlar için).

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

Kapsüllemek içten CountDownLatch kullanarak belirli, üst düzey sınıf içinde bu davranış,:

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

Sen garsonlar iptal etme olanağını sağlar CountDownLatch etrafında sarıcı oluşturabilir. Bekleyen iş parçacıklarını izlemeli ve zaman aşımına uğradıklarında onları serbest bırakmalılar ve aynı zamanda gelecekteki çağrıların await araya gireceğini hatırlatır.

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(); 
    } 
} 

Answer 4

Sen kendi CountDownLatch koruma altındaki getWaitingThreads yöntemine erişime izin veren bir ReentrantLock kullanarak dışarı rulo başladı.

Örnek:

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(); 
     } 
    } 
} 

Bunu iptal edildikten sonra await yeni çağrılarda bir InterruptedException atmak için bu sınıfı değiştirmek isteyebilirsiniz. Bu işleve ihtiyacınız varsa, bu sınıfa CountDownLatch'u bile sahip olabilirsiniz.