java - 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 是完全线程安全的，因为 map 是线程安全的。但是考虑到 AsyncWorker 本身不是线程安全的，那么 submit 又如何呢？另外，如何在不使用同步的情况下以最有效的方式使其线程安全(使用 ConcurrentHashMap 的内置保证)？

Answer 1

有 compute() 方法，该函数采用键、当前值或 null(如果没有当前映射)，并返回要保留在映射中的值，或者如果返回 null 则将其删除。

@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 顺序，则应该使用更合适的数据结构，例如带有 synchronized 语句的 LinkedHashMap。