Это «плохая форма», чтобы использовать функцию карты, которая отправляет электронную почту в потоке Java?

Question

Я пытаюсь использовать потоки Java для параллельной обработки списка клиентов. Это краткосрочная помощь в решении проблемы, которую мы решаем в рамках перестройки всей нашей системы.

Я начинаю с структуры List<Customer> Customers, которая содержит контактную информацию клиента и все соответствующие данные транзакции. Концептуально код, который я заменяю, выглядит следующим образом:

long emailsSent = 0;
List<Customer> customers = methodLoadingAllrelevantData();
for (Customer customer: customers) {
    boolean isEmailSent = sendEmail(customer);
    if (isEmailSent) {
        emailsSent++;
    }
}

Функция sendMail(customer):

1. Определяет, следует ли отправлять электронное письмо
2. Форматирует электронную почту
3. Попытки отправить письмо
4. Возвращает истину, если электронное письмо было успешно отправлено

Не очень хороший код, но я просто пытаюсь добиться большей скорости от существующего кода, не пытаясь сделать его лучше. Метод и все его вызовы на 100 процентов безопасны для потоков.

Я поместил его в следующую структуру потока:

ForkJoinPool limitedParallelThreadPool = new ForkJoinPool(numberOfThreads);
emailsSent = limitedParallelThreadPool.submit( () ->
    customers.stream().parallel()
        .map(this::_emailCustomer)
        .filter(b -> b == true).count()
).get();

Это работает должным образом, возвращая те же данные, что и последовательная версия.

Мои вопросы: поскольку цель моего метода состоит в том, чтобы сгенерировать электронное письмо, неужели я плохо использую функцию map? Есть ли лучший ответ? По сути, я сопоставляю Customer с логическим значением, но часть этого сопоставления требует, чтобы процесс вызывал электронную почту.

Изначально я пытался использовать оператор forEach(), но я не мог понять, как получить счетчик, не добавляя информацию о состоянии в функцию sendMail, которая мешает его поточнобезопасности.

Answer 1

Поскольку вам все равно, в каком порядке отправляются эти электронные письма, я бы сказал, что в этом примере вы вроде бы в порядке. Просто вы полагаетесь на побочные эффекты промежуточной операции map и потенциально можете укусить вас. Например это:

Stream.of(1,2,3,4)
      .map(x -> x + 1)
      .count();

не будет выполнять map вообще (начиная с java-9), так как все, что вам нужно, это count и map не изменит окончательного числа. Ваш пример безопасен от этого, поскольку вы фильтруете, поэтому окончательный счет неизвестен, поэтому необходимо выполнить map. Как уже говорилось, для параллельной среды нет никакой гарантии относительно порядка, в котором будет выполняться map.

Жаль, что ваш sendEmail вообще что-то возвращает, все почтовые сервисы, которые я написал, были больше похожи на события - запускать и забывать; но я не могу сказать, какой именно вам нужен сценарий.

Подумайте о том факте, что ваша map операция будет блокироваться, пока вы не получите ответ, и это может вызвать следующую часть документации, на которую вам нужно обратить внимание:

ForkJoinPool создается с заданным целевым уровнем параллелизма; по умолчанию равно количеству доступных процессоров. Пул пытается поддерживать достаточно активных (или доступных) потоков путем динамического добавления, приостановки или возобновления внутренних рабочих потоков, даже если некоторые задачи останавливаются в ожидании присоединения к другим

Answer 2

Возвращает true, если электронное письмо было успешно отправлено

Не было бы худшей идеей воспользоваться тем, что ваш метод _emailCustomer возвращает boolean, поэтомувы можете использовать Stream#filter вместо комбинации обоих Stream#map и Stream#filter:

customers.parallelStream()
         .filter(this::_emailCustomer)
         .count()

Для ответа на ваш вопрос, однако, это будет зависеть от варианта использования, будет ли Stream#mapправильная промежуточная операция для использования.Согласно документации из Stream#map, аргумент Function, который принимает метод, должен быть:

a не мешающий , функция без сохранения состояния, применяемая к каждому элементу

Если ваш метод _emailCustomer либо создает помехи, либо сохраняет состояние, то я бы воздержался от вызова его в Stream#map, особенно в параллельном контексте.