Потокобезопасные коллекции

Обычные ArrayList и HashMap не рассчитаны на работу из нескольких потоков: при одновременной записи данные теряются или портятся. Можно защищать их замками, но удобнее взять потокобезопасные коллекции из пакета java.util.concurrent. Они сами синхронизируют доступ и делают это эффективнее, чем один общий замок.

ConcurrentHashMap

ConcurrentHashMap — словарь, с которым могут одновременно работать много потоков. Метод merge атомарно обновляет значение: если ключа нет, кладёт начальное, если есть — объединяет старое и новое:

void main() throws InterruptedException {
    Map<String, Integer> votes = new ConcurrentHashMap<>();
    List<String> ballots = List.of("чай", "кофе", "чай", "какао", "кофе", "чай");
 
    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));
}

Результат:

{какао=1, кофе=2, чай=3}

Каждый голос учитывается в отдельном потоке, но ни один не потерялся. TreeMap здесь нужен только для вывода ключей по алфавиту.

Атомарные методы словаря

У ConcurrentHashMap есть и другие методы, которые выполняют «проверить и изменить» за один шаг: putIfAbsent, computeIfAbsent, compute. Например, computeIfAbsent создаёт значение, только если ключа ещё нет:

void main() {
    Map<String, List<String>> groups = new ConcurrentHashMap<>();
 
    groups.computeIfAbsent("бэкенд", k -> new CopyOnWriteArrayList<>()).add("Олег");
    groups.computeIfAbsent("бэкенд", k -> new CopyOnWriteArrayList<>()).add("Ира");
    groups.computeIfAbsent("фронтенд", k -> new CopyOnWriteArrayList<>()).add("Анна");
 
    IO.println(new TreeMap<>(groups));
}

Результат:

{бэкенд=[Олег, Ира], фронтенд=[Анна]}

CopyOnWriteArrayList

Если перебирать обычный ArrayList и одновременно его менять, будет ConcurrentModificationException — даже в одном потоке. CopyOnWriteArrayList при каждом изменении создаёт новую копию массива, поэтому перебор идёт по «снимку» и не ломается:

void main() {
    List<String> listeners = new CopyOnWriteArrayList<>(List.of("лог", "почта"));
 
    for (String listener : listeners) {
        IO.println("Уведомляю: " + listener);
        listeners.add("смс");
    }
 
    IO.println(listeners);
}

Результат:

Уведомляю: лог
Уведомляю: почта
[лог, почта, смс, смс]

Цикл прошёл только по двум исходным элементам. Копирование при каждой записи дорого, поэтому такой список подходит, когда читают часто, а меняют редко — например, для списка подписчиков на события.

BlockingQueue

BlockingQueue — очередь, которая умеет ждать. put добавляет элемент, а take забирает его или ждёт, пока элемент появится. На ней строят схему производитель — потребитель: один поток создаёт задания, другой их обрабатывает:

void main() throws InterruptedException {
    BlockingQueue<String> queue = new LinkedBlockingQueue<>();
 
    Thread consumer = Thread.ofPlatform().start(() -> {
        try {
            while (true) {
                String order = queue.take();
                if (order.equals("СТОП")) break;
                IO.println("Готовлю " + order);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    });
 
    for (String order : List.of("латте", "капучино", "эспрессо")) {
        queue.put(order);
    }
    queue.put("СТОП");
    consumer.join();
    IO.println("Смена окончена");
}

Результат:

Готовлю латте
Готовлю капучино
Готовлю эспрессо
Смена окончена

Очередь выдаёт элементы в порядке добавления, поэтому заказы готовятся по очереди. Специальное значение «СТОП» сообщает потребителю, что заданий больше не будет.

Частая ошибка

Потокобезопасна каждая отдельная операция, но не их сочетание. Проверка containsKey и последующий put — это два шага, и между ними другой поток успевает изменить словарь:

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

Результат (число меняется от запуска к запуску, но почти всегда меньше 200 000):

Заменяйте такие пары на один атомарный метод: counter.merge("клики", 1, Integer::sum) всегда даст 200 000.

Спасибо, что читаете!

ConcurrentHashMap

Атомарные методы словаря

CopyOnWriteArrayList

BlockingQueue

Частая ошибка

Спасибо, что читаете!