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

Инструменты и паттерны для многопоточной среды

Программирование в многопоточной среде — одна из самых сложных задач в Java. За годы разработки я использовал различные подходы и инструменты для обеспечения безопасности, производительности и надёжности многопоточного кода.

1. Синхронизация и Locks

synchronized (базовый уровень)

// Синхронизированный метод
public synchronized void increment() {
    counter++;
}

// Синхронизированный блок
public void updateData() {
    synchronized (this) {
        // Только один поток может выполнять этот код одновременно
        data.modify();
    }
}

java.util.concurrent.locks

private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

// Для операций чтения (много потоков)
public void read() {
    lock.readLock().lock();
    try {
        int value = data.getValue();
    } finally {
        lock.readLock().unlock();
    }
}

// Для операций записи (только один поток)
public void write(int value) {
    lock.writeLock().lock();
    try {
        data.setValue(value);
    } finally {
        lock.writeLock().unlock();
    }
}

// ReentrantLock с попыткой захвата
private final ReentrantLock lock = new ReentrantLock();

if (lock.tryLock(1, TimeUnit.SECONDS)) {
    try {
        // критическая секция
    } finally {
        lock.unlock();
    }
} else {
    // Не удалось захватить лок за 1 секунду
    handleTimeout();
}

2. Thread-Safe коллекции

// ConcurrentHashMap - безопасная для многопоточности
ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
map.put("key", 1);
map.putIfAbsent("key", 2); // атомарная операция
map.computeIfPresent("key", (k, v) -> v + 1); // атомарная трансформация

// CopyOnWriteArrayList - эффективна для чтения
CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
list.add("item"); // создаёт копию массива

// BlockingQueue - для производитель-потребитель паттерна
BlockingQueue<Task> queue = new LinkedBlockingQueue<>(100);

// Producer
public void produce(Task task) throws InterruptedException {
    queue.put(task); // ждёт, если очередь полна
}

// Consumer
public Task consume() throws InterruptedException {
    return queue.take(); // ждёт, если очередь пуста
}

3. AtomicXXX классы

// Атомарные операции без явной синхронизации
private final AtomicInteger counter = new AtomicInteger(0);

public void increment() {
    counter.incrementAndGet(); // атомарно
}

public void reset() {
    counter.set(0);
}

// AtomicReference для объектов
AtomicReference<User> currentUser = new AtomicReference<>(user1);
currentUser.compareAndSet(user1, user2); // CAS операция

// AtomicLong для счётчиков и таймеров
private final AtomicLong requestCount = new AtomicLong(0);

public void handleRequest() {
    requestCount.incrementAndGet();
}

4. ExecutorService и ThreadPool

// Создание пула потоков
ExecutorService executor = Executors.newFixedThreadPool(10);

// Отправка задач
Future<Integer> future = executor.submit(() -> {
    return complexCalculation();
});

// Ожидание результата
try {
    Integer result = future.get(5, TimeUnit.SECONDS);
} catch (TimeoutException e) {
    future.cancel(true);
}

// Использование schedule
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5);
scheduler.scheduleAtFixedRate(() -> {
    checkHealth();
}, 0, 30, TimeUnit.SECONDS);

// Корректное завершение
executor.shutdown();
if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
    executor.shutdownNow();
}

5. CountDownLatch и CyclicBarrier

// CountDownLatch - для ожидания завершения N потоков
CountDownLatch latch = new CountDownLatch(3);

// В каждом потоке
public void doWork() {
    try {
        workOnTask();
    } finally {
        latch.countDown();
    }
}

// Главный поток ждёт
latch.await(); // ждёт пока countDown() не вызовется 3 раза

// CyclicBarrier - для синхронизации N потоков в точке
CyclicBarrier barrier = new CyclicBarrier(3, () -> {
    System.out.println("Все потоки достигли барьера");
});

public void synchronizedWork() {
    doPhase1();
    barrier.await(); // ждём других потоков
    doPhase2();
}

6. Semaphore для ограничения доступа

// Ограничение одновременных операций до 5
private final Semaphore semaphore = new Semaphore(5);

public void limitedResource() {
    semaphore.acquire(); // ждём, если 5 потоков уже используют
    try {
        // только 5 потоков одновременно здесь
        expensiveOperation();
    } finally {
        semaphore.release();
    }
}

// Connection pool реализация
public class ConnectionPool {
    private final Semaphore available = new Semaphore(POOL_SIZE);
    private final Queue<Connection> pool = new LinkedList<>();
    
    public Connection getConnection() throws InterruptedException {
        available.acquire();
        return pool.poll();
    }
    
    public void releaseConnection(Connection conn) {
        pool.offer(conn);
        available.release();
    }
}

7. CompletableFuture (современный подход)

// Асинхронные вычисления без явного управления потоками
CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> fetchData())
    .thenApply(data -> processData(data))
    .thenApply(processed -> formatResult(processed))
    .exceptionally(e -> {
        logger.error("Error", e);
        return "Error occurred";
    });

String result = future.get();

// Объединение нескольких асинхронных операций
CompletableFuture<String> f1 = asyncFetch1();
CompletableFuture<String> f2 = asyncFetch2();
CompletableFuture<String> f3 = asyncFetch3();

CompletableFuture<Void> allDone = CompletableFuture.allOf(f1, f2, f3);

allDone.whenComplete((v, e) -> {
    if (e != null) {
        handleError(e);
    } else {
        String r1 = f1.join();
        String r2 = f2.join();
        String r3 = f3.join();
        combineResults(r1, r2, r3);
    }
});

8. StampedLock (Java 8+)

// Более эффективен чем ReadWriteLock для read-heavy workload
private final StampedLock lock = new StampedLock();
private double x, y;

public double distanceFromOrigin() {
    // Optimistic read
    long stamp = lock.tryOptimisticRead();
    double currentX = x;
    double currentY = y;
    
    if (!lock.validate(stamp)) {
        // Данные изменились, читаем заново с блокировкой
        stamp = lock.readLock();
        try {
            currentX = x;
            currentY = y;
        } finally {
            lock.unlockRead(stamp);
        }
    }
    
    return Math.sqrt(currentX * currentX + currentY * currentY);
}

9. Реальный пример: Thread-safe кэш

public class ThreadSafeCache<K, V> {
    private final ConcurrentHashMap<K, V> cache = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<K, CompletableFuture<V>> loading = new ConcurrentHashMap<>();
    
    public V get(K key, Function<K, V> loader) throws Exception {
        V value = cache.get(key);
        if (value != null) {
            return value; // Fast path
        }
        
        // Загружаем значение один раз
        CompletableFuture<V> future = loading.computeIfAbsent(key, k -> {
            return CompletableFuture.supplyAsync(() -> loader.apply(k));
        });
        
        value = future.get();
        cache.put(key, value);
        loading.remove(key);
        
        return value;
    }
}

Best Practices

1. Избегайте synchronized где возможно — используй concurrent коллекции и AtomicXXX
2. Используй ExecutorService вместо создания потоков вручную
3. Избегайте deadlock'ов — захватывай локи в одном порядке
4. Минимизируй критические секции — они уменьшают параллелизм
5. Предпочитай CompletableFuture для асинхронного кода
6. Тестируй с ThreadSanitizer или аналогами для обнаружения гонок

Итог: Многопоточное программирование требует правильного выбора инструментов. От базовой synchronized до продвинутых StampedLock и CompletableFuture — каждый инструмент имеет своё место в зависимости от требований производительности и сложности.