Как запустишь обработку нескольких потоков

# Обработка нескольких потоков в Java

В Java есть несколько способов запустить параллельную обработку данных в нескольких потоках. Рассмотрю их в порядке рекомендации от современного к классическому.

1. ExecutorService (РЕКОМЕНДУЕТСЯ)

Это самый популярный и удобный способ управления потоками. ExecutorService автоматически управляет пулом потоков.

import java.util.concurrent.*;

public class ThreadPoolExample {
    
    public static void main(String[] args) throws InterruptedException {
        // Создаём пул из 4 потоков
        ExecutorService executor = Executors.newFixedThreadPool(4);
        
        // Отправляем 10 задач в пул
        for (int i = 1; i <= 10; i++) {
            final int taskId = i;
            executor.submit(() -> {
                System.out.println("Задача " + taskId + " в потоке " + 
                    Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);  // Имитация работы
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        
        // Останавливаем приём новых задач и ждём завершения старых
        executor.shutdown();
        
        // Ждём, пока все задачи завершатся (макс 5 сек)
        if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
            executor.shutdownNow();  // Принудительно прерываем оставшиеся
        }
        
        System.out.println("Все задачи завершены");
    }
}

2. ForkJoinPool (для рекурсивных задач)

Идеален для разделяй-и-властвуй алгоритмов (например, обработка деревьев, сортировка).

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class ForkJoinExample {
    
    // Задача суммирования массива рекурсивно
    static class SumTask extends RecursiveTask<Long> {
        private static final int THRESHOLD = 1000;
        private int[] arr;
        private int start, end;
        
        SumTask(int[] arr, int start, int end) {
            this.arr = arr;
            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 += arr[i];
                }
                return sum;
            } else {
                // Рекурсивный случай — разделяем на две задачи
                int mid = (start + end) / 2;
                SumTask task1 = new SumTask(arr, start, mid);
                SumTask task2 = new SumTask(arr, mid, end);
                
                task1.fork();  // Запускаем в другом потоке
                long sum2 = task2.compute();  // Вычисляем сами
                long sum1 = task1.join();  // Ждём результата первой задачи
                
                return sum1 + sum2;
            }
        }
    }
    
    public static void main(String[] args) {
        int[] array = new int[10000];
        for (int i = 0; i < array.length; i++) {
            array[i] = i;
        }
        
        ForkJoinPool pool = new ForkJoinPool();
        long result = pool.invoke(new SumTask(array, 0, array.length));
        
        System.out.println("Сумма: " + result);
    }
}

3. Callable и Future (с получением результата)

Когда нужен результат из потока.

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

public class CallableExample {
    
    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        
        List<Future<Integer>> futures = new ArrayList<>();
        
        // Отправляем 5 задач, которые возвращают результат
        for (int i = 1; i <= 5; i++) {
            final int taskId = i;
            Future<Integer> future = executor.submit(() -> {
                System.out.println("Запуск задачи " + taskId);
                Thread.sleep(1000);
                return taskId * taskId;  // Возвращаем результат
            });
            futures.add(future);
        }
        
        // Получаем результаты
        for (Future<Integer> future : futures) {
            try {
                Integer result = future.get();  // Блокирует до получения результата
                System.out.println("Результат: " + result);
            } catch (ExecutionException e) {
                System.err.println("Ошибка в задаче: " + e.getCause());
            }
        }
        
        executor.shutdown();
    }
}

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

Позволяет строить цепочки асинхронных операций.

import java.util.concurrent.*;

public class CompletableFutureExample {
    
    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        
        // Создаём асинхронную задачу
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("Вычисляем значение в потоке: " + 
                Thread.currentThread().getName());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return 42;
        }, executor);
        
        // Цепочка операций
        CompletableFuture<String> result = future
            .thenApply(num -> num * 2)  // Умножаем на 2
            .thenApply(num -> "Результат: " + num)  // Форматируем
            .whenComplete((res, ex) -> {
                if (ex != null) {
                    System.err.println("Ошибка: " + ex.getMessage());
                } else {
                    System.out.println(res);
                }
            });
        
        result.get();  // Ждём завершения
        executor.shutdown();
    }
}

5. Традиционный способ — Thread (не рекомендуется)

public class ThreadExample {
    
    public static void main(String[] args) throws InterruptedException {
        // Создаём потоки вручную
        Thread thread1 = new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                System.out.println("Поток 1: " + i);
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });
        
        Thread thread2 = new Thread(() -> {
            for (int i = 0; i < 5; i++) {
                System.out.println("Поток 2: " + i);
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });
        
        thread1.start();
        thread2.start();
        
        thread1.join();  // Ждём завершения потока 1
        thread2.join();  // Ждём завершения потока 2
        
        System.out.println("Все потоки завершены");
    }
}

6. Параллельные потоки (Parallel Streams)

Для обработки коллекций с использованием нескольких ядер.

import java.util.*;
import java.util.stream.*;

public class ParallelStreamsExample {
    
    public static void main(String[] args) {
        List<Integer> numbers = IntStream.rangeClosed(1, 1000)
            .boxed()
            .collect(Collectors.toList());
        
        // Параллельная обработка
        long sum = numbers.parallelStream()
            .filter(n -> n % 2 == 0)  // Чётные числа
            .map(n -> n * n)  // Квадрат
            .reduce(0, Integer::sum);  // Сумма
        
        System.out.println("Сумма квадратов чётных чисел: " + sum);
        
        // Параллельная сортировка
        numbers.parallelSort();
    }
}

Сравнение подходов

Best Practices

// ✅ Правильно — используем try-with-resources
try (ExecutorService executor = Executors.newFixedThreadPool(4)) {
    // Отправляем задачи
} finally {
    // Автоматически закроется
}

// ✅ Обработка InterruptedException
try {
    Thread.sleep(1000);
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();  // Восстанавливаем статус прерывания
}

// ✅ Выбирайте размер пула правильно
int poolSize = Runtime.getRuntime().availableProcessors();  // Для CPU-bound
int poolSize = Runtime.getRuntime().availableProcessors() * 2;  // Для I/O-bound

// ❌ Избегайте
// executor.shutdownNow();  // Может потерять данные
// Используйте graceful shutdown

Выводы

1. ExecutorService — стандартный выбор для большинства задач
2. ForkJoinPool — для разделяй-и-властвуй алгоритмов
3. CompletableFuture — для асинхронных цепочек операций
4. Parallel Streams — для простой параллельной обработки коллекций
5. Всегда корректно завершайте потоки — используйте shutdown() и обрабатывайте InterruptedException