并行化流被分红多个块,每一个块独立处理,结果在最后汇总。html
CPU密集型代码以下:java
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到最大值之间的素数的数量。数字流由range方法建立,切换到并行模式,过滤掉非素数,剩余的计算总数。因为isPrime 方法极其无效且占用大量CPU,咱们能够利用并行化并利用全部可用的CPU内核。api
咱们来看另外一个例子:网络
private List<StockInfo> getStockInfo(Stream<String> symbols) { return symbols.parallel() .map(this::getStockInfo) //slow network operation .collect(toList()); } |
输入是一个股票代码列表,咱们必须调用慢速网络操做来获取有关股票的一些细节。在这里,咱们不处理CPU密集型操做,但咱们也能够利用并行化。并行执行多个网络请求是个好主意。一样,并行流的一个很好的任务,你赞成吗?this
若是您这样作,请再次查看上一个示例。有一个很大的错误。你看到了吗?问题是全部并行流都使用公共fork-join线程池。若是提交长时间运行的任务,则会有效地阻塞池中的全部线程。所以,您将阻塞使用并行流的全部其余任务。spa
想象一下servlet环境,当一个请求调用时getStockInfo() ,另外一个请求调用 countPrimes()。即便每一个都须要不一样的资源,也会阻止另外一个。更糟糕的是,你不能为并行流指定线程池; 整个类加载器必须使用相同的。.net
让咱们在下面的例子中说明它:线程
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() ); } |
在这里,咱们模拟系统中的六个线程。全部这些都在执行CPU密集型任务,第一个被“暂停”,在它找到素数后就睡了一秒钟。这只是一我的为的例子; 你能够想象一个被卡住或执行阻塞操做的线程。code
问题是:执行此代码时会发生什么?咱们有六个任务; 其中一个将须要一成天才能完成,其他的应该更快完成。绝不奇怪,每次执行代码时,都会获得不一样的结果。你想在生产系统中有这样的行为吗?一个杜塞的任务取消了应用程序的其他部分?我猜不会。htm
关于如何确保永远不会发生这样的事情,只有两种选择。第一个是确保提交到公共fork-join池的全部任务都不会卡,必须在合理的时间内完成。但这提及来容易作起来难,尤为是在复杂的应用程序中。
另外一种选择是不使用并行流,并等到Oracle容许咱们指定用于并行流的线程池。