Java – the following code is thread safe

I have a scenario. I must maintain a map that can be filled by multiple threads. Each thread modifies the corresponding list (the unique identifier / key is the thread name). When the list size of threads exceeds the fixed batch size, we must maintain the records in the DB

The example code is as follows:

private volatile ConcurrentHashMap<String,List<T>>  instrumentMap = new ConcurrentHashMap<String,List<T>>();
private ReadWriteLock lock ;

public void addAll(List<T> entityList,String threadName) {
    try {
        lock.readLock().lock();
        List<T> instrumentList = instrumentMap.get(threadName);
        if(instrumentList == null) {
            instrumentList = new ArrayList<T>(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();
    }

}

A separate thread will be run every 2 minutes to save all records in the map (to ensure that we will continue to have some content every 2 minutes and the map size will not be too large) and it will block all other threads when it starts (check readlock and writelock usahere writelock have higher priority)

if(//Some condition) {
                    Thread.sleep(//2 minutes);
                    aggregator.getLock().writeLock().lock();
                    List<T> 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();
                }

This solution is applicable to almost every scenario we test, unless sometimes we see that some records are lost. Although they are well added in the map, they are not retained at all

My question is what's wrong with this code? Isn't concurrent HashMap the best solution? Is there a problem with the use of read / write locks here? Should I handle it in sequence?

Solution

No, it's not thread safe

The problem is that you are using readwritelock's read lock This does not guarantee exclusive access to updates You need to use write locking

But you don't need a separate lock at all You just need to use concurrenthashmap Compute method:

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

This allows you to update items in the list while also guaranteeing exclusive access to the list for a given key

I suspect you can split the calculation call into computeifabsent (add list if it doesn't exist) and then computeifpresent (update / persist list): atomicity of these two operations is not required here But there is no real point in separating them

In addition, instrumentmap should almost certainly not be volatile Unless you really want to redistribute its value (given this code, I doubt), delete volatile and make it final

Similarly, non - Final locks are problematic If you insist on using locking, do it last