以下代码线程安全

我有一个场景,我必须维护一个可以由多个线程填充的Map,每个线程都修改相应的List(唯一标识符/键是线程名称),当线程的列表大小超过固定批量大小时,我们必须保持DB中的记录.

示例代码如下:

private volatile ConcurrentHashMap>  instrumentMap = new ConcurrentHashMap>();
private ReadWriteLock lock ;

public void addAll(List entityList, String threadName) {
    try {
        lock.readLock().lock();
        List instrumentList = instrumentMap.get(threadName);
        if(instrumentList == null) {
            instrumentList = new ArrayList(batchSize);
            instrumentMap.put(threadName, instrumentList);
        }

        if(instrumentList.size() >= batchSize -1){
            instrumentList.addAll(entityList);
            recordSaver.persist(instrumentList); 
            instrumentList.clear();
        } else {
            instrumentList.addAll(entityList);  
        }
    } finally {
        lock.readLock().unlock();
    }

}

每隔2分钟就会再运行一个单独的线程来保存Map中的所有记录(以确保每隔2分钟后我们会持续存在一些内容并且地图大小不会太大)并且当它启动时它会阻止所有其他线程(检查readLock和writeLock usawhere writeLock具有更高的优先级)

if(//Some condition) {
                    Thread.sleep(//2 minutes);
                    aggregator.getLock().writeLock().lock();
                    List instrumentList = instrumentMap .values().stream().flatMap(x->x.stream()).collect(Collectors.toList());
                    if(instrumentList.size() > 0) {

                        saver.persist(instrumentList);
                        instrumentMap .values().parallelStream().forEach(x -> x.clear());
                    aggregator.getLock().writeLock().unlock();
                }

这个解决方案几乎适用于我们测试的每个场景,除非有时候我们看到一些记录丢失了,尽管它们在Map中添加得很好但根本没有保留

我的问题是这段代码有什么问题？ConcurrentHashMap不是最好的解决方案吗？读/写锁的使用在这里有问题吗？我应该顺序处理吗？

不,这不是线程安全的.

问题是您正在使用ReadWriteLock 的读锁定.这并不保证进行更新的独占访问权限.您需要使用写锁定.

但是你根本不需要使用单独的锁.您可以简单地使用该ConcurrentHashMap.compute方法:

instrumentMap.compute(threadName, (tn, instrumentList) -> {
  if (instrumentList == null) {
    instrumentList = new ArrayList<>();
  }

  if(instrumentList.size() >= batchSize -1) {
    instrumentList.addAll(entityList); 
    recordSaver.persist(instrumentList); 
    instrumentList.clear();
  } else {
    instrumentList.addAll(entityList);
  }

  return instrumentList;
});

这允许您更新列表中的项目,同时还保证对给定键的列表的独占访问权限.

我怀疑你可以将compute调用拆分为computeIfAbsent(如果不存在则添加列表)然后a computeIfPresent(更新/持久化列表):这里不需要这两个操作的原子性.但是将它们分开并没有真正的意义.

此外,instrumentMap几乎可以肯定不应该是不稳定的.除非你真的想重新分配它的价值(给定这个代码,我怀疑),删除volatile并使其成为最终.

同样,非最终锁也是有问题的.如果你坚持使用锁定,也要做到最后.