Elastyczne CountDownLatch?

Question

Napotkałem dwukrotnie problem polegający na tym, że wątek producenta tworzy N elementów roboczych, przesyła je do ExecutorService, a następnie musi poczekać, aż wszystkie N zostaną przetworzone.

Ostrzeżenia

• N nie jest znana z góry. Gdyby tak było, po prostu utworzyłbym CountDownLatch, a następnie miał wątek producenta await(), aż wszystkie prace zostały zakończone.
• Używanie CompletionService jest niewłaściwe, ponieważ mimo, że mój wątek producent musi blokować (tzn wywołując take()) nie żaden sposób sygnalizowania, że ​​wszystkie prace są kompletne, aby spowodować wątek producent przestanie czekać.

My obecne rozwiązanie preferowane jest użycie licznika całkowitą oraz przyrostu tym, gdy element pracy jest złożony i ubytek, gdy element roboczy jest przetwarzany. Po przejściu wszystkich zadań N mój wątek producenta będzie musiał poczekać na blokadę, sprawdzając, czy counter == 0, gdy jest powiadomiony. Gwinty konsumenckie będą musiały powiadomić producenta, jeśli zmniejszyło licznik, a nowa wartość to 0.

Czy istnieje lepsze podejście do tego problemu lub czy istnieje odpowiednia konstrukcja w java.util.concurrent Powinienem raczej użyć niż "toczyć własne"?

Z góry dziękuję.

Answer 1

java.util.concurrent.Phaser wygląda na to, że będzie dobrze działać. Planowane jest wydanie na platformie Java 7, ale najbardziej stabilną wersję można znaleźć na stronie grupy zainteresowań jsr166.

Fazer to uwielbiona Cykliczna Bariera. Możesz zarejestrować liczbę stron N i kiedy będziesz gotowy czekać na ich awans w konkretnej fazie.

Szybki przykład jak to działa:

final Phaser phaser = new Phaser(); 

public Runnable getRunnable(){ 
    return new Runnable(){ 
     public void run(){ 
      ..do stuff... 
      phaser.arriveAndDeregister(); 
     } 
    }; 
} 
public void doWork(){ 
    phaser.register();//register self 
    for(int i=0 ; i < N; i++){ 
     phaser.register(); // register this task prior to execution 
     executor.submit(getRunnable()); 
    } 
    phaser.arriveAndAwaitAdvance(); 
} 

Answer 2

Można oczywiście użyć CountDownLatch chroniony przez AtomicReference tak, że zadania się owinięty w ten sposób:

public class MyTask extends Runnable { 
    private final Runnable r; 
    public MyTask(Runnable r, AtomicReference<CountDownLatch> l) { this.r = r; } 

    public void run() { 
     r.run(); 
     while (l.get() == null) Thread.sleep(1000L); //handle Interrupted 
     l.get().countDown(); 
    } 
} 

Wskazówka że zadania uruchomić swoją pracę, a następnie wirowania aż odliczanie jest set (tzn. znana jest całkowita liczba zadań). Po ustawieniu odliczania odliczają i znikają. Należą się składać w następujący sposób:

AtomicReference<CountDownLatch> l = new AtomicReference<CountDownLatch>(); 
executor.submit(new MyTask(r, l)); 

Po momencie jej tworzenia/złożenia pracy, gdy wiesz, ile zadań utworzono:

latch.set(new CountDownLatch(nTasks)); 
latch.get().await(); 

Answer 3

I zakładamy, że producent nie nie musi wiedzieć, kiedy kolejka jest pusta, ale musi wiedzieć, kiedy ostatnie zadanie zostało zakończone.

Dodałbym do konsumenta metodę waitforWorkDone(producer). Producent może dodać swoje N zadań i wywołać metodę wait. Metoda wait blokuje wątek przychodzący, i kolejka robocza nie jest pusta i nie są wykonywane żadne zadania.

Nitki konsumenckie notifyAll() w funkcji waitfor lock iff jej zadanie zostało zakończone, kolejka jest pusta i żadne inne zadanie nie jest wykonywane.

Answer 4

Użyłem się ExecutorCompletionService na coś takiego:

ExecutorCompletionService executor = ...; 
int count = 0; 
while (...) { 
    executor.submit(new Processor()); 
    count++; 
} 

//Now, pull the futures out of the queue: 
for (int i = 0; i < count; i++) { 
    executor.take().get(); 
} 

Wiąże utrzymując kolejkę zadań, które zostały złożone, więc jeśli lista jest dowolnie długo, metoda może być lepiej.

Ale upewnij się, że do koordynacji używasz AtomicInteger, dzięki czemu będziesz mógł go zwiększyć w jednym wątku i zmniejszyć go w wątkach roboczych.

Answer 5

Czy bierzesz pod uwagę tego:

https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ExecutorService.html#invokeAll(java.util.Collection)

Oto przykład napisałem: https://github.com/moulaali/java_fun/blob/master/src/WaitForAllConcurrencyExample.java

Answer 6

Co ty opisany jest dość podobny do używania standardowego semafora, ale używany jest "wstecz".

• Twój semaforów zaczyna się od 0 zezwoleń
• Każda jednostka pracy uwalnia jedno zezwolenie, gdy zakończy
• zablokować czekając nabyć N pozwala

Plus masz elastyczność tylko nabycia M < N zezwoleń, które są przydatne, jeśli chcesz sprawdzić stan pośredni. Na przykład testuję asynchroniczną, ograniczoną kolejkę komunikatów, więc oczekuję, że kolejka będzie pełna dla jakiegoś M < N, więc mogę pobrać M i sprawdzić, czy kolejka jest rzeczywiście pełna, a następnie uzyskać pozostałe pozwolenia N - M po pobraniu wiadomości od kolejka.