Что такое параллелизм?

Параллелизм (Parallelism)

Параллелизм — это одновременное выполнение нескольких задач на разных физических ядрах процессора. В отличие от конкурентности (concurrency), когда потоки чередуются на одном ядре, параллелизм означает истинное одновременное выполнение нескольких операций на разных ядрах.

Различие между параллелизмом и конкурентностью

Конкурентность (1 ядро, несколько потоков):
Время |----T1----|----T2----|----T1----|----T2----|
      0     100    200     300    400    500
      
Параллелизм (2 ядра, 2 потока):
Ядро 1 |--------T1--------|--------T1--------|
Ядро 2 |--------T2--------|--------T2--------|
Время  0       200        400        600

Когда используется параллелизм

CPU-Bound задачи (Вычислительно интенсивные)

Задачи, которые требуют много вычислений:

public class CPUBoundExample {
    // Вычисление суммы всех чисел (CPU-bound)
    public static long computeSum(long from, long to) {
        long sum = 0;
        for (long i = from; i <= to; i++) {
            sum += i;
        }
        return sum;
    }
    
    // Последовательное выполнение (медленно)
    public static void sequentialApproach() {
        long sum = computeSum(1, 1_000_000_000);
        System.out.println("Sum: " + sum);
    }
    
    // Параллельное выполнение (быстро)
    public static void parallelApproach() {
        long sum = IntStream.range(1, 1_000_000_001)
            .asLongStream()
            .parallel()
            .sum();
        System.out.println("Parallel Sum: " + sum);
    }
}

IO-Bound задачи (Вход-выход)

Задачи, требующие ввода-вывода (конкурентность более уместна):

public class IOBoundExample {
    public static void fetchData(String url) throws IOException {
        URL connection = new URL(url);
        try (InputStream in = connection.openStream()) {
            byte[] buffer = new byte[1024];
            int bytesRead;
            while ((bytesRead = in.read(buffer)) != -1) {
                System.out.println("Read: " + bytesRead + " bytes");
            }
        }
    }
    
    public static void parallelIORequests() throws IOException {
        ExecutorService executor = Executors.newFixedThreadPool(4);
        List<String> urls = Arrays.asList(
            "http://example.com/1",
            "http://example.com/2",
            "http://example.com/3",
            "http://example.com/4"
        );
        
        for (String url : urls) {
            executor.submit(() -> {
                try {
                    fetchData(url);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        }
        
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
    }
}

Stream API для параллелизма

public class ParallelStreamsExample {
    public static void main(String[] args) {
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        
        // Последовательный поток
        long start = System.nanoTime();
        int sequentialSum = numbers.stream()
            .mapToInt(n -> {
                simulateWork();
                return n * 2;
            })
            .sum();
        long sequentialTime = System.nanoTime() - start;
        System.out.println("Sequential: " + sequentialSum + " Time: " + sequentialTime);
        
        // Параллельный поток
        start = System.nanoTime();
        int parallelSum = numbers.parallelStream()
            .mapToInt(n -> {
                simulateWork();
                return n * 2;
            })
            .sum();
        long parallelTime = System.nanoTime() - start;
        System.out.println("Parallel: " + parallelSum + " Time: " + parallelTime);
    }
    
    private static void simulateWork() {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

Fork/Join Framework

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

public class ForkJoinExample extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10000;
    private int[] array;
    private int start, end;
    
    public ForkJoinExample(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;
            ForkJoinExample leftTask = new ForkJoinExample(array, start, mid);
            ForkJoinExample rightTask = new ForkJoinExample(array, mid, end);
            
            leftTask.fork();
            long rightResult = rightTask.compute();
            long leftResult = leftTask.join();
            
            return leftResult + rightResult;
        }
    }
    
    public static void main(String[] args) {
        int[] array = new int[1000000];
        for (int i = 0; i < array.length; i++) {
            array[i] = i + 1;
        }
        
        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinExample task = new ForkJoinExample(array, 0, array.length);
        long result = pool.invoke(task);
        
        System.out.println("Sum: " + result);
    }
}

Использование ExecutorService для параллелизма

public class ExecutorServiceParallelism {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService executor = Executors.newFixedThreadPool(4);
        
        List<Callable<Integer>> tasks = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            int taskNum = i;
            tasks.add(() -> {
                System.out.println("Task " + taskNum + " running on " + 
                    Thread.currentThread().getName());
                Thread.sleep(1000);
                return taskNum * 2;
            });
        }
        
        List<Future<Integer>> futures = executor.invokeAll(tasks);
        
        for (Future<Integer> future : futures) {
            System.out.println("Result: " + future.get());
        }
        
        executor.shutdown();
    }
}

Когда НЕ использовать параллелизм

// Плохо: overhead параллелизма больше, чем экономия времени
int[] numbers = {1, 2, 3, 4, 5};
int sum = numbers.parallelStream().sum(); // Слишком маленький набор данных

// Хорошо
int[] numbers = new int[10_000_000];
int sum = Arrays.stream(numbers).parallel().sum();

Сравнение производительности

public class PerformanceComparison {
    public static void main(String[] args) {
        int[] array = new int[100_000_000];
        for (int i = 0; i < array.length; i++) {
            array[i] = i + 1;
        }
        
        // Sequential
        long start = System.nanoTime();
        long seqSum = 0;
        for (int i = 0; i < array.length; i++) {
            seqSum += Math.sqrt(array[i]);
        }
        long seqTime = System.nanoTime() - start;
        System.out.println("Sequential: " + seqTime + "ns");
        
        // Parallel
        start = System.nanoTime();
        long parSum = Arrays.stream(array)
            .parallel()
            .mapToLong(x -> (long)Math.sqrt(x))
            .sum();
        long parTime = System.nanoTime() - start;
        System.out.println("Parallel: " + parTime + "ns");
        
        System.out.println("Speedup: " + ((double)seqTime / parTime));
    }
}

Лучшие практики

1. Используй параллелизм для CPU-bound задач

   • Вычисления, обработка больших данных

2. Используй конкурентность для IO-bound задач

   • Сетевые запросы, файловые операции

3. Избегай параллелизма для малых наборов данных

   • Overhead создания потоков больше, чем экономия

4. Будь осторожен с shared state

   • Параллельные потоки могут конфликтовать

5. Измеряй производительность

   • Не всегда параллелизм быстрее

Количество потоков

public class ThreadPoolSize {
    public static void main(String[] args) {
        // Для CPU-bound
        int cpuBoundPoolSize = Runtime.getRuntime().availableProcessors();
        
        // Для IO-bound
        int ioBoundPoolSize = Runtime.getRuntime().availableProcessors() * 2;
        
        ExecutorService cpuExecutor = Executors.newFixedThreadPool(cpuBoundPoolSize);
        ExecutorService ioExecutor = Executors.newFixedThreadPool(ioBoundPoolSize);
    }
}

Выводы

• Параллелизм — одновременное выполнение на разных ядрах
• CPU-bound задачи выигрывают от параллелизма
• IO-bound задачи выигрывают от конкурентности
• Stream API с parallel() упрощает параллельные операции
• Fork/Join для сложного разделения работы
• Всегда измеряй — параллелизм не всегда быстрее
• Количество потоков зависит от типа задач