Я скопировал CountDownLatch и реализовал метод reset(), который сбрасывает внутренний класс Sync в его начальное состояние (начальный счет) :) Кажется, работает нормально. Больше не нужно создавать ненужные объекты \ o / Невозможно создать подкласс, потому что sync был закрытым. Boo.
<code>import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
/**
* A synchronization aid that allows one or more threads to wait until
* a set of operations being performed in other threads completes.
*
* <p>A {@code CountDownLatch} is initialized with a given <em>count</em>.
* The {@link #await await} methods block until the current count reaches
* zero due to invocations of the {@link #countDown} method, after which
* all waiting threads are released and any subsequent invocations of
* {@link #await await} return immediately. This is a one-shot phenomenon
* -- the count cannot be reset. If you need a version that resets the
* count, consider using a {@link CyclicBarrier}.
*
* <p>A {@code CountDownLatch} is a versatile synchronization tool
* and can be used for a number of purposes. A
* {@code CountDownLatch} initialized with a count of one serves as a
* simple on/off latch, or gate: all threads invoking {@link #await await}
* wait at the gate until it is opened by a thread invoking {@link
* #countDown}. A {@code CountDownLatch} initialized to <em>N</em>
* can be used to make one thread wait until <em>N</em> threads have
* completed some action, or some action has been completed N times.
*
* <p>A useful property of a {@code CountDownLatch} is that it
* doesn't require that threads calling {@code countDown} wait for
* the count to reach zero before proceeding, it simply prevents any
* thread from proceeding past an {@link #await await} until all
* threads could pass.
*
* <p><b>Sample usage:</b> Here is a pair of classes in which a group
* of worker threads use two countdown latches:
* <ul>
* <li>The first is a start signal that prevents any worker from proceeding
* until the driver is ready for them to proceed;
* <li>The second is a completion signal that allows the driver to wait
* until all workers have completed.
* </ul>
*
* <pre>
* class Driver { // ...
* void main() throws InterruptedException {
* CountDownLatch startSignal = new CountDownLatch(1);
* CountDownLatch doneSignal = new CountDownLatch(N);
*
* for (int i = 0; i < N; ++i) // create and start threads
* new Thread(new Worker(startSignal, doneSignal)).start();
*
* doSomethingElse(); // don't let run yet
* startSignal.countDown(); // let all threads proceed
* doSomethingElse();
* doneSignal.await(); // wait for all to finish
* }
* }
*
* class Worker implements Runnable {
* private final CountDownLatch startSignal;
* private final CountDownLatch doneSignal;
* Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
* this.startSignal = startSignal;
* this.doneSignal = doneSignal;
* }
* public void run() {
* try {
* startSignal.await();
* doWork();
* doneSignal.countDown();
* } catch (InterruptedException ex) {} // return;
* }
*
* void doWork() { ... }
* }
*
*
*
*
Другое типичное использование - разделить проблему на N частей,
* опишите каждую часть с помощью Runnable, который выполняет эту часть и
* считает на защелке и ставит все Runnables в очередь
* Исполнитель. Когда все части завершены, координирующий поток
* сможет пройти через ждать. (Когда темы должны повторяться
* отсчитывать таким образом, вместо этого используйте {@link CyclicBarrier}.)
*
*
* class Driver2 { // ...
* void main() throws InterruptedException {
* CountDownLatch doneSignal = new CountDownLatch(N);
* Executor e = ...
*
* for (int i = 0; i < N; ++i) // create and start threads
* e.execute(new WorkerRunnable(doneSignal, i));
*
* doneSignal.await(); // wait for all to finish
* }
* }
*
* class WorkerRunnable implements Runnable {
* private final CountDownLatch doneSignal;
* private final int i;
* WorkerRunnable(CountDownLatch doneSignal, int i) {
* this.doneSignal = doneSignal;
* this.i = i;
* }
* public void run() {
* try {
* doWork(i);
* doneSignal.countDown();
* } catch (InterruptedException ex) {} // return;
* }
*
* void doWork() { ... }
* }
*
*
*
*
Эффекты согласованности памяти: действия в потоке перед вызовом
* {@code countDown ()}
* произойдет до
* действия после успешного возвращения из соответствующего
* {@code await ()} в другом потоке.
*
* @ с 1,5
* @author Даг Ли
* /
открытый класс ResettableCountDownLatch {
/ **
* Синхронизация управления для CountDownLatch.
* Использует состояние AQS для представления количества.
* /
закрытый статический финальный класс Sync extends AbstractQueuedSynchronizer {
приватный статический финал long serialVersionUID = 4982264981922014374L;
public final int startCount;
Sync (int count) {
this.startCount = count;
SetState (startCount);
}
int getCount () {
return getState ();
}
public int tryAcquireShared (int acquies) {
return getState () == 0? 1: -1;
}
public boolean tryReleaseShared (int Releases) {
// Decrement count; сигнал при переходе на ноль
за (;;) {
int c = getState ();
если (с == 0)
вернуть ложь;
int nextc = c-1;
if (compareAndSetState (c, nextc))
return nextc == 0;
}
}
public void reset () {
SetState (startCount);
}
}
приватная финальная синхронизация синхронизации;
/ **
* Создает {@code CountDownLatch}, инициализированный с заданным количеством.
*
* @param count, сколько раз {@link #countDown} должен быть вызван
* до того, как потоки могут пройти через {@link #await}
* @throws IllegalArgumentException, если {@code count} отрицателен
* /
public ResettableCountDownLatch (int count) {
if (count <0) выдает новое IllegalArgumentException ("count <0");
this.sync = новая синхронизация (количество);
}
/ **
* Заставляет текущий поток ждать, пока защелка не обратится к
* ноль, если поток не является {@linkplain Thread # прерывание прервано}.
*
* <p>Если текущий счетчик равен нулю, этот метод сразу возвращается.
*
*
Если текущий счетчик больше нуля, то текущий
* поток становится отключенным для целей планирования потока и лжи
* бездействует, пока не произойдет одно из двух:
*
- Счет достигает нуля из-за вызовов
* {@link #countDown} метод; или же
*
- Какой-то другой поток {@linkplain Thread # прерывания прерывания}
* текущая тема.
*
*
*
Если текущий поток:
*
- имеет статус прерывания, установленный при входе в этот метод; или же
*
- во время ожидания {@linkplain Thread # interrupt interruptpted}
*
* затем {@link InterruptedException} выбрасывается и текущий поток
* Прерванный статус очищается.
*
* @throws InterruptedException, если текущий поток прерывается
* во время ожидания
* /
public void await () выдает InterruptedException {
sync.acquireSharedInterruptibly (1);
}
public void reset () {
sync.reset ();
}
/ **
* Заставляет текущий поток ждать, пока защелка не обратится к* ноль, если поток не является {@linkplain Thread # прерывание прервано},
* или указанное время ожидания истекло.
*
*
Если текущий счетчик равен нулю, этот метод сразу возвращается
* со значением {@code true}.
*
*
Если текущий счетчик больше нуля, то текущий
* поток становится отключенным для целей планирования потока и лжи
* бездействует, пока не произойдет одно из трех
*
- Счет достигает нуля из-за вызовов
* {@link #countDown} метод; или же
*
- Какой-то другой поток {@linkplain Thread # прерывания прерывания}
* текущий поток; или же
*
- Указанное время ожидания истекло.
*
*
*
Если счетчик достигает нуля, метод возвращается с
* значение {@code true}.
*
*
Если текущий поток:
*
- имеет статус прерывания, установленный при входе в этот метод; или же
*
- во время ожидания {@linkplain Thread # прерывание прервано},
*
* затем {@link InterruptedException} выбрасывается и текущий поток
* Прерванный статус очищается.
*
*
Если указанное время ожидания истекло, тогда значение {@code false}
* возвращается. Если время меньше или равно нулю, метод
* не будет ждать вообще.
*
* @param timeout максимальное время ожидания
* @param unit единица времени аргумента {@code timeout}
* @return {@code true}, если счетчик достиг нуля и {@code false}
* если время ожидания истекло до того, как счетчик достиг нуля
* @throws InterruptedException, если текущий поток прерывается
* во время ожидания
* /
public boolean await (длительный тайм-аут, единица времени)
бросает InterruptedException {
return sync.tryAcquireSharedNanos (1, unit.toNanos (timeout));
}
/ **
* Уменьшает количество защелок, освобождая все ожидающие потоки, если
* количество достигает нуля.
*
*
Если текущий счетчик больше нуля, то он уменьшается.
* Если новый счетчик равен нулю, то все ожидающие потоки повторно включаются для
* планирование потоков.
*
*
Если текущий счетчик равен нулю, то ничего не происходит.
* /
public void countDown () {
sync.releaseShared (1);
}
/ **
* Возвращает текущий счет.
*
*
Этот метод обычно используется для отладки и тестирования.
*
* @ вернуть текущий счет
* /
public long getCount () {
return sync.getCount ();
}
/ **
* Возвращает строку, идентифицирующую эту защелку, а также ее состояние.
* Состояние в скобках включает в себя строку {@code "Count ="}
* с последующим текущим счетом.
*
* @ вернуть строку, идентифицирующую эту защелку, а также ее состояние
* /
public String toString () {
return super.toString () + "[Count =" + sync.getCount () + "]";
}
}