Какие знаешь инструменты для многопоточной работы?

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

Многопоточность - это один из самых сложных и важных аспектов Java разработки. Существует множество инструментов и подходов для работы с потоками.

1. Thread - базовый класс для работы с потоками

Фундаментальный класс для создания и управления потоками:

public class ThreadExample {
    // Способ 1: наследование от Thread
    static class MyThread extends Thread {
        @Override
        public void run() {
            System.out.println("Running in thread: " + Thread.currentThread().getName());
        }
    }
    
    // Способ 2: реализация Runnable (лучше)
    static class MyRunnable implements Runnable {
        @Override
        public void run() {
            System.out.println("Running in thread: " + Thread.currentThread().getName());
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        // Способ 1
        Thread t1 = new MyThread();
        t1.start(); // запустить новый поток
        
        // Способ 2
        Thread t2 = new Thread(new MyRunnable());
        t2.start();
        
        // Способ 3: Lambda (Java 8+)
        Thread t3 = new Thread(() -> {
            System.out.println("Lambda thread");
        });
        t3.start();
        
        // Получить информацию о потоке
        System.out.println("Main thread: " + Thread.currentThread().getName());
        System.out.println("Thread count: " + Thread.activeCount());
        System.out.println("Thread state: " + t1.getState());
        
        // Ждать завершения потока
        t1.join();
        t2.join();
        t3.join();
        System.out.println("All threads finished");
    }
}

2. ExecutorService - управление пулом потоков

Вместо создания потоков вручную, лучше использовать пул:

public class ExecutorServiceExample {
    public static void main(String[] args) throws Exception {
        // Fixed thread pool - 5 потоков
        ExecutorService executor = Executors.newFixedThreadPool(5);
        
        try {
            // Отправить задачи
            for (int i = 0; i < 10; i++) {
                int taskId = i;
                executor.submit(() -> {
                    System.out.println("Task " + taskId + " running in thread " +
                        Thread.currentThread().getName());
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                });
            }
            
            // Shutdown и wait
            executor.shutdown();
            if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                executor.shutdownNow(); // force shutdown
            }
        } finally {
            if (!executor.isShutdown()) {
                executor.shutdownNow();
            }
        }
    }
}

Типы ExecutorService:

// Fixed размер
ExecutorService fixed = Executors.newFixedThreadPool(5);

// Кэшированный (растёт по необходимости)
ExecutorService cached = Executors.newCachedThreadPool();

// Single threaded
ExecutorService single = Executors.newSingleThreadExecutor();

// Scheduled работа
ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(2);
scheduled.schedule(() -> System.out.println("Delayed"), 5, TimeUnit.SECONDS);
scheduled.scheduleAtFixedRate(() -> System.out.println("Periodic"), 0, 1, TimeUnit.SECONDS);

// Work stealing (для CPU-intensive tasks)
ExecutorService workStealing = Executors.newWorkStealingPool();

3. Future и CompletableFuture - асинхронные результаты

public class FutureExample {
    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        
        try {
            // Future - простой результат
            Future<Integer> future = executor.submit(() -> {
                Thread.sleep(2000);
                return 42;
            });
            
            System.out.println("Waiting for result...");
            Integer result = future.get(); // блокирующий вызов
            System.out.println("Result: " + result);
            
            // CompletableFuture - более гибкий
            CompletableFuture<Integer> cfuture = CompletableFuture.supplyAsync(() -> {
                System.out.println("Computing...");
                return 100;
            });
            
            // Трансформация результата
            CompletableFuture<Integer> transformed = cfuture
                .thenApply(x -> x * 2)
                .thenApply(x -> x + 10);
            
            System.out.println("Final result: " + transformed.get());
            
            // Комбинирование множественных futures
            CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> 10);
            CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(() -> 20);
            
            CompletableFuture<Integer> combined = cf1.thenCombine(cf2, (a, b) -> a + b);
            System.out.println("Combined: " + combined.get());
            
            // Error handling
            CompletableFuture<Integer> withError = CompletableFuture.supplyAsync(() -> {
                if (true) throw new RuntimeException("Error!");
                return 42;
            })
            .exceptionally(ex -> {
                System.out.println("Exception caught: " + ex.getMessage());
                return 0;
            });
            
            System.out.println("Error handled: " + withError.get());
        } finally {
            executor.shutdown();
        }
    }
}

4. Synchronization - синхронизация потоков

public class SynchronizationExample {
    static class Counter {
        private int count = 0;
        
        // Способ 1: synchronized метод
        public synchronized void increment() {
            count++;
        }
        
        public synchronized int getCount() {
            return count;
        }
    }
    
    static class BetterCounter {
        private int count = 0;
        private final Object lock = new Object();
        
        // Способ 2: synchronized блок (более гранулярный)
        public void increment() {
            synchronized(lock) {
                count++;
            }
        }
        
        public int getCount() {
            synchronized(lock) {
                return count;
            }
        }
    }
    
    static class BestCounter {
        private final AtomicInteger count = new AtomicInteger(0);
        
        // Способ 3: AtomicInteger (лучший для простых случаев)
        public void increment() {
            count.incrementAndGet();
        }
        
        public int getCount() {
            return count.get();
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        Counter counter = new Counter();
        ExecutorService executor = Executors.newFixedThreadPool(10);
        
        try {
            for (int i = 0; i < 10000; i++) {
                executor.submit(counter::increment);
            }
            
            executor.shutdown();
            executor.awaitTermination(5, TimeUnit.SECONDS);
            
            System.out.println("Count: " + counter.getCount()); // 10000
        } finally {
            executor.shutdownNow();
        }
    }
}

5. Lock и ReentrantLock - явное управление блокировками

public class LockExample {
    static class Resource {
        private String data = "initial";
        private final Lock lock = new ReentrantLock();
        
        // Явное управление блокировкой
        public void write(String newData) {
            lock.lock();
            try {
                Thread.sleep(100);
                this.data = newData;
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                lock.unlock();
            }
        }
        
        public String read() {
            lock.lock();
            try {
                return data;
            } finally {
                lock.unlock();
            }
        }
        
        // Try lock с timeout
        public boolean tryWrite(String newData, long timeout, TimeUnit unit) {
            try {
                if (lock.tryLock(timeout, unit)) {
                    try {
                        this.data = newData;
                        return true;
                    } finally {
                        lock.unlock();
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return false;
        }
    }
}

6. CountDownLatch и CyclicBarrier - координация потоков

public class CoordinationExample {
    // CountDownLatch - один поток ждёт завершения других
    public static void countDownLatchExample() throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3);
        ExecutorService executor = Executors.newFixedThreadPool(3);
        
        for (int i = 0; i < 3; i++) {
            int taskId = i;
            executor.submit(() -> {
                System.out.println("Task " + taskId + " starting");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("Task " + taskId + " finished");
                latch.countDown();
            });
        }
        
        System.out.println("Waiting for all tasks...");
        latch.await(); // ждём все 3 задачи
        System.out.println("All tasks completed");
        executor.shutdown();
    }
    
    // CyclicBarrier - все потоки ждут друг друга
    public static void cyclicBarrierExample() throws InterruptedException {
        int numWorkers = 3;
        CyclicBarrier barrier = new CyclicBarrier(numWorkers, () -> {
            System.out.println("Phase complete!");
        });
        ExecutorService executor = Executors.newFixedThreadPool(numWorkers);
        
        for (int i = 0; i < numWorkers; i++) {
            int workerId = i;
            executor.submit(() -> {
                try {
                    for (int phase = 1; phase <= 3; phase++) {
                        System.out.println("Worker " + workerId + " phase " + phase);
                        barrier.await(); // ждём остальных
                    }
                } catch (InterruptedException | BrokenBarrierException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        
        executor.shutdown();
        executor.awaitTermination(10, TimeUnit.SECONDS);
    }
}

7. Semaphore - контроль доступа

public class SemaphoreExample {
    static class DatabaseConnection {
        private static final Semaphore semaphore = new Semaphore(5); // макс 5 соединений
        
        public static void useConnection() {
            try {
                semaphore.acquire();
                System.out.println(Thread.currentThread().getName() + " using connection");
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                semaphore.release();
                System.out.println(Thread.currentThread().getName() + " released connection");
            }
        }
    }
}

8. ThreadLocal - локальные данные потока

public class ThreadLocalExample {
    // Каждый поток имеет собственное значение
    private static final ThreadLocal<SimpleDateFormat> dateFormat =
        ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd"));
    
    public static String formatDate(Date date) {
        return dateFormat.get().format(date);
    }
    
    // Использование с cleanup
    public static void processRequest(String request) {
        try {
            ThreadLocal<String> context = new ThreadLocal<>();
            context.set(request);
            // работа
        } finally {
            context.remove(); // важно очищать!
        }
    }
}

9. ForkJoinPool - параллельная обработка

public class ForkJoinExample {
    static class SumTask extends RecursiveTask<Long> {
        private static final int THRESHOLD = 1000;
        private int[] array;
        private int start, end;
        
        public SumTask(int[] array, int start, int end) {
            this.array = array;
            this.start = start;
            this.end = end;
        }
        
        @Override
        protected Long compute() {
            if (end - start <= THRESHOLD) {
                // Базовый случай
                long sum = 0;
                for (int i = start; i < end; i++) {
                    sum += array[i];
                }
                return sum;
            } else {
                // Разделить и завоевать
                int mid = (start + end) / 2;
                SumTask left = new SumTask(array, start, mid);
                SumTask right = new SumTask(array, mid, end);
                
                left.fork(); // запустить в другом потоке
                long rightResult = right.compute();
                long leftResult = left.join();
                
                return leftResult + rightResult;
            }
        }
    }
    
    public static void main(String[] args) {
        int[] array = new int[10000];
        for (int i = 0; i < array.length; i++) {
            array[i] = 1;
        }
        
        ForkJoinPool pool = new ForkJoinPool();
        Long result = pool.invoke(new SumTask(array, 0, array.length));
        System.out.println("Sum: " + result);
    }
}

10. Virtual Threads (Java 19+) - лёгкие потоки

public class VirtualThreadExample {
    public static void main(String[] args) throws InterruptedException {
        // Создать миллионы virtual threads (в реальных потоках это невозможно)
        ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
        
        for (int i = 0; i < 1_000_000; i++) {
            int taskId = i;
            executor.submit(() -> {
                System.out.println("Task " + taskId);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.HOURS);
    }
}

Сравнение инструментов

Best Practices

1. Используй ExecutorService вместо создания Thread вручную
2. Избегай synchronized, используй Lock и Atomic
3. Для async операций используй CompletableFuture
4. Очищай ThreadLocal после использования
5. Всегда обрабатывай InterruptedException
6. Для новых проектов рассмотри Virtual Threads (Java 19+)
7. Профилируй перед оптимизацией concurrency

Многопоточность - это мощный инструмент, но требует тщательного понимания и уважения к сложностям распределённого программирования.