ConcurrentHashMapを使用したキューのスレッド安全性

Question

import java.util.Map; 
import java.util.concurrent.ConcurrentHashMap; 
import java.util.concurrent.Future; 
import org.slf4j.Logger; 
import org.slf4j.LoggerFactory; 


public class DedupingQueue<E> implements QueueWrapper<E> { 
    private static final Logger LOGGER = LoggerFactory.getLogger(DedupingQueue.class); 

private final Map<E, Future<E>> itemsBeingWorkedOn = new ConcurrentHashMap<>(); 
private final AsyncWorker<E> asyncWorker; //contains async method backed by a thread pool 

public DedupingQueue(AsyncWorker<E> asyncWorker) { 
    this.asyncWorker = asyncWorker; 
} 

@Override 
public Future<E> submit(E e) { 
    if (!itemsBeingWorkedOn.containsKey(e)) { 
     itemsBeingWorkedOn.put(e, asyncWorker.executeWorkAsync(e, this)); 
    } else { 
     LOGGER.debug("Rejected [{}] as it's already being worked on", e); 
    } 
    return itemsBeingWorkedOn.get(e); 
} 

@Override 
public void complete(E e) { 
    LOGGER.debug("Completed [{}]", e); 
    itemsBeingWorkedOn.remove(e); 
} 

@Override 
public void rejectAndRetry(E e) { 
    itemsBeingWorkedOn.putIfAbsent(e, asyncWorker.executeWorkAsync(e, this)); 
} 

} 

私は上記のコードのスレッド安全性を推論するいくつかの困難を抱えています。

私はマップがスレッドセーフなので、completeとrejectAndretryは完全にスレッドセーフであると考えます。しかし、submitについては、AsyncWorkerそれ自体はスレッドセーフではありません。また、どのようにして​​（ConcurrentHashMapの組み込み保証を使用して）を使用せずに、スレッドを最も効率的な方法で安全にすることができますか？

Answer 1

がそこには現在のマッピングません、とnullが返された場合、それをマップしておく、または削除する値を返す場合、ファンクションキー、現在の値またはnullを取るcompute()方法、あります。

@Override 
public Future<E> submit(E e) { 
    return itemsBeingWorkedOn.compute(e, (k, v) -> { 
     if (v == null) { 
      return asyncWorker.executeWorkAsync(k, this); 
     } else { 
      LOGGER.debug("Rejected [{}] as it's already being worked on", k); 
      return v; 
     } 
    }); 
} 

---

あなたのキューを発注されていないことに注意してください。 FIFOまたはLIFO注文が必要な場合は、LinkedHashMapと​​のようなより適切なデータ構造を使用する必要があります。

Answer 2

submit()あなたのcheck-then-actシーケンスはアトミックではないため、スレッドセーフではありません。あるスレッドの特定のキーの状態は、チェックの直後と呼び出しの前に他のスレッドによって変更することができます。この問題を解決するには、ConcurrentHashMapのアトミックメソッド（おそらくcomputeIfAbsent()）を使用します。

if (!itemsBeingWorkedOn.containsKey(e)) { 
    itemsBeingWorkedOn.put(e, asyncWorker.executeWorkAsync(e, this)); 
} 

AsyncWorkerがスレッドセーフでない場合、どのクラス全体もスレッドセーフではありません。しかし、それはAsyncWorkerソースコードなしではそれを理由づけるのは難しいです。