Java Concurrency

Question

I'm trying to implement some kind of accumulation logic in a multi threaded environment; I’m wondering is there any better/faster way to do it without the lock and synchronized keyword? The following is my current code:

public class ConcurrentHashMapTest {

    private static final int NB_THREADS = 1_000;

    private final Map<String, Integer> cnts = new HashMap<>();

    private static final Lock RWLOCK = new ReentrantLock(true);

    private static final String[] NAMES = {
        "A", "B"
    };

    public void testIt() {
        ExecutorService executor =
            Executors.newFixedThreadPool(NB_THREADS);
        for (int i = 0; i < NB_THREADS; i++) {
            Runnable task = new WorkerThread();
            executor.submit(task);
        }
        executor.shutdown();
        try {
            executor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println(cnts);
    }

    private void accumulate(String name) {
        RWLOCK.lock();
        try {
            Integer cnt = cnts.get(name);
            if (cnt == null) {
                cnts.put(name, 1);
            } else {
                cnts.put(name, cnt + 1);
            }
        } finally {
            RWLOCK.unlock();
        }
    }

    private class WorkerThread implements Runnable {
        @Override
        public void run() {
            accumulate(NAMES[ThreadLocalRandom.current().nextInt(0, NAMES.length)]);
        }
    }
} 

Answer 1

Java 8:

private final Map<String, AtomicInteger> cnts =
        new ConcurrentHashMap<>();

private void accumulate(String name) {
    cnts.computeIfAbsent(name, k -> new AtomicInteger()).incrementAndGet();
}

The ConcurrentHashMap can be freely accessed from multiple threads. The computeIfAbsent method takes a lambda to evaluate to get a value for the key if the key is not present in the map, and adds it if and only if there is no such mapping, and then returns that value. It's effectively putIfAbsent followed by get. The value is a new AtomicInteger with the value 0. Whether there was an existing value, or whether a new one with value 0 was just added, in either case increment it.

Java 7:

private final ConcurrentMap<String, AtomicInteger> cnts =
        new ConcurrentHashMap<>();

private void accumulate(String name) {
    cnts.putIfAbsent(name, new AtomicInteger());
    cnts.get(name).incrementAndGet();
}

For Java 7, there is no computeIfAbsent method, but that effectively just does a putIfAbsent followed by a get, so the same effect is achieved by calling those methods. There is no concern that the value already existed in the map; a new, zero AtomicInteger is added if and only if the map had no value for that key. Even if another thread got in there before us and added a zero, both threads would then see and increment that same AtomicInteger instance.

Answer 2

use a concurrent hash map with String and AtomicInteger. Both are thread safe and thus can be used freely.

Answer 3

I'd be wary of using fairness on your ReentrantLock in this case, as there's no benefit to your accumulator if longer waiting threads get access first. Take a look at Brian Goetz's 'Java Concurrency in Practice'

Why wouldn't we want to make all locks fair? After all, fairness is good and unfairness is bad, right? (It's not accidental that whenever kids want to appeal a decision, "that's not fair" almost certainly comes up. We think fairness is pretty important, and they know it.) In reality, the fairness guarantee for locks is a very strong one, and comes at a significant performance cost. The bookkeeping and synchronization required to ensure fairness mean that contended fair locks will have much lower throughput than unfair locks. As a default, you should set fair to false unless it is critical to the correctness of your algorithm that threads be serviced in exactly the order they queued up.

Answer 4

You could use a Map of name to AtomicInteger and use double-check locking when there is no counter in the map at all. Be aware that you need to use the volatile keyword for effective double-check locking.

This way you will only lock the whole map for actually adding brand new entries, the rest of the processing can happen in parallel.

You risk massively over-complicating your program here though and possibly even reducing performance in real-world cases. Is contention on this map really a performance bottle-neck?

Answer 5

According to Oracle Java 7 API : implementation of HashMap is not synchronized.

You can use Hashtable implementation or declare : private final Map<String, Integer> cnts = Collections.synchronizedMap(new HashMap<String, Integer>());

Answer 6

I think what you are looking for is a Multiton:

/**
 * Holds a thread-safe map of unique create-once items.
 *
 * Contract:
 *
 * Only one object will be made for each key presented.
 *
 * Thread safe.
 *
 * @author OldCurmudgeon
 * @param <K>
 * @param <V>
 */
public class Multiton<K, V> {

    // Map from the key to the futures of the items.
    private final ConcurrentMap<K, Future<V>> multitons = new ConcurrentHashMap<>();
    // The creator can create an item of type V.
    private final Creator<K, V> creator;

    public Multiton(Creator<K, V> creator) {
        this.creator = creator;
    }

    /**
     * There can be only one.
     *
     * Use a FutureTask to do the creation to ensure only one construction.
     *
     * @param key
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    public V get(final K key) throws InterruptedException, ExecutionException {
        // Already made?
        Future<V> f = multitons.get(key);
        if (f == null) {
            // Plan the future but do not create as yet.
            FutureTask<V> ft = new FutureTask<>(() -> creator.create(key));
            // Store it.
            f = multitons.putIfAbsent(key, ft);
            if (f == null) {
                // It was successfully stored - it is the first (and only)
                f = ft;
                // Make it happen.
                ft.run();
            }
        }
        // Wait for it to finish construction and return the constructed.
        return f.get();
    }

    /**
     * Returns a Map indicating the current state.
     *
     * @return a Map which should reflect the current state.
     *
     * @throws java.lang.InterruptedException
     * @throws java.util.concurrent.ExecutionException
     */
    public Map<K, V> getMap() throws InterruptedException, ExecutionException {
        Map<K, V> map = new HashMap<>();
        for (Map.Entry<K, Future<V>> e : multitons.entrySet()) {
            map.put(e.getKey(), e.getValue().get());
        }
        return map;
    }

    /**
     * User provides one of these to do the construction.
     *
     * @param <K>
     * @param <V>
     */
    public abstract static class Creator<K, V> {

        // Return a new item under the key.
        abstract V create(K key) throws ExecutionException;

    }

}

Usage - for demonstration - adds up all integers up to 999, keying on their first digit:

Multiton<String, AtomicInteger> counts = new Multiton<>(
        new Creator<String, AtomicInteger>() {

            @Override
            AtomicInteger create(String key) throws ExecutionException {
                return new AtomicInteger();
            }
        }
);

public void test() throws InterruptedException, ExecutionException {
    for (int i = 0; i < 1000; i++) {
        counts.get(Integer.toString(i).substring(0, 1)).addAndGet(i);
    }
    System.out.println(counts.getMap());
}

Prints:

{0=0, 1=15096, 2=25197, 3=35298, 4=45399, 5=55500, 6=65601, 7=75702, 8=85803, 9=95904}

Java < 8 version:

/**
 * Holds a thread-safe map of unique create-once items.
 *
 * Contract:
 *
 * Only one object will be made for each key presented.
 *
 * Thread safe.
 *
 * @author OldCurmudgeon
 * @param <K>
 * @param <V>
 */
public class Multiton<K, V> {

  // Map from the key to the futures of the items.
  private final ConcurrentMap<K, Future<V>> multitons = new ConcurrentHashMap<>();
  // The creator can create an item of type V.
  private final Creator<K, V> creator;

  public Multiton(Creator<K, V> creator) {
    this.creator = creator;
  }

  /**
   * There can be only one.
   *
   * Use a FutureTask to do the creation to ensure only one construction.
   *
   * @param key
   * @return
   * @throws InterruptedException
   * @throws ExecutionException
   */
  public V get(final K key) throws InterruptedException, ExecutionException {
    // Already made?
    Future<V> f = multitons.get(key);
    if (f == null) {
      // Plan the future but do not create as yet.
      FutureTask<V> ft = new FutureTask<>(new Callable<V>() {

        @Override
        public V call() throws Exception {
          // Doing this inline may be a little contrived but it maintains the linkage with the Java-8 version.
          return creator.create(key);
        }

      }
      );
      // Store it.
      f = multitons.putIfAbsent(key, ft);
      if (f == null) {
        // It was successfully stored - it is the first (and only)
        f = ft;
        // Make it happen.
        ft.run();
      }
    }
    // Wait for it to finish construction and return the constructed.
    return f.get();
  }

  /**
   * Returns a Map indicating the current state.
   *
   * @return a Map which should reflect the current state.
   *
   * @throws java.lang.InterruptedException
   * @throws java.util.concurrent.ExecutionException
   */
  public Map<K, V> getMap() throws InterruptedException, ExecutionException {
      Map<K, V> map = new HashMap<>();
      for (Map.Entry<K, Future<V>> e : multitons.entrySet()) {
          map.put(e.getKey(), e.getValue().get());
      }
      return map;
  }

  /**
   * User provides one of these to do the construction.
   *
   * @param <K>
   * @param <V>
   */
  public abstract static class Creator<K, V> {

    // Return a new item under the key.
    abstract V create(K key) throws ExecutionException;

  }

}