Как использовать Map в многопоточной среде?

Как использовать Map в многопоточной среде

Это критичный вопрос о параллелизме в Java. Покажу все основные подходы с примерами.

1. Проблема: обычная HashMap не потокобезопасна

Обычная HashMap вызывает race condition с несколькими потоками:

public class UnsafeMapExample {
    private static Map<String, Integer> unsafeMap = new HashMap<>();
    
    public static void main(String[] args) throws InterruptedException {
        // Запусти 1000 потоков, каждый добавляет 1000 значений
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                for (int j = 0; j < 1000; j++) {
                    unsafeMap.put("key" + j, j);
                }
            }).start();
        }
        
        Thread.sleep(2000);
        System.out.println("Expected: 1000, Actual: " + unsafeMap.size());
        // Результат: 500-900 (зависит от timing race conditions)
    }
}

Проблема: HashMap может потерять данные или даже зависнуть в infinite loop из-за:

• Race condition при resize
• Потери записей при одновременном доступе
• ConcurrentModificationException при итерации

2. Решение 1: Collections.synchronizedMap() — синхронизация целой карты

Самый простой способ — обернуть HashMap в synchronizedMap:

public class SynchronizedMapExample {
    private static Map<String, Integer> syncMap = 
        Collections.synchronizedMap(new HashMap<>());
    
    public static void main(String[] args) throws InterruptedException {
        // Запусти потоки
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                for (int j = 0; j < 1000; j++) {
                    syncMap.put("key" + j, j);
                }
            }).start();
        }
        
        Thread.sleep(2000);
        System.out.println("Size: " + syncMap.size()); // 1000 (правильно!)
    }
}

Как это работает:

// Внутри synchronized Map каждая операция синхронизирована
public class SynchronizedMap<K, V> extends AbstractMap<K, V> {
    private final Map<K, V> m;
    final Object mutex = this;
    
    public V get(Object key) {
        synchronized(mutex) {
            return m.get(key);
        }
    }
    
    public V put(K key, V value) {
        synchronized(mutex) {
            return m.put(key, value);
        }
    }
}

Минусы: блокирует ВСЮ карту, даже если разные потоки работают с разными ключами.

Thread 1: map.put("alice", 1)  // Locked
Thread 2: map.put("bob", 2)    // Ждёт разблокировки (неэффективно!)
Thread 3: map.get("charlie")   // Ждёт

3. Решение 2: ConcurrentHashMap — блокировка по сегментам (лучше!)

ConcurrentHashMap делит карту на сегменты и блокирует только нужный сегмент:

public class ConcurrentMapExample {
    private static Map<String, Integer> concurrentMap = new ConcurrentHashMap<>();
    
    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                for (int j = 0; j < 1000; j++) {
                    concurrentMap.put("key" + j, j);
                }
            }).start();
        }
        
        Thread.sleep(2000);
        System.out.println("Size: " + concurrentMap.size()); // 1000
    }
}

Как это работает:

ConcurrentHashMap разделена на 16 (default) сегментов:

┌──────────────┬──────────────┬──────────────┬──────────────┐
│  Segment 0   │  Segment 1   │  Segment 2   │  Segment 3   │
│  (lock 0)    │  (lock 1)    │  (lock 2)    │  (lock 3)    │
├──────────────┼──────────────┼──────────────┼──────────────┤
│ key0, key16, │ key1, key17, │ key2, key18, │ key3, key19, │
│ key32...     │ key33...     │ key34...     │ key35...     │
└──────────────┴──────────────┴──────────────┴──────────────┘

Thread 1: put("key0")   → блокирует Segment 0
Thread 2: put("key1")   → блокирует Segment 1 (параллельно!)
Thread 3: get("key2")   → блокирует Segment 2 (параллельно!)

Преимущества:

• Высокая пропускная способность (throughput)
• Несколько потоков могут работать одновременно
• По умолчанию 16 сегментов (Java 8+: динамические)

private static Map<String, Integer> concurrentMap = 
    new ConcurrentHashMap<>(16); // Начальная ёмкость

4. Атомарные операции в ConcurrentHashMap

ConcurrentHashMap предоставляет атомарные методы (не требуют дополнительной синхронизации):

public class AtomicOperationsExample {
    private static ConcurrentHashMap<String, Integer> counts = 
        new ConcurrentHashMap<>();
    
    public static void main(String[] args) throws InterruptedException {
        // Проблема: без защиты может быть race condition
        for (int i = 0; i < 100; i++) {
            new Thread(() -> {
                for (int j = 0; j < 100; j++) {
                    // ❌ НЕ потокобезопасно (три операции: get, +1, put)
                    counts.put("counter", counts.getOrDefault("counter", 0) + 1);
                }
            }).start();
        }
        
        Thread.sleep(2000);
        System.out.println("Count: " + counts.get("counter")); 
        // Ожидается 10000, получится ~7000-9000
    }
}

Решение — атомарные методы:

public class AtomicSafeExample {
    private static ConcurrentHashMap<String, Integer> counts = 
        new ConcurrentHashMap<>();
    
    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 100; i++) {
            new Thread(() -> {
                for (int j = 0; j < 100; j++) {
                    // ✅ Атомарная операция (выполняется без разрыва)
                    counts.compute("counter", (k, v) -> (v == null ? 0 : v) + 1);
                    
                    // Или более читаемо с Java 8+
                    counts.merge("counter", 1, Integer::sum);
                }
            }).start();
        }
        
        Thread.sleep(2000);
        System.out.println("Count: " + counts.get("counter")); // 10000 (правильно!)
    }
}

Атомарные методы ConcurrentHashMap:

// compute: вычисли новое значение на основе старого
mapCountMap.compute("key", (k, v) -> v == null ? 1 : v + 1);

// merge: объедини значения
map.merge("key", 1, Integer::sum);

// computeIfAbsent: добавь, если ключа нет
map.computeIfAbsent("key", k -> new ArrayList<>()).add(item);

// computeIfPresent: обнови, если ключ есть
map.computeIfPresent("key", (k, v) -> v + 1);

// putIfAbsent: добавь, если нет
map.putIfAbsent("key", defaultValue);

5. ReadWriteLock для специальных случаев

Если много читателей, мало писателей — используй ReadWriteLock:

public class ReadWriteLockExample {
    private Map<String, String> map = new HashMap<>();
    private ReadWriteLock lock = new ReentrantReadWriteLock();
    
    public String read(String key) {
        lock.readLock().lock();
        try {
            return map.get(key); // Много потоков могут читать одновременно
        } finally {
            lock.readLock().unlock();
        }
    }
    
    public void write(String key, String value) {
        lock.writeLock().lock();
        try {
            map.put(key, value); // Только один поток может писать
        } finally {
            lock.writeLock().unlock();
        }
    }
}

Сценарий использования:

• Cache с частыми чтениями и редкими обновлениями
• 1000 потоков читают, 1 поток обновляет

6. ConcurrentSkipListMap для отсортированного доступа

Если нужна отсортированная карта с параллелизмом:

public class SkipListExample {
    private Map<String, Integer> sortedConcurrentMap = 
        new ConcurrentSkipListMap<>();
    
    public void example() {
        sortedConcurrentMap.put("charlie", 3);
        sortedConcurrentMap.put("alice", 1);
        sortedConcurrentMap.put("bob", 2);
        
        // Автоматически отсортирована!
        sortedConcurrentMap.forEach((k, v) -> 
            System.out.println(k + ": " + v));
        // Output: alice: 1, bob: 2, charlie: 3
        
        // Диапазонные запросы
        Map<String, Integer> range = 
            sortedConcurrentMap.subMap("alice", "charlie");
    }
}

7. Практический пример: кэш с истечением (Cache)

public class ThreadSafeCacheExample {
    private static class CacheEntry<V> {
        V value;
        long expiryTime;
        
        CacheEntry(V value, long ttlMillis) {
            this.value = value;
            this.expiryTime = System.currentTimeMillis() + ttlMillis;
        }
        
        boolean isExpired() {
            return System.currentTimeMillis() > expiryTime;
        }
    }
    
    public class Cache<K, V> {
        private ConcurrentHashMap<K, CacheEntry<V>> cache = 
            new ConcurrentHashMap<>();
        
        public void put(K key, V value, long ttlMillis) {
            cache.put(key, new CacheEntry<>(value, ttlMillis));
        }
        
        public V get(K key) {
            CacheEntry<V> entry = cache.get(key);
            if (entry == null) return null;
            
            if (entry.isExpired()) {
                cache.remove(key);
                return null;
            }
            
            return entry.value;
        }
    }
}

8. Сравнение всех вариантов

┌────────────────────┬──────────┬──────────┬────────────────┐
│ Map тип            │ Скорость │ Сегменты │ Сортировка     │
├────────────────────┼──────────┼──────────┼────────────────┤
│ HashMap            │ ++++++   │ нет      │ нет (небезопас)│
│ SynchronizedMap    │ +        │ нет      │ нет            │
│ ConcurrentHashMap  │ +++      │ да (16+) │ нет            │
│ ConcurrentSkipList │ ++       │ da       │ да (O(log n))  │
│ ReadWriteLock      │ +++++    │ нет      │ нет (чтение)   │
└────────────────────┴──────────┴──────────┴────────────────┘

Вывод и рекомендации

Используй ConcurrentHashMap по умолчанию — это best practice для многопоточных приложений.

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

• Single-threaded → HashMap
• Multi-threaded (общий случай) → ConcurrentHashMap
• Много чтений, мало записей → ReadWriteLock + HashMap
• Нужна сортировка → ConcurrentSkipListMap
• Наследование (legacy) → Collections.synchronizedMap()

ConcurrentHashMap — правильный выбор в 99% случаев многопоточных приложений.