Java Thread-sichere Übergabe von Sammlungsobjekten von einem Thread zum anderen

Question

Ich habe eine Java-Anwendung, die Worker-Threads hat, um Jobs zu verarbeiten. Ein Arbeiter erzeugt ein Ergebnisobjekt, so etwas sagen wie:

class WorkerResult{ 
    private final Set<ResultItems> items; 
    public Worker(Set<ResultItems> pItems){ 
     items = pItems; 
    } 
} 

Wenn der Arbeiter beendet, es tut diese Operation:

... 
final Set<ResultItems> items = new SomeNonThreadSafeSetImplSet<ResultItems>(); 
for(Item producedItem : ...){ 
     items.add(item); 
} 
passToGatherThread(items); 

Der items Satz ist eine Art "unit-of-Arbeit" hier . Die Methode passToGatherThread übergibt die items-Gruppe an einen Sammel-Thread, von dem zur Laufzeit nur einer existiert.

Synchronisation ist hier nicht erforderlich, da Race-Bedingungen nicht auftreten können, weil nur ein Thread (Gather-Thread) den items-Satz liest. AFAICS, der Gather-Thread sieht möglicherweise nicht alle Elemente, weil das Set nicht Thread-sicher ist, oder?

Angenommen, ich kann passToGatherThread nicht synchronisieren, sagen, weil es eine 3rd-Party-Bibliothek ist. Was ich grundsätzlich befürchte ist, dass der Sammel-Thread nicht alle Elemente wegen Caching, VM-Optimierungen, etc. sieht. Hier kommt die Frage: Wie man die Elemente thread-sicher weitergibt, so dass der Gather-Thread "sieht" die richtige Menge von Gegenständen?

Answer 1

Ich habe darüber nachgedacht, (und diskutiert) diese Frage viel und ich habe mit einer anderen Antwort kommen, die, wie ich hoffe, die beste Lösung sein.

Das Überschreiten einer synchronisierten Sammlung ist in Bezug auf die Effizienz nicht gut, da jede nachfolgende Operation in dieser Sammlung synchronisiert wird. Wenn es viele Operationen gibt, kann sich dies als Handicap erweisen.

Auf den Punkt gebracht: lassen Sie uns einige Annahmen getroffen werden (was ich nicht zustimmen):

• die erwähnte passToGatherThread Verfahren in der Tat unsicher ist jedoch unwahrscheinlich es scheint
• Compiler können die Ereignisse im Code neu anordnen so dass die passToGatherThread vor der Sammlung gefüllt
genannt wird

die einfachste, sauberste und möglicherweise der effizienteste Weg, um sicherzustellen, dass die zu gatherer Methode übergebene Sammlung ist fertig und abgeschlossen ist t o setzen die Sammlung Push in einem synchronisierten Block, wie folgt aus:

synchronized(items) { 
    passToGatherThread(items); 
} 

diese Weise können wir eine Speicher Synchronisation gewährleisten und eine gültige passieren-vor-Sequenz vor der Sammlung übergeben wird, damit sicher, dass alle Objekte korrekt übergeben werden.

Answer 2

Es scheint hier kein Synchronisationsproblem zu geben. Sie erstellen für jedes passToGatherThread ein neues Set-Objekt, und zwar nach dem Ändern des Sets. Keine Objekte werden verloren gehen.

Auf Set (und die meisten Java-Sammlungen) kann gleichzeitig mit vielen Threads zugegriffen werden, sofern keine Änderungen an der Auflistung vorgenommen werden. Das ist, was Collections.unmodifiableCollection ist.

Da die erwähnte Methode passToGatherThread als Kommunikation mit einem anderen Thread dient, muss sie eine Art von Synchronisation verwenden - und jede Synchronisation gewährleistet die Konsistenz des Speichers zwischen den Threads.

Auch - bitte beachten Sie, dass alle Schreibvorgänge auf die Objekte in der übergebenen Sammlung gemacht werden, bevor es an den anderen Thread übergeben wird. Selbst wenn der Speicher in den lokalen Cache des Threads kopiert wird, hat er denselben unveränderten Wert wie der andere Thread.

Answer 3

Sie können einfach eine der Thread-sicheren Implementierungen von Set verwenden, die Java für Ihre WorkerResult bereitstellt. Siehe zum Beispiel:

• CopyOnWriteArraySet
• ConcurrentSkipListSet

Eine weitere Option ist Collections.synchronizedSet() zu verwenden.

Answer 4

Der Arbeiter aufrufbar implementiert und gibt WorkerResult:

class Worker implements Callable<WorkerResult> { 
    private WorkerInput in; 

    public Worker(WorkerInput in) { 
     this.in = in; 
    } 

    public WorkerResult call() { 
     // do work here 
    } 
} 

Dann verwenden wir ein ExecutorService den Thread-Pool zu verwalten, und sammeln Sie die Ergebnisse über zukünftige Verwendung.

public class PooledWorkerController { 

    private static final int MAX_THREAD_POOL = 3; 
    private final ExecutorService pool = 
     Executors.newFixedThreadPool(MAX_THREAD_POOL); 

    public Set<ResultItems> process(List<WorkerInput> inputs) 
      throws InterruptedException, ExecutionException{   
     List<Future<WorkerResult>> submitted = new ArrayList<>(); 
     for (WorkerInput in : inputs) { 
      Future<WorkerResult> future = pool.submit(new Worker(in)); 
      submitted.add(future); 
     } 
     Set<ResultItems> results = new HashSet<>(); 
     for (Future<WorkerResult> future : submitted) { 
      results.addAll(future.get().getItems()); 
     } 
     return results; 
    } 
}