Исправление потока потоковой обработки потока

Question

Я обнаружил, что существует простой пул потоков, используемый JVM для параллельной обработки потоков. У нас была блокированная функция ввода-вывода на большом потоке, которая вызывала проблемы с живимостью для несвязанных и в противном случае быстрых функций, используемых с несвязанными параллельными потоками.

В потоке нет методов, позволяющих использовать альтернативный пул потоков.

Есть ли простой способ избежать этой проблемы, возможно, каким-то образом определить, какой пул потоков использовать?

Answer 1

Это, возможно, похож на Custom thread pool in Java 8 parallel stream

Проблема дополнительно обсуждается в this blog.

ForkJoinPool forkJoinPool = new ForkJoinPool(2); 
forkJoinPool.submit(() -> 
    //parallel task here, for example 
    range(1, 1_000_000).parallel().filter(PrimesPrint::isPrime).collect(toList()) 
).get(); 

Answer 2

Я написал небольшую библиотеку под названием StreamEx, которая может представить задачу к пользовательскому ПСС. Таким образом, вы можете написать

ForkJoinPool forkJoinPool = new ForkJoinPool(2); 
int[] primes = IntStreamEx.range(1, 1_000_000) 
    .parallel(forkJoinPool) 
    .filter(PrimesPrint::isPrime).toArray(); 

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

Answer 3

Вы можете обернуть блокирующей операцию в ForkJoinPool.ManagedBlocker, вдоль линий этого:

static <T> Supplier<T> blocking(Supplier<T> supplier) { 
    return new Supplier<T>() { 
     volatile T result; 

     @Override 
     public T get() { 
      try { 
       ForkJoinPool.managedBlock(new ManagedBlocker() { 
        @Override 
        public boolean block() { 
         result = supplier.get(); 
         return true; 
        } 

        @Override 
        public boolean isReleasable() { 
         return result != null; 
        } 
       }); 
      } 
      catch (InterruptedException e) { 
       throw new RuntimeException(e); 
      } 

      return result; 
     } 
    }; 
} 

использовать его тогда, например, такие:

Stream.generate(blocking(() -> ...)) 
     .parallel() 
     ... 
     .collect(...); 

Более подробную информацию можно найти в это сообщение в блоге: http://zeroturnaround.com/rebellabs/java-parallel-streams-are-bad-for-your-health/

jOOλ предлагает обертки для всех Java 8 FunctionalInterface типы, как выше, с помощью org.jooq.lambda.Blocking, так что вы можете написать:

Stream.generate(Blocking.supplier(() -> ...)) 
     .parallel() 
     ... 
     .collect(...); 

Или, например, когда фильтр блокируется:

Stream.... 
     .parallel() 
     .filter(Blocking.predicate(t -> blockingTest(t))) 
     .collect(...); 

(Отказ от ответственности, я работаю в компании позади jOOλ).