あるスレッドから別のスレッドにコレクションオブジェクトをJavaスレッドで安全に渡す

Question

ジョブを処理するワーカースレッドを持つJavaアプリケーションがあります。労働者は、結果オブジェクトを生成し、のようなものを言う：

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

作業員が終了すると、それはこの操作を行います。

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

itemsセットは「作業単位」ここの一種であります。 passToGatherThreadメソッドは、itemsセットを収集スレッドに渡します。そのうちの1つのみが実行時に存在します。

ここで同期化は必要ありません。これは、1つのスレッド（ギャザースレッド）のみがitemsのセットを読み取るため、競合状態が発生しないためです。 AFAICSでは、ギャザースレッドは、スレッドセーフではないため、すべてのアイテムを表示しない場合があります。

passToGatherThreadは、サードパーティのライブラリであるため、同期させることはできません。私は基本的に恐れているのは、キャッシング、VMの最適化などのために、ギャザースレッドがすべてのアイテムを見ることができないということです。ここでは、スレッドセーフな方法でアイテムを渡す方法です。ギャザースレッドは、適切な項目のセット？

Answer 1

私が考えた（と議論）しているこの質問に多く、私は、私は願って、最適なソリューションとなり、別の答え、が出ています。

同期コレクションを渡すことは、コレクションの後続の各操作が同期されるため、効率的ではありません。多くの操作がある場合は、ハンディキャップである可能性があります。ポイントへの

：さんは（私は同意しない）いくつかの仮定をしてみましょう：

• 言及passToGatherThread方法は確かに安全ではありません、しかし、ありそうもないことが
• コンパイラは、コード内のイベントを並べ替えることができそうですコレクションは

採集メソッドに渡されたコレクションの準備ができていることを保証するための最も簡単な、クリーンそしておそらく最も効率的な方法を満たし、完全をtされる前passToGatherThreadが呼び出されるようにコレクションが渡される前に、我々はこのようにすべてのオブジェクトが正しく渡されていることを確認して、メモリの同期化と有効起こる-前にシーケンスを確保するその方法

synchronized(items) { 
    passToGatherThread(items); 
} 

：Oこのように、同期ブロックに収集プッシュを置きます。

Answer 2

ここで同期の問題はないようです。 passToGatherThreadごとに新しいSetオブジェクトを作成し、セットを変更した後に行います。オブジェクトは失われません。

セット（およびほとんどのJavaコレクション）は、コレクションに変更を加えない限り、多数のスレッドによって同時にアクセスできます。それはCollections.unmodifiableCollectionのためです。

この方法は他のスレッドと通信するため、ある種の同期を使用しなければならず、各同期によってスレッド間のメモリの一貫性が保証されます。

また、渡されたコレクション内のオブジェクトへのすべての書き込みは、の前に行われ、の前に行われます。他のスレッドに渡されます。メモリがスレッドのローカルキャッシュにコピーされても、他のスレッドと同じ変更されていない値を持ちます。

Answer 3

WorkerResultにJavaが提供するSetのスレッドセーフ実装のいずれかを使用できます。例えば参照：

• CopyOnWriteArraySet
• ConcurrentSkipListSet

別のオプションはCollections.synchronizedSet()を使用することです。

Answer 4

労働者が呼び出し可能な実装とWorkerResultを返します。

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

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

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

その後、我々は、スレッドプールを管理するためにExecutorServiceのを使用して、将来を使用して経由で結果を収集。

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