Какой знаешь способ внедрения многопоточности в сервис, совершающий много однотипных тяжелых операций?

Какой знаешь способ внедрения многопоточности в сервис, совершающий много однотипных тяжелых операций

Многопоточность необходима для обработки тяжелых операций параллельно. Есть несколько проверенных подходов, каждый с собственными плюсами и минусами.

1. ExecutorService с Thread Pool

Самый надёжный и используемый подход в production:

import java.util.concurrent.*;

public class HeavyOperationService {
    private final ExecutorService executor;
    
    public HeavyOperationService(int poolSize) {
        this.executor = Executors.newFixedThreadPool(poolSize);
    }
    
    public void processItems(List<Item> items) {
        List<Future<?>> futures = new ArrayList<>();
        
        for (Item item : items) {
            Future<?> future = executor.submit(() -> {
                try {
                    performHeavyOperation(item);
                } catch (Exception e) {
                    handleError(e);
                }
            });
            futures.add(future);
        }
        
        for (Future<?> future : futures) {
            try {
                future.get();
            } catch (InterruptedException | ExecutionException e) {
                Thread.currentThread().interrupt();
                handleError(e);
            }
        }
    }
    
    private void performHeavyOperation(Item item) {
        Thread.sleep(1000);
    }
    
    private void handleError(Exception e) {
        e.printStackTrace();
    }
    
    public void shutdown() {
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
        }
    }
}

2. CountDownLatch для синхронизации

Для отслеживания завершения всех задач:

public class BatchProcessor {
    private final ExecutorService executor;
    
    public BatchProcessor(int poolSize) {
        this.executor = Executors.newFixedThreadPool(poolSize);
    }
    
    public void processBatch(List<Item> items) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(items.size());
        
        for (Item item : items) {
            executor.submit(() -> {
                try {
                    performHeavyOperation(item);
                } finally {
                    latch.countDown();
                }
            });
        }
        
        boolean completed = latch.await(5, TimeUnit.MINUTES);
        if (!completed) {
            System.out.println("Processing timeout!");
        }
    }
    
    private void performHeavyOperation(Item item) {
    }
}

3. CompletableFuture для асинхронности

Современный подход с цепочками операций:

import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

public class AsyncHeavyService {
    private final ExecutorService executor;
    
    public AsyncHeavyService(int poolSize) {
        this.executor = Executors.newFixedThreadPool(poolSize);
    }
    
    public CompletableFuture<List<Result>> processItemsAsync(List<Item> items) {
        List<CompletableFuture<Result>> futures = items.stream()
            .map(item -> CompletableFuture.supplyAsync(
                () -> performHeavyOperation(item),
                executor
            ))
            .collect(Collectors.toList());
        
        return CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[0])
        ).thenApply(v -> 
            futures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList())
        );
    }
    
    private Result performHeavyOperation(Item item) {
        return new Result(item.getId(), "processed");
    }
}

4. Fork/Join Framework для divide-and-conquer

Для рекурсивного разделения задач:

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

public class HeavyOperationTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10;
    private List<Item> items;
    
    public HeavyOperationTask(List<Item> items) {
        this.items = items;
    }
    
    @Override
    protected Long compute() {
        if (items.size() <= THRESHOLD) {
            return processDirectly();
        } else {
            int mid = items.size() / 2;
            HeavyOperationTask left = new HeavyOperationTask(
                items.subList(0, mid)
            );
            HeavyOperationTask right = new HeavyOperationTask(
                items.subList(mid, items.size())
            );
            
            left.fork();
            long rightResult = right.compute();
            long leftResult = left.join();
            
            return leftResult + rightResult;
        }
    }
    
    private long processDirectly() {
        return items.stream()
            .mapToLong(item -> performOperation(item))
            .sum();
    }
    
    private long performOperation(Item item) {
        return 1;
    }
}

ForkJoinPool pool = ForkJoinPool.commonPool();
long result = pool.invoke(new HeavyOperationTask(items));

5. Stream API parallelStream()

Самый простой способ для стандартных операций:

public class SimpleParallelService {
    public List<Result> processItems(List<Item> items) {
        return items.parallelStream()
            .map(this::performHeavyOperation)
            .collect(Collectors.toList());
    }
    
    private Result performHeavyOperation(Item item) {
        return new Result(item.getId(), "processed");
    }
}

6. Размер Thread Pool

Общее правило:

public class PoolSizing {
    public static void main(String[] args) {
        int cpuBound = Runtime.getRuntime().availableProcessors();
        int ioBound = Runtime.getRuntime().availableProcessors() * 2;
        int veryHeavy = Math.max(10, Runtime.getRuntime().availableProcessors() * 4);
        
        System.out.println("CPU-bound threads: " + cpuBound);
        System.out.println("I/O-bound threads: " + ioBound);
        System.out.println("Very heavy threads: " + veryHeavy);
    }
}

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

• ExecutorService: средняя простота, высокий контроль, хорошо масштабируется — лучше для production сервисов
• CompletableFuture: высокая простота, средний контроль, очень хорошо масштабируется — для асинхронных операций
• Fork/Join: сложная в реализации, очень высокий контроль, отличная масштабируемость — для рекурсивных задач
• parallelStream(): очень высокая простота, низкий контроль, средняя масштабируемость — для простых операций

Рекомендация для production: используй ExecutorService с FixedThreadPool для надёжности и контроля, или CompletableFuture если нужна асинхронность.