How to sync multi thread map update

Question

I'm wondering how to sync the following multi thread so that I can still run the threads in parallel without causing multiple objects to be saved.

public void setupRender() {
    ExecutorService executorService = Executors.newFixedThreadPool(10);

    final Map<Integer, String> map = Collections.synchronizedMap(new HashMap<Integer, String>());

    for (int i = 0; i < 10; i++) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                parse(map); 
            }
        });
        System.out.println("map size " + map.size() + " loop " + i);
    }
}

public void parse(Map<Integer, String> map) {
    for (int j = 0; j < 100; j++) {
        if (map.containsKey(j)) {
            System.out.println("Update");
            //session.update(map.getKey(j));
        } else {
            System.out.println("add to map " + j);
            String obj = "test";
            map.put(j, obj);
            System.out.println("save");
            //session.save(obj);
        }

        if (j % 50 == 0) {
            System.out.println("commit");
            //tx.commit();
        }
    }
}

Outcome, notice how multiple objects of the same key are added to the map, but worse yet saved to the database resulting in duplicate database enteries.

map size 0 loop 0
map size 0 loop 1
add to map 0
commit
add to map 1
add to map 0
commit
add to map 2
add to map 3
add to map 4
add to map 2
add to map 5
add to map 6
map size 1 loop 2
add to map 7
add to map 8
add to map 9
commit
map size 10 loop 3
commit
add to map 5

Answer 1

You can use ConcurrentHashMap and putIfAbsent to guarantee consistency and performance. In this case your session updates can run simultaneously.

putIfAbsent:
    - returns the previous value associated with the specified key, 
      or null if there was no mapping for the key

public void parse(ConcurrentHashMap<Integer, String> map) {
    for (int j = 0; j < 100; j++) {    
        String obj = "test";

        Object returnedValue = map.putIfAbsent(j, obj);

        boolean wasInMap = returnedValue != null;

        if (wasInMap) {
            System.out.println("Update");
            //session.update(map.getKey(j));
        } else {
            System.out.println("add to map " + j);

            map.put(j, obj);
            System.out.println("save");
            //session.save(obj);
        }
    }
}

You can also put there a callback which would create a new instance conditionally:

class Callback{
    abstract void saveOrUpdate(boolean wasInMap);    
}

public void parse(Map<Integer, String> map) {
    for (int j = 0; j < 100; j++) {    
        String obj = "test";

        Callback callback = (wasInMap) -> { //Java 8 syntax to be short
            if (wasInMap) {
                System.out.println("Update");
                //session.update(map.getKey(j));
            } else {
                System.out.println("add to map " + j);

                map.put(j, obj);
                System.out.println("save");
                //session.save(obj);
            }
        }

        Object returnedValue = map.putIfAbsent(j, callback);

        callback.saveOrUpdate(returnedValue != null);
    }
}

Answer 2

Use the syncronized keyword to ensure that checking if j is in the map and adding it if not, is done atomically. However, it defeats the purpose of multithreading as most threads will be blocked while one thread is looping.

public void setupRender() {
    ExecutorService executorService = Executors.newFixedThreadPool(10);

final Map<Integer, String> map = Collections.synchronizedMap(new HashMap<Integer, String>());

for (int i = 0; i < 10; i++) {
    executorService.execute(new Runnable() {
        @Override
        public void run() {
            parse(map); 
        }
    });
    System.out.println("map size " + map.size() + " loop " + i);
}
}

public void parse(Map<Integer, String> map) {
    synchronized(map) {
    for (int j = 0; j < 100; j++) {
        if (map.containsKey(j)) {
            System.out.println("Update");
            //session.update(map.getKey(j));
        } else {
            System.out.println("add to map " + j);
            String obj = "test";
            map.put(j, obj);
            System.out.println("save");
            //session.save(obj);
        }
        }

        if (j % 50 == 0) {
            System.out.println("commit");
            //tx.commit();
        }
    }
}