Статьи

Дважды подумайте, прежде чем использовать параллельные потоки Java 8

Если вы слушаете людей из Oracle, которые говорят о выборе дизайна Java 8, вы часто будете слышать, что параллелизм был основной мотивацией. Распараллеливание было движущей силой лямбд, потоковых API и других. Давайте посмотрим на пример потокового API.

private long countPrimes(int max) {
    return range(1, max).parallel().filter(this::isPrime).count();
}

private boolean isPrime(long n) {
    return n > 1 && rangeClosed(2, (long) sqrt(n)).noneMatch(divisor -> n % divisor == 0);
}

Здесь у нас есть метод,  countPrimes который считает число простых чисел от 1 до нашего максимума. Поток чисел создается методом диапазона. Затем поток переключается в параллельный режим; числа, не являющиеся простыми числами, отфильтровываются, а остальные числа считаются.

Вы можете видеть, что потоковый API позволяет нам описать проблему аккуратно и компактно. Более того, распараллеливание — это просто вызов  parallel()  метода. Когда мы делаем это, поток разделяется на несколько частей, причем каждый кусок обрабатывается независимо, а результат суммируется в конце. Поскольку наша реализация  isPrime метода крайне неэффективна и требует интенсивной загрузки ЦП, мы можем использовать преимущества распараллеливания и использовать все доступные ядра ЦП.

Давайте посмотрим на другой пример:

private List<StockInfo> getStockInfo(Stream<String> symbols) {
     return symbols.parallel()
            .map(this::getStockInfo) //slow network operation
            .collect(toList());
}

У нас есть список биржевых символов на входе, и нам нужно вызвать медленную сетевую операцию, чтобы получить некоторую информацию об акции. Здесь мы не имеем дело с процессором, интенсивно использующим процессор, но мы также можем воспользоваться преимуществами распараллеливания. Это хорошая идея, чтобы выполнить несколько сетевых запросов параллельно. Опять же, хорошая задача для параллельных потоков, вы согласны?

Если вы это сделаете, пожалуйста, посмотрите на предыдущий пример еще раз. Есть большая ошибка. Вы видите это? Проблема заключается в том, что все параллельные потоки используют общий пул потоков fork-join , и если вы отправляете долгосрочную задачу, вы фактически блокируете все потоки в пуле. Следовательно, вы блокируете все другие задачи, которые используют параллельные потоки. Представьте себе среду сервлета, когда один запрос вызывает  getStockInfo() другой countPrimes(). Один будет блокировать другой, даже если каждый из них требует разных ресурсов. Что еще хуже, вы не можете указать пул потоков для параллельных потоков; весь загрузчик классов должен использовать один и тот же.

Давайте проиллюстрируем это на следующем примере:

private void run() throws InterruptedException {  ExecutorService es = Executors.newCachedThreadPool();  // Simulating multiple threads in the system  // if one of them is executing a long-running task.  // Some of the other threads/tasks are waiting  // for it to finish  es.execute(() -> countPrimes(MAX, 1000)); //incorrect task  es.execute(() -> countPrimes(MAX, 0));  es.execute(() -> countPrimes(MAX, 0));  es.execute(() -> countPrimes(MAX, 0));  es.execute(() -> countPrimes(MAX, 0));  es.execute(() -> countPrimes(MAX, 0));  es.shutdown();  es.awaitTermination(60, TimeUnit.SECONDS);}private void countPrimes(int max, int delay) {  System.out.println(     range(1, max).parallel()        .filter(this::isPrime).peek(i -> sleep(delay)).count()  );}

Здесь мы моделируем шесть потоков в системе. Все они выполняют нагрузку на процессор, первая «сломана» и спит секунду после того, как находит простое число. Это всего лишь искусственный пример; Вы можете представить себе поток, который застрял или вместо этого выполняет блокирующую операцию.

Вопрос: что произойдет, когда мы выполним этот код? У нас шесть заданий; Один из них займет целый день, остальные должны закончить гораздо раньше. Не удивительно, что каждый раз, когда вы выполняете код, вы получаете другой результат. Иногда все здоровые задачи заканчиваются; в других случаях некоторые из них застряли за медленным. Хотите ли вы иметь такое поведение в производственной системе? С одной сломанной задачей, удаляющей остальную часть приложения? Я думаю, нет.

Есть только два варианта, как сделать так, чтобы такого никогда не было. Во-первых, убедитесь, что все задачи, отправленные в общий пул fork-join, не застрянут и не завершатся в разумные сроки . Но это легче сказать, чем сделать, особенно в сложных приложениях. Другой вариант — не использовать параллельные потоки и ждать, пока Oracle позволит нам указать пул потоков, который будет использоваться для параллельных потоков .

Ресурсы:

Первоначально опубликовано 17.08.15


Если вам понравилась эта статья и вы хотите узнать больше о потоках Java, ознакомьтесь с этой коллекцией руководств и статей  по всем вопросам, связанным с потоками Java.