Что такое ConcurrentHashSet?

ConcurrentHashSet: потокобезопасное множество

ConcurrentHashSet — это потокобезопасная реализация интерфейса Set для многопоточной среды. В стандартной библиотеке Java нет класса с таким именем, но его функциональность доступна через ConcurrentHashMap или другие средства.

Почему нет встроённого ConcurrentHashSet?

Java имеет ConcurrentHashMap (потокобезопасное отображение), но не имеет отдельного ConcurrentHashSet. Это скорее историческое решение, так как Set можно эффективно реализовать на основе ConcurrentHashMap.

Как создать потокобезопасное множество?

1. Collections.newSetFromMap() — официальный способ

import java.util.concurrent.ConcurrentHashMap;
import java.util.Collections;

// Создание потокобезопасного Set
Set<String> concurrentSet = Collections.newSetFromMap(
    new ConcurrentHashMap<String, Boolean>()
);

// Использование
concurrentSet.add("item1");
concurrentSet.add("item2");
concurrentSet.contains("item1"); // true
concurrentSet.remove("item1");

Это официально рекомендуемый подход в Java документации.

2. ConcurrentHashMap как Set (альтернатива)

// Используем ConcurrentHashMap как Set
ConcurrentHashMap<String, Boolean> map = new ConcurrentHashMap<>();

// Добавление элемента
map.put("key1", Boolean.TRUE);
map.put("key2", Boolean.TRUE);

// Проверка наличия
boolean exists = map.containsKey("key1"); // true

// Получение всех элементов
Set<String> keys = map.keySet();

3. Кастомная реализация (если нужны специфичные операции)

public class ConcurrentHashSet<E> extends AbstractSet<E> {
    private final ConcurrentHashMap<E, Object> map = new ConcurrentHashMap<>();
    private static final Object PRESENT = new Object();

    @Override
    public boolean add(E e) {
        return map.putIfAbsent(e, PRESENT) == null;
    }

    @Override
    public boolean remove(Object o) {
        return map.remove(o) == PRESENT;
    }

    @Override
    public boolean contains(Object o) {
        return map.containsKey(o);
    }

    @Override
    public Iterator<E> iterator() {
        return map.keySet().iterator();
    }

    @Override
    public int size() {
        return map.size();
    }

    @Override
    public void clear() {
        map.clear();
    }
}

Потокобезопасность: почему ConcurrentHashMap?

ConcurrentHashMap использует bucketing (шардирование) вместо полной синхронизации:

// Типичная синхронизация (медленно)
Hashtable<String, String> hashtable = new Hashtable<>();
// Весь класс синхронизирован - один lock на весь объект

// ConcurrentHashMap (быстро)
ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
// Разделён на сегменты (buckets), каждый со своим lock
// Несколько потоков могут писать одновременно в разные segmenty

Архитектура ConcurrentHashMap (Java 8+)

В Java 8+ используется node-based locking:

// Внутренняя структура (упрощено)
public class ConcurrentHashMap<K, V> {
    // Массив buckets (узлов)
    transient volatile Node<K, V>[] table;
    
    // Каждый bucket может быть заблокирован отдельно
    // Позволяет параллельным операциям работать одновременно
    
    public V put(K key, V value) {
        // Блокируется только нужный bucket, не весь объект
        // Другие потоки могут работать с другими buckets
    }
}

Практический пример с потокобезопасностью

import java.util.concurrent.*;
import java.util.Collections;

public class ConcurrentSetExample {
    public static void main(String[] args) throws InterruptedException {
        // Создание потокобезопасного Set
        Set<Integer> concurrentSet = Collections.newSetFromMap(
            new ConcurrentHashMap<Integer, Boolean>()
        );
        
        // 10 потоков добавляют элементы
        ExecutorService executor = Executors.newFixedThreadPool(10);
        
        for (int t = 0; t < 10; t++) {
            int threadId = t;
            executor.submit(() -> {
                for (int i = 0; i < 1000; i++) {
                    concurrentSet.add(threadId * 1000 + i);
                }
            });
        }
        
        executor.shutdown();
        executor.awaitTermination(5, TimeUnit.SECONDS);
        
        System.out.println("Размер: " + concurrentSet.size()); // 10000
    }
}

Сравнение реализаций Set

Операции и их характеристики

Set<String> set = Collections.newSetFromMap(
    new ConcurrentHashMap<String, Boolean>()
);

// ADD операция - O(1) в среднем, потокобезопасна
set.add("element"); // Может быть заблокирован другим потоком

// CONTAINS операция - O(1) в среднем
set.contains("element"); // Читает без блокировки (обычно)

// REMOVE операция - O(1) в среднем
set.remove("element"); // Потокобезопасна

// SIZE операция - O(n) в ConcurrentHashMap!
// Нельзя быстро получить размер без итерации
int size = set.size();

// ITERATION - обеспечивает слабую консистентность
// Может видеть элементы, добавленные во время итерации
for (String element : set) {
    System.out.println(element);
}

Особенности и ограничения

1. Слабая консистентность — итератор не гарантирует snapshot'ы

   // Может добавиться элемент во время итерации
   set.add("item1");
   for (String item : set) {
       set.add("item2"); // Может быть видим в итерации
   }
   

2. Нет гарантий атомарности составных операций

   // Не гарантирует atomicity!
   if (!set.contains("key")) {
       set.add("key"); // Другой поток может добавить раньше
   }
   
   // Используй putIfAbsent в ConcurrentHashMap
   map.putIfAbsent("key", value);
   

3. Память и производительность

   • Размер по умолчанию 16 buckets, разрастается по мере добавления
   • Каждый bucket может быть заблокирован отдельно
   • Хорошее соотношение памяти и производительности

Когда использовать?

• Высокая конкуренция — много потоков читают и пишут одновременно
• Web приложения — обработка параллельных запросов
• Real-time системы — нужна минимальная задержка
• Кэширование — кэши часто многопоточны

Когда НЕ использовать?

• Однопоточные приложения — используй обычный HashSet
• Нужны гарантии упорядочивания — используй ConcurrentSkipListSet
• Нужны сложные блокировки — используй ReentrantReadWriteLock