ConcurrentHashMap: примеры (JAVA)

Базовые примеры

Ниже приведены короткие примеры типичных операций. Код и ожидаемый вывод.

Пример 1 - простые put и get

import java.util.concurrent.ConcurrentHashMap;
public class Ex1 {
    public static void main(String[] args) {
        ConcurrentHashMap m = new ConcurrentHashMap<>();
        m.put("a", 1);
        m.put("b", 2);
        System.out.println(m.get("a"));
        System.out.println(m.get("c"));
    }
}

1
null

Пример 2 - putIfAbsent и replace

import java.util.concurrent.ConcurrentHashMap;
public class Ex2 {
    public static void main(String[] args) {
        ConcurrentHashMap m = new ConcurrentHashMap<>();
        System.out.println(m.putIfAbsent("k", "v1"));
        System.out.println(m.putIfAbsent("k", "v2"));
        System.out.println(m.replace("k", "v2"));
        System.out.println(m.replace("k", "v1", "v2"));
        System.out.println(m.get("k"));
    }
}

null
v1
v1
true
v2

Пример 3 - computeIfAbsent для ленивой инициализации

import java.util.concurrent.ConcurrentHashMap;
public class Ex3 {
    public static void main(String[] args) {
        ConcurrentHashMap m = new ConcurrentHashMap<>();
        Integer val = m.computeIfAbsent("count", k -> 100);
        System.out.println(val);
        System.out.println(m.get("count"));
    }
}

100
100

Аналоги в Java и их особенности

Несколько альтернатив внутри Java с краткими отличительными чертами:

• Hashtable - устаревшая потокобезопасная реализация, блокирует всю таблицу при операциях; менее производительна по сравнению с ConcurrentHashMap.
• Collections.synchronizedMap(new HashMap<>()) - синхронизированная обёртка, требует внешней синхронизации при итерации и комбинированных операциях; подходит при простых сценариях с малым параллелизмом.
• ConcurrentSkipListMap - потокобезопасная отсортированная карта; обеспечивает логарифмическое время операций и упорядоченность ключей, полезна когда нужен порядок.
• ConcurrentHashMap с LongAdder/Atomic types - комбинация для снижения конкуренции при инкрементах: хранение LongAdder или AtomicLong в качестве значения.

Выбор зависит от требований: при высоком уровне конкуренции предпочтительнее ConcurrentHashMap; при необходимости упорядоченности ключей - ConcurrentSkipListMap; при простоте и редком параллелизме - synchronizedMap или Hashtable (для совместимости).

Аналоги в других языках и их отличия

Краткие соответствия и отличия по языкам с примерами.

• C#: System.Collections.Concurrent.ConcurrentDictionary<K,V>. Очень похожа по семантике на ConcurrentHashMap, предоставляет atomic-операции GetOrAdd, AddOrUpdate.

  using System;
  using System.Collections.Concurrent;
  class P { static void Main(){
    var d = new ConcurrentDictionary();
    Console.WriteLine(d.GetOrAdd("x", 5));
  }}

  5

• Go: sync.Map для конкурентного доступа; API упрощён, но типобезопасность достигается приведением типов; для большинства случаев в Go используются обычные map с mutex.

  package main
  import (
    "fmt"
    "sync"
  )
  func main(){
    var m sync.Map
    m.Store("a", 1)
    v, _ := m.Load("a")
    fmt.Println(v)
  }

  1

• Python: встроенный dict не потокобезопасен для параллельных изменений. Частые подходы: использовать threading.Lock, collections.defaultdict с защитой или multiprocessing.Manager().dict() для межпроцессного доступа; альтернативно concurrent.futures для управления задачами.

  from threading import Lock
  m = {}
  lock = Lock()
  with lock:
      m['k'] = m.get('k', 0) + 1
  print(m)

  {'k': 1}

• JavaScript: структура Map не предназначена для многопоточности в браузере; в Node.js потоковая модель основана на событийном цикле и worker threads, где общая память редка. Для разделяемых данных применяются специальные механизмы (например, SharedArrayBuffer).
• PHP: ассоциативные массивы не потокобезопасны в многопроцессных сценариях; чаще используются внешние механизмы: memcached, Redis, базы данных или расширение pthreads для потоков.
• Lua: таблицы не защищены; многопоточность редко применяется в классическом варианте, используются сообщаемые очереди или внешние синхронизации в C-модулях.
• Kotlin: на JVM может использовать java.util.concurrent.ConcurrentHashMap; в Kotlin/Native требуются другие подходы.
• SQL: вместо in-memory-структур применяются транзакции и блокировки на уровне строк/таблиц; хороши для консистентности между процессами, но медленнее по сравнению с in-memory картами.

Отличия от Java-реализации в основном в гарантиях атомарности, API для условных операций и удобстве использования. C# ConcurrentDictionary наиболее близка по возможностям; Go sync.Map имеет упрощённый интерфейс; Python требует явной блокировки.

Типичные ошибки и примеры

Частые ошибки при работе с ConcurrentHashMap и последствия.

• Ожидание атомарности сложных операций. Комбинация get затем put без атомарного метода приводит к гонкам.

  import java.util.concurrent.*;
  public class Race {
    public static void main(String[] args) throws Exception {
      ConcurrentHashMap m = new ConcurrentHashMap<>();
      m.put("x", 0);
      Runnable r = () -> {
        Integer v = m.get("x");
        m.put("x", v + 1);
      };
      ExecutorService es = Executors.newFixedThreadPool(2);
      es.execute(r); es.execute(r);
      es.shutdown(); es.awaitTermination(1, java.util.concurrent.TimeUnit.SECONDS);
      System.out.println(m.get("x"));
    }
  }

  1  // возможен результат 1 вместо ожидаемых 2

  Причина: отсутствие атомарной операции. Правильнее использовать merge, compute или LongAdder.

• Попытки вставить null как ключ или значение приводят к NullPointerException.

  ConcurrentHashMap m = new ConcurrentHashMap<>();
  m.put(null, "v");

  Exception in thread "main" java.lang.NullPointerException

• Непонимание слабой консистентности итераторов: итераторы ConcurrentHashMap слабоконсистентны и не бросают ConcurrentModificationException; они могут не отражать все изменения, но остаются безопасными.
• Использование size() в горячем коде как дешёвой операции; в больших картах она может быть затратной. Для точного подсчёта рекомендуется mappingCount() если требуется long.
• Побочные эффекты в функциях remapping: функция, переданная в compute или merge, должна быть детерминированной и по возможности без побочных эффектов, иначе при параллельном выполнении поведение может быть трудноотслеживаемым.

Изменения в реализации и API

Краткая история основных изменений:

• Java 8 - значительная переработка внутренней архитектуры: сегменты были убраны в пользу CAS- и синхронизированных полос с улучшенной хеш-таблицей; добавлены bulk-операции (forEach, search, reduce) и методы интерфейса Map вроде computeIfAbsent, compute, merge.
• Параметр concurrencyLevel в конструкторе остался для совместимости, но в современных реализациях не оказывает существенного влияния на внутреннюю конкурентность.
• В последующих версиях ядро оптимизировалось для лучшей масштабируемости и снижения затрат аллокаторов. Специфические улучшения реализуются в JVM-реализациях и зависят от релизов JDK.

Рекомендация: при необходимости новых API проверять документацию конкретной версии JDK, поскольку оптимизации могут влиять на производительность в разных сценариях.

Продвинутые и неочевидные примеры

Несколько расширенных сценариев с кодом и результатом.

Пример 1 - счётчик слов с использованием merge, безопасный для параллельных потоков

import java.util.concurrent.*;
import java.util.*;
public class WordCount {
  public static void main(String[] args) throws Exception {
    ConcurrentHashMap counts = new ConcurrentHashMap<>();
    String[] words = {"a","b","a","c","b","a"};
    Arrays.stream(words).parallel().forEach(w ->
      counts.merge(w, 1, Integer::sum)
    );
    System.out.println(counts);
  }
}

{a=3, b=2, c=1}

Пояснение: merge выполняет атомическое обновление и подходит для частых инкрементов.

Пример 2 - снижение конкуренции с помощью LongAdder в значениях

import java.util.concurrent.*;
import java.util.concurrent.atomic.LongAdder;
public class AdderMap {
  public static void main(String[] args) throws Exception {
    ConcurrentHashMap map = new ConcurrentHashMap<>();
    Runnable r = () -> {
      map.computeIfAbsent("x", k -> new LongAdder()).increment();
    };
    ExecutorService es = Executors.newFixedThreadPool(4);
    for (int i=0;i<1000;i++) es.submit(r);
    es.shutdown(); es.awaitTermination(1, TimeUnit.SECONDS);
    System.out.println(map.get("x").sum());
  }
}

1000

Пояснение: LongAdder снижает конкуренцию по сравнению с AtomicLong при большом числе инкрементов.

Пример 3 - параллельные агрегирующие операции: поиск ключа по условию

import java.util.concurrent.*;
import java.util.*;
public class SearchExample {
  public static void main(String[] args) {
    ConcurrentHashMap map = new ConcurrentHashMap<>();
    for (int i=0;i<1000;i++) map.put(i, "v"+i);
    Integer found = map.search(10L, (k,v) -> v.contains("5") ? k : null);
    System.out.println(found);
  }
}

5  // пример результата: любой ключ, соответствующий условию

Пояснение: search может выполняться параллельно и вернуть первый найденный результат, подходящий под условие.

Пример 4 - мемоизация тяжёлой функции с computeIfAbsent и CompletableFuture

import java.util.concurrent.*;
public class Memo {
  private final ConcurrentHashMap> cache = new ConcurrentHashMap<>();
  public CompletableFuture get(String key) {
    return cache.computeIfAbsent(key, k ->
      CompletableFuture.supplyAsync(() -> heavyCalc(k))
    );
  }
  private String heavyCalc(String k) { try { Thread.sleep(200); } catch(Exception e){} return "res:"+k; }
  public static void main(String[] args) throws Exception {
    Memo m = new Memo();
    CompletableFuture f1 = m.get("a");
    CompletableFuture f2 = m.get("a");
    System.out.println(f1.get());
    System.out.println(f2.get());
  }
}

res:a
res:a

Пояснение: использование CompletableFuture в значении предотвращает повторные тяжёлые вычисления и позволяет нескольким потокам ждать одного результата.

Пример 5 - использование forEach с порогом параллелизма

import java.util.concurrent.*;
public class ForEachPar {
  public static void main(String[] args) {
    ConcurrentHashMap m = new ConcurrentHashMap<>();
    for (int i=0;i<100;i++) m.put(i, "v"+i);
    m.forEach(10L, (k,v) -> System.out.println(Thread.currentThread().getName()+":"+k));
  }
}

ForkJoinPool.commonPool-worker-1:0
main:1
...  // вывод из нескольких потоков

Пояснение: bulk-операции принимают порог, ниже которого выполняются последовательно, выше которого - параллельно.