Как на каждый Email отправить сообщение через ExecutorService

# Отправка сообщений на каждый Email через ExecutorService

ExecutorService - это потокопул (thread pool) для асинхронного выполнения задач. Это идеальное решение для отправки писем на список адресов параллельно, что значительно ускоряет обработку.

Базовый пример

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

public class EmailSender {
    private ExecutorService executorService;
    
    public EmailSender() {
        // Создаем пул из 5 потоков
        executorService = Executors.newFixedThreadPool(5);
    }
    
    public void sendEmailsToAll(List<String> emails, String subject, String body) {
        for (String email : emails) {
            // Отправляем каждое письмо в отдельном потоке
            executorService.submit(() -> sendEmail(email, subject, body));
        }
    }
    
    private void sendEmail(String email, String subject, String body) {
        try {
            // Логика отправки письма
            System.out.println("Отправка письма на " + email);
            Thread.sleep(100); // Имитация задержки
            System.out.println("Письмо отправлено на " + email);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Ошибка отправки письма на " + email + ": " + e.getMessage());
        }
    }
    
    public void shutdown() {
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

// Использование
public static void main(String[] args) {
    EmailSender sender = new EmailSender();
    List<String> emails = List.of(
        "user1@example.com",
        "user2@example.com",
        "user3@example.com"
    );
    
    sender.sendEmailsToAll(emails, "Hello", "Welcome!");
    sender.shutdown();
}

Разные типы ExecutorService

1. newFixedThreadPool(n) - пул фиксированного размера

ExecutorService executor = Executors.newFixedThreadPool(5);
// Всегда 5 потоков, неиспользуемые потоки остаются в памяти

2. newCachedThreadPool() - динамический пул

ExecutorService executor = Executors.newCachedThreadPool();
// Создает потоки по мере необходимости, переиспользует свободные
// Идеально для небольшого количества коротких задач

3. newSingleThreadExecutor() - один поток

ExecutorService executor = Executors.newSingleThreadExecutor();
// Выполняет задачи последовательно в одном потоке
// Полезен когда порядок выполнения критичен

4. ForkJoinPool - разделяй и властвуй

ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
// Для сложных задач, которые разделяются на подзадачи

Обработка результатов с Future

public class EmailSenderWithFuture {
    private ExecutorService executorService = Executors.newFixedThreadPool(5);
    
    public List<Future<Boolean>> sendEmailsAsync(List<String> emails) {
        List<Future<Boolean>> futures = new ArrayList<>();
        
        for (String email : emails) {
            // submit возвращает Future для отслеживания результата
            Future<Boolean> future = executorService.submit(() -> sendEmailAndReturnStatus(email));
            futures.add(future);
        }
        
        return futures;
    }
    
    public void waitForCompletion(List<Future<Boolean>> futures) {
        for (Future<Boolean> future : futures) {
            try {
                Boolean success = future.get(30, TimeUnit.SECONDS); // Ждет до 30 сек
                if (success) {
                    System.out.println("Письмо успешно отправлено");
                } else {
                    System.out.println("Ошибка отправки письма");
                }
            } catch (TimeoutException e) {
                future.cancel(true);
                System.out.println("Отправка заняла слишком много времени");
            } catch (InterruptedException | ExecutionException e) {
                System.err.println("Ошибка: " + e.getMessage());
            }
        }
    }
    
    private Boolean sendEmailAndReturnStatus(String email) {
        try {
            System.out.println("Отправка на " + email);
            Thread.sleep(100);
            return true;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }
}

// Использование
public static void main(String[] args) {
    EmailSenderWithFuture sender = new EmailSenderWithFuture();
    List<String> emails = List.of("a@example.com", "b@example.com");
    
    List<Future<Boolean>> futures = sender.sendEmailsAsync(emails);
    sender.waitForCompletion(futures);
}

Batch отправка с контролем ошибок

public class BatchEmailSender {
    private ExecutorService executor = Executors.newFixedThreadPool(3);
    
    public void sendEmailsBatch(List<String> emails, int batchSize) {
        List<String> batch = new ArrayList<>();
        
        for (String email : emails) {
            batch.add(email);
            
            // Отправляем батч когда достигли нужного размера
            if (batch.size() == batchSize) {
                List<String> currentBatch = new ArrayList<>(batch);
                executor.submit(() -> processBatch(currentBatch));
                batch.clear();
            }
        }
        
        // Отправляем оставшиеся
        if (!batch.isEmpty()) {
            executor.submit(() -> processBatch(batch));
        }
    }
    
    private void processBatch(List<String> batch) {
        System.out.println("Обработка батча из " + batch.size() + " писем");
        int successful = 0;
        int failed = 0;
        
        for (String email : batch) {
            try {
                sendEmail(email);
                successful++;
            } catch (Exception e) {
                failed++;
                System.err.println("Ошибка для " + email + ": " + e.getMessage());
            }
        }
        
        System.out.println("Батч завершён: " + successful + " успешно, " + failed + " ошибок");
    }
    
    private void sendEmail(String email) throws Exception {
        // Реальная отправка
        System.out.println("Отправляю на " + email);
    }
}

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

public class EmailSenderWithLatch {
    public void sendEmailsWithWait(List<String> emails) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(5);
        CountDownLatch latch = new CountDownLatch(emails.size());
        
        for (String email : emails) {
            executor.submit(() -> {
                try {
                    sendEmail(email);
                } finally {
                    latch.countDown(); // Уменьшаем счетчик
                }
            });
        }
        
        latch.await(); // Ждет пока все 5 задач закончатся
        System.out.println("Все письма отправлены!");
        executor.shutdown();
    }
    
    private void sendEmail(String email) {
        System.out.println("Отправляю на " + email);
    }
}

Правильное корректное завершение

public void gracefulShutdown(ExecutorService executor) {
    executor.shutdown(); // Останавливаем прием новых задач
    
    try {
        // Ждем завершения текущих задач (максимум 10 сек)
        if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
            // Если не завершились, прерываем их
            executor.shutdownNow();
            
            // Ждем еще раз (уже для прерванных)
            if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
                System.err.println("ExecutorService не завершился");
            }
        }
    } catch (InterruptedException e) {
        executor.shutdownNow();
        Thread.currentThread().interrupt();
    }
}

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

✅ Используй try-with-resources для ExecutorService
✅ Выбирай размер пула на основе CPU cores и типа задач
✅ Обрабатывай исключения в Runnable/Callable
✅ Используй Future для отслеживания результатов
✅ Всегда корректно закрывай executor (shutdown/awaitTermination)
✅ Избегай блокирующих операций в потоках
✅ Логируй ошибки отправки для переотправки

Вывод: ExecutorService - это мощный инструмент для параллельной обработки писем. Правильное управление потоками значительно ускоряет отправку и повышает надежность приложения.