Concurrent collections

The ordinary ArrayList and HashMap aren't built for use from several threads: with simultaneous writes, data gets lost or corrupted. You could guard them with locks, but it's easier to take the thread-safe collections from the java.util.concurrent package. They synchronize access themselves and do it more efficiently than a single shared lock.

ConcurrentHashMap

ConcurrentHashMap is a map that many threads can work with at once. The merge method updates a value atomically: if the key is missing it puts the initial value, if it exists it combines the old and new values:

void main() throws InterruptedException {
    Map<String, Integer> votes = new ConcurrentHashMap<>();
    List<String> ballots = List.of("tea", "coffee", "tea", "cocoa", "coffee", "tea");
 
    List<Thread> threads = new ArrayList<>();
    for (String choice : ballots) {
        threads.add(Thread.ofPlatform().start(() -> votes.merge(choice, 1, Integer::sum)));
    }
    for (Thread t : threads) {
        t.join();
    }
 
    IO.println(new TreeMap<>(votes));
}

Output:

{cocoa=1, coffee=2, tea=3}

Each vote is counted in a separate thread, yet none is lost. The TreeMap is only there to print the keys in alphabetical order.

Atomic map methods

ConcurrentHashMap has other methods that do "check and change" in a single step: putIfAbsent, computeIfAbsent, compute. For example, computeIfAbsent creates a value only if the key isn't there yet:

void main() {
    Map<String, List<String>> groups = new ConcurrentHashMap<>();
 
    groups.computeIfAbsent("backend", k -> new CopyOnWriteArrayList<>()).add("Oleg");
    groups.computeIfAbsent("backend", k -> new CopyOnWriteArrayList<>()).add("Ira");
    groups.computeIfAbsent("frontend", k -> new CopyOnWriteArrayList<>()).add("Anna");
 
    IO.println(new TreeMap<>(groups));
}

Output:

{backend=[Oleg, Ira], frontend=[Anna]}

CopyOnWriteArrayList

If you iterate over an ordinary ArrayList and change it at the same time, you get a ConcurrentModificationException — even in a single thread. CopyOnWriteArrayList makes a new copy of its array on every change, so iteration goes over a "snapshot" and doesn't break:

void main() {
    List<String> listeners = new CopyOnWriteArrayList<>(List.of("log", "email"));
 
    for (String listener : listeners) {
        IO.println("Notifying: " + listener);
        listeners.add("sms");
    }
 
    IO.println(listeners);
}

Output:

Notifying: log
Notifying: email
[log, email, sms, sms]

The loop went over only the two original elements. Copying on every write is expensive, so such a list suits cases where reads are frequent and changes are rare — a list of event subscribers, for example.

BlockingQueue

BlockingQueue is a queue that can wait. put adds an element, and take removes one or waits until one appears. It is the basis of the producer-consumer pattern: one thread creates jobs, another processes them:

void main() throws InterruptedException {
    BlockingQueue<String> queue = new LinkedBlockingQueue<>();
 
    Thread consumer = Thread.ofPlatform().start(() -> {
        try {
            while (true) {
                String order = queue.take();
                if (order.equals("STOP")) break;
                IO.println("Making " + order);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    });
 
    for (String order : List.of("latte", "cappuccino", "espresso")) {
        queue.put(order);
    }
    queue.put("STOP");
    consumer.join();
    IO.println("Shift is over");
}

Output:

Making latte
Making cappuccino
Making espresso
Shift is over

The queue hands out elements in the order they were added, so the orders are made one after another. The special value "STOP" tells the consumer there will be no more jobs.

Common mistake

Each individual operation is thread-safe, but a combination of them is not. A containsKey check followed by put is two steps, and another thread can change the map in between:

void main() throws InterruptedException {
    Map<String, Integer> counter = new ConcurrentHashMap<>();
 
    Runnable task = () -> {
        for (int i = 0; i < 100_000; i++) {
            if (counter.containsKey("clicks")) {
                counter.put("clicks", counter.get("clicks") + 1);
            } else {
                counter.put("clicks", 1);
            }
        }
    };
 
    Thread a = new Thread(task);
    Thread b = new Thread(task);
    a.start();
    b.start();
    a.join();
    b.join();
 
    IO.println(counter.get("clicks"));
}

Output (the number changes from run to run, but is almost always below 200,000):

Replace such pairs with a single atomic method: counter.merge("clicks", 1, Integer::sum) always gives 200,000.

Thanks for reading!

ConcurrentHashMap

Atomic map methods

CopyOnWriteArrayList

BlockingQueue

Common mistake

Thanks for reading!