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

← назад к разделу

Стандартные коллекции Java — HashMap, ArrayList, LinkedList — не рассчитаны на работу из нескольких потоков одновременно. Как только несколько потоков начинают читать и писать в одну коллекцию, программа начинает вести себя непредсказуемо. В Java есть готовые альтернативы — и у каждой свой внутренний алгоритм потокобезопасности со своими сильными сторонами и ограничениями.

Что идёт не так с обычными коллекциями

Представьте: два потока одновременно добавляют элементы в ArrayList. Внутри ArrayList хранит массив и счётчик размера. Если оба потока прочитают счётчик одновременно (допустим, он равен 5), оба запишут новый элемент на позицию 5 — и один элемент потеряется. Счётчик при этом станет 6, хотя реально добавилось два элемента.

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

У HashMap ситуация ещё опаснее. При добавлении элемента карта иногда перестраивает внутреннюю структуру (rehash). Если два потока попадут на rehash одновременно, можно получить бесконечный цикл при обходе или потерю данных — и это в Java 7 и более ранних версиях приводило к зависанию приложений.

Типичные симптомы:

• потеря записей (добавили элемент, но он не появился);
• ArrayIndexOutOfBoundsException или ConcurrentModificationException в неожиданных местах;
• программа зависает без видимой причины.

Все потокобезопасные коллекции решают это одним из трёх способов: блокировкой (замок вокруг данных), копированием (пишем в копию, читаем без замка) или неблокирующими атомарными операциями (CAS — compare-and-swap). Разница между ними — это и есть разница в производительности и ограничениях.

ConcurrentHashMap — потокобезопасная карта

ConcurrentHashMap — основная замена HashMap в многопоточном коде. Несколько потоков могут работать с разными частями карты одновременно, не мешая друг другу.

живой пример

import java.util.concurrent.ConcurrentHashMap;

public class CounterDemo {
    public static void main(String[] args) {
        ConcurrentHashMap<String, Integer> counter = new ConcurrentHashMap<>();
        counter.put("events", 1);

        // compute: чтение старого значения и запись нового — одна неделимая операция
        counter.compute("events", (key, value) -> value == null ? 1 : value + 1);

        // merge — то же самое, но короче для счётчиков
        counter.merge("events", 1, Integer::sum);

        System.out.println(counter);
    }
}

Ключевое преимущество — методы compute, merge, computeIfAbsent выполняются атомарно: никакой другой поток не вклинится между чтением старого значения и записью нового. Это избавляет от классической ошибки «прочитали, пока не заблокировали» — вот она на глаз, в двух потоках:

живой пример

import java.util.concurrent.ConcurrentHashMap;

public class LostUpdateDemo {
    public static void main(String[] args) throws InterruptedException {
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
        map.put("getPut", 0);
        map.put("merge", 0);

        Runnable job = () -> {
            for (int i = 0; i < 50_000; i++) {
                // так не надо: между get и put вклинится другой поток
                map.put("getPut", map.get("getPut") + 1);
                // так надо: одна неделимая операция
                map.merge("merge", 1, Integer::sum);
            }
        };

        Thread a = new Thread(job);
        Thread b = new Thread(job);
        a.start();
        b.start();
        a.join();
        b.join();

        System.out.println("get + put: " + map.get("getPut") + " из 100000");
        System.out.println("merge:     " + map.get("merge") + " из 100000");
    }
}

Первая строка вывода каждый раз разная и не дотягивает до ста тысяч — это потерянные обновления. Вторая всегда ровно сто тысяч.

У этой атомарности есть цена, и знать о ней надо заранее: функция, которую вы передаёте в compute, merge или computeIfAbsent, выполняется под замком корзины. Отсюда два правила.

Первое: внутри функции нельзя трогать ту же карту. Положить в неё что-то ещё — напрямую или рекурсивным вызовом, как в мемоизации чисел Фибоначчи, — заканчивается IllegalStateException: Recursive update, если новый ключ попал в ту же корзину. А если попал в другую, код сегодня отработает, завтра при другом размере таблицы упадёт, и в худшем случае оставит корзину запертой навсегда. Карту меняют до вызова или после, но не изнутри.

Второе: функция должна быть короткой. Поход в базу или HTTP-вызов внутри computeIfAbsent держит замок корзины всё время запроса, и все, чьи ключи попали в ту же корзину, стоят и ждут. Дорогое значение считают снаружи, а в карту кладут уже готовым.

null в качестве ключа или значения ConcurrentHashMap не допускает — в отличие от HashMap. Это не каприз, а защита от неустранимой двусмысленности: в обычной HashMap вопрос «get вернул null — ключа нет или значение равно null?» решается вызовом containsKey. В конкурентной карте пара вызовов get + containsKey не атомарна — между ними другой поток мог вставить или удалить запись, и согласованного ответа не получить. Запрет null делает get(key) == null однозначным: ключа нет. Если нужно «значение-пустышка» — используйте маркер-объект.

Внутри: полосовая блокировка и CAS

HashMap внутри — это массив «корзин» (bucket): ключ по хешу попадает в одну из ячеек массива. Проблема многопоточности в том, что один общий замок на всю карту убил бы параллелизм: потоки, работающие с разными ключами, всё равно ждали бы друг друга.

ConcurrentHashMap дробит блокировку. В Java 7 карта делилась на 16 сегментов (Segment), каждый со своим замком (ReentrantLock) — это называют полосовой блокировкой (lock striping): ключ попадает в сегмент по старшим битам хеша, и два ключа из разных сегментов блокируются независимо.

В Java 8+ блокировку сделали ещё мельче — на уровне отдельной корзины:

• если корзина пустая, первый узел кладётся вообще без замка — атомарной операцией CAS (compare-and-swap: «запиши, только если там всё ещё пусто»);
• если в корзине уже есть элементы (коллизия), поток берёт synchronized только на голову этой корзины;
• когда цепочка в корзине разрастается (8+ элементов при таблице ≥64), она превращается в сбалансированное дерево — поиск в ней остаётся быстрым.

Ограничения, о которых важно помнить:

• Согласованного снимка не будет — и это ограничение не итератора, а самой задачи. Обход не бросает ConcurrentModificationException, но и не обещает, что вы увидите изменения, сделанные другими потоками по ходу дела: получается состояние «где-то между» началом и концом обхода. Поэтому «пройти по карте и посчитать» атомарно здесь нельзя в принципе: пока вы идёте по одному краю, меняется другой. Такие итераторы называют слабо согласованными (weakly consistent).
• size() точен ровно до тех пор, пока карту никто не меняет. Чтобы не блокировать всю карту ради счётчика, размер хранится «размазанным» по нескольким ячейкам-счётчикам (CounterCell), и size() их складывает. Когда записей нет, сумма верна; под нагрузкой это оценка — годится для метрик, не годится для решений в коде. И ещё: size() возвращает int и на очень больших картах упирается в Integer.MAX_VALUE — для них есть mappingCount() с результатом long.
• Атомарность — только на один ключ. compute/merge неделимы для одного ключа, но нельзя атомарно изменить два ключа сразу или «заблокировать всю карту» — для таких сценариев нужен внешний замок или другая структура.

Два родственника, которых не хватает чаще всего. Множества отдельного класса не получили: потокобезопасное множество делают из той же карты, Set<String> seen = ConcurrentHashMap.newKeySet(); — это Set с теми же гарантиями, что у карты, и он же возвращается из keySet(). Обёртка Collections.newSetFromMap делает то же самое руками и теперь не нужна.

А когда нужен порядок ключей — диапазонные запросы, «следующий больший», обход по возрастанию, — берут ConcurrentSkipListMap (и ConcurrentSkipListSet): это конкурентный аналог TreeMap на списке с пропусками вместо дерева, без глобального замка, с операциями за O(log n). Обычный TreeMap конкурентного варианта не имеет вовсе, и Collections.synchronizedSortedMap это один грубый замок.

CopyOnWriteArrayList — список для редких записей

CopyOnWriteArrayList решает проблему иначе: при каждой модификации (добавлении, удалении) она создаёт полную копию внутреннего массива. Увидеть это можно и в одном потоке — начав обход и дописав элемент посреди него:

живой пример

import java.util.Iterator;
import java.util.concurrent.CopyOnWriteArrayList;

public class SnapshotDemo {
    public static void main(String[] args) {
        CopyOnWriteArrayList<String> listeners = new CopyOnWriteArrayList<>();
        listeners.add("listener-1");
        listeners.add("listener-2");

        Iterator<String> walk = listeners.iterator();
        listeners.add("listener-3");   // запись после начала обхода — новая копия массива

        while (walk.hasNext()) {
            System.out.println("обход видит: " + walk.next());
        }
        System.out.println("в списке уже: " + listeners.size());
    }
}

Обход печатает два элемента, хотя в списке их уже три: итератор держит тот массив, который был на момент его создания.

Внутри: копирование при записи

Внутри лежит volatile ссылка на массив. Чтение (get, обход) просто берёт текущий массив по этой ссылке — без всякого замка, поэтому читатели никогда не ждут. Запись же:

1. берёт единственный замок, общий на всех писателей (они выстраиваются в очередь друг за другом);
2. копирует массив целиком в новый, чуть большего размера;
3. вносит изменение в копию;
4. volatile-записью переставляет ссылку на новый массив.

Читатель, который начал обход до шага 4, продолжает работать со старым массивом — тем самым «снимком», который был актуален на момент старта. Именно поэтому итерация безопасна и не бросает ConcurrentModificationException: массив под читателем в принципе не меняется, его просто заменяют целиком.

Ограничения:

• Каждая запись — это O(n) и новая аллокация: копируется весь массив. На больших списках или при частых изменениях это дорого по времени и по памяти (нагрузка на сборщик мусора).
• Итератор видит устаревший снимок: если во время обхода кто-то добавил элемент, вы его в этом обходе не увидите. И iterator().remove() не поддерживается — бросит UnsupportedOperationException.

Уместность сводится к двум числам: сколько элементов и как часто пишут. Сотни элементов с записью раз в минуту — идеальный случай: копия в сотню ссылок стоит микросекунды, читатели не ждут вовсе. Десятки тысяч элементов или запись в каждом запросе — уже нет: каждая запись копирует весь массив, и добавление N элементов по одному стоит O(N²) работы и N временных массивов, которые тут же становятся мусором. Практический ориентир: список обработчиков событий, набор включённых функций, редко меняющаяся конфигурация — да; накопление данных, буфер, кэш — нет, там ConcurrentHashMap или очередь.

И тот же тезис, что для карты: потокобезопасная коллекция не делает потокобезопасной составную операцию. Для списка это выглядит так:

if (!listeners.contains(listener)) {   // атомарная операция раз
    listeners.add(listener);           // атомарная операция два
}

Два потока успевают проверить одновременно и добавить обработчик дважды. У CopyOnWriteArrayList для этого есть готовый атомарный метод, addIfAbsent(listener); в общем случае составную операцию закрывают внешним замком или выбирают структуру, где она выражается одним вызовом.

Collections.synchronizedX — обёртки и их ограничения

Методы Collections.synchronizedList, Collections.synchronizedMap и аналоги оборачивают обычную коллекцию и добавляют synchronized на каждый метод — на одном общем мьютексе (объекте-обёртке).

Внутри это один грубый замок на всю коллекцию: в любой момент времени с ней работает ровно один поток, остальные ждут. Настоящей конкуренции нет — в отличие от ConcurrentHashMap, где потоки идут параллельно по разным корзинам. Плюс итерация не защищена: если один поток обходит список, а другой удаляет элемент — получим ConcurrentModificationException. Придётся вручную блокировать весь обход:

живой пример

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class SyncWrapperDemo {
    public static void main(String[] args) {
        List<String> syncList = Collections.synchronizedList(new ArrayList<>());
        syncList.add("a");   // каждый вызов защищён
        syncList.add("b");

        synchronized (syncList) {   // а обход приходится закрывать вручную
            for (String s : syncList) {
                System.out.println(s);
            }
        }
    }
}

Это неудобно и легко забыть. Поэтому ConcurrentHashMap и CopyOnWriteArrayList предпочтительнее в новом коде — они проектировались для конкуренции, а не адаптировались. Обёртка уместна в одном случае: чужой код отдаёт вам обычную коллекцию, поменять её тип нельзя, а обращений к ней мало.

BlockingQueue — мост между производителем и потребителем

BlockingQueue — особый вид очереди, который блокирует поток, если операция невозможна прямо сейчас:

• put блокирует производителя, если очередь заполнена;
• take блокирует потребителя, если очередь пуста.

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

живой пример

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class ProducerConsumerDemo {
    private static final String STOP = "стоп";

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(2);   // ёмкость 2

        Thread producer = new Thread(() -> {
            try {
                for (int i = 1; i <= 5; i++) {
                    queue.put("задача-" + i);   // ждёт, если очередь полна
                }
                queue.put(STOP);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread consumer = new Thread(() -> {
            try {
                String task;
                while (!(task = queue.take()).equals(STOP)) {   // ждёт, если очередь пуста
                    System.out.println("обрабатываю: " + task);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();
        producer.join();
        consumer.join();
    }
}

Последний элемент — метка окончания (STOP), её называют отравленной пилюлей: иначе потребитель навсегда уснёт на пустой очереди, и программа не завершится. Ловушка в том, что метку забирает только один потребитель. Потребителей трое — нужно положить три метки, иначе двое останутся висеть на take() навсегда, и приложение не остановится. Либо метку кладут по числу потребителей, либо вместо неё останавливают потоки прерыванием, а take() сам выйдет с InterruptedException.

В сервисе, кроме блокирующих put и take, почти всегда нужны их варианты с таймаутом — именно они, а не put, стоят в рабочем коде:

if (!queue.offer(task, 200, TimeUnit.MILLISECONDS)) {
    metrics.increment("queue.rejected");       // очередь забита: отказываем быстро
    throw new OverloadedException();
}
String task = queue.poll(5, TimeUnit.SECONDS); // null, если за 5 секунд ничего не пришло

Разница принципиальная: put на полной очереди вешает поток намертво, и под перегрузкой сервис перестаёт отвечать вместо того, чтобы честно отказать. offer с таймаутом превращает перегрузку в управляемый отказ, а poll с таймаутом даёт рабочему потоку периодически просыпаться и проверять, не пора ли завершаться. Есть и мгновенные формы без ожидания: offer(task) возвращает false, poll() возвращает null.

Внутри: один лок против двух

Блокировку «жди, пока нельзя» строят на ReentrantLock + Condition — это очередь ожидания, где поток засыпает и его будят, когда условие изменилось. А вот сколько замков — принципиально разное у двух главных реализаций:

• ArrayBlockingQueue (кольцевой массив фиксированной ёмкости) — один замок и два условия (notEmpty, notFull). Производитель и потребитель делят один замок: пока один работает с очередью, другой ждёт.
• LinkedBlockingQueue (связные узлы) — два раздельных замка: putLock на хвосте и takeLock на голове. Производитель кладёт в хвост, а потребитель забирает из головы одновременно, не мешая друг другу. За счётом элементов следит атомарный счётчик (AtomicInteger).

Есть и другие реализации. PriorityBlockingQueue отдаёт не первый положенный, а наименьший по компаратору. DelayQueue отдаёт элемент, только когда истёк его срок, — на ней строят отложенные задачи. SynchronousQueue вместимости не имеет вовсе: put ждёт, пока кто-то заберёт элемент из рук в руки.

Последняя выглядит бесполезной, пока не узнаёшь, где она стоит: это очередь по умолчанию у Executors.newCachedThreadPool(), и именно поэтому такой пул на каждую задачу, которую некому взять прямо сейчас, заводит новый поток — очередь его не примет. Отсюда и его репутация опасного, о чём статья про пулы. BlockingQueue лежит в основе ExecutorService — именно через неё задачи передаются потокам пула.

Ограничения: у ArrayBlockingQueue единый замок ограничивает пропускную способность при высокой нагрузке с обеих сторон; ёмкость фиксирована навсегда. У LinkedBlockingQueue по умолчанию ёмкость Integer.MAX_VALUE — фактически безграничная: если производитель стабильно быстрее потребителя, очередь растёт до OutOfMemoryError. Почти всегда стоит задавать ёмкость явно.

ConcurrentLinkedQueue — очередь без блокировок

Иногда блокировки не нужны вовсе. ConcurrentLinkedQueue — неограниченная очередь, которая обходится без единого замка, только атомарными CAS. Она построена на алгоритме Майкла — Скотта (Michael-Scott queue) — это классика неблокирующих структур данных.

живой пример

import java.util.concurrent.ConcurrentLinkedQueue;

public class LockFreeQueueDemo {
    public static void main(String[] args) {
        ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();
        queue.offer("a");   // добавление без блокировки
        queue.offer("b");

        System.out.println(queue.poll());
        System.out.println(queue.poll());
        System.out.println(queue.poll());   // очередь пуста — null, а не ожидание
    }
}

Внутри: добавление через CAS с повтором

Очередь — это связный список узлов с указателями на голову и хвост. Добавление элемента:

1. создаём новый узел;
2. пытаемся атомарным CAS прицепить его к next последнего узла: «поставь ссылку, только если она всё ещё null»;
3. если CAS удался — сдвигаем хвост на новый узел (тоже CAS);
4. если CAS не удался (другой поток успел раньше) — повторяем с шага 2.

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

Ограничения:

• Нет блокировки — это и плюс, и минус. Для схемы производитель — потребитель ConcurrentLinkedQueue не годится: у неё нет take, который бы усыпил потребителя на пустой очереди — третий poll() в примере выше просто вернул null. Пришлось бы крутить poll() в цикле (busy-wait) и жечь процессор. Для этого сценария берите LinkedBlockingQueue.
• size() — O(n) и неточен: чтобы посчитать размер, нужно пройти по всем узлам, а параллельные изменения делают результат приблизительным. isEmpty() дешёвый, size() в горячем коде — нет.
• Очередь неограниченная: как и у LinkedBlockingQueue по умолчанию, при перекосе скоростей память утечёт.

Как выбрать нужную коллекцию

Коротко

• Потокобезопасность строят тремя способами: блокировка, копирование при записи и неблокирующий CAS — от этого зависят скорость и ограничения.
• ConcurrentHashMap закрывает отдельную корзину, а не всю карту (в пустую пишет через CAS), поэтому потоки идут параллельно; расплата — слабо согласованные итераторы, приблизительный size() и неделимость только внутри одного вызова: пара get + put теряет обновления и здесь.
• CopyOnWriteArrayList копирует массив на каждую запись и переставляет ссылку — идеально для «часто читаем, редко пишем», дорого при частых изменениях; итератор видит старый снимок.
• Collections.synchronizedX — один грубый замок: настоящей конкуренции нет, а итерацию надо синхронизировать вручную.
• BlockingQueue блокирует поток вместо ошибки: ArrayBlockingQueue — один замок и фиксированная ёмкость, LinkedBlockingQueue — два замка (голова/хвост), но по умолчанию безгранична и может съесть память.
• ConcurrentLinkedQueue работает без замков на CAS (алгоритм Майкла — Скотта); зато у неё нет блокирующего take, а size() — O(n).
• Множество это ConcurrentHashMap.newKeySet(), порядок ключей и диапазоны — ConcurrentSkipListMap; CopyOnWriteArrayList хорош на сотнях элементов с редкой записью и плох на десятках тысяч.
• У очередей в сервисе берут offer/poll с таймаутом, а не put/take: перегрузка превращается в честный отказ; отравленных пилюль кладут по числу потребителей.
• Составную операцию коллекция не защищает и для списка: contains плюс add дают дубль, нужен addIfAbsent или внешний замок.

Что почитать дальше

• Атомарные переменные и CAS — как работает CAS, на котором держатся ConcurrentHashMap и ConcurrentLinkedQueue.
• Явные блокировки: Lock и ReentrantLock — тот самый замок, на котором держится ожидание в блокирующих очередях.
• ExecutorService и пулы потоков — как BlockingQueue используется внутри пулов потоков.
• Состояние гонки — откуда берутся ошибки конкуренции и как их находить.