并行流 線程池

并行數(shù)據(jù)庫流

在上一篇文章中 ，我寫了關(guān)于使用并行流和Speedment并行處理數(shù)據(jù)庫內(nèi)容的文章。 在許多情況下，并行流可能比通常的順序數(shù)據(jù)庫流快得多。

線程池

Speedment是一個(gè)開源的Stream ORM Java工具包和Runtime Java工具，它將現(xiàn)有數(shù)據(jù)庫及其表包裝到Java 8流中。 我們可以使用現(xiàn)有的數(shù)據(jù)庫并運(yùn)行Speedment工具，它將生成與我們使用該工具選擇的表相對(duì)應(yīng)的POJO類。 Speedment的一個(gè)獨(dú)特功能是它支持并行數(shù)據(jù)庫流，并且可以使用不同的并行策略來進(jìn)一步優(yōu)化性能。默認(rèn)情況下，并行流是在公共ForkJoinPool上執(zhí)行的，它們可能會(huì)與其他任務(wù)競爭。 在本文中，我們將學(xué)習(xí)如何根據(jù)自己的習(xí)慣執(zhí)行并行數(shù)據(jù)庫流。
ForkJoinPool，可以更好地控制我們的執(zhí)行環(huán)境。

加速入門

前往GitHub上的開放源Speedment ，學(xué)習(xí)如何開始Speedment項(xiàng)目。 將工具連接到現(xiàn)有數(shù)據(jù)庫確實(shí)非常容易。 閱讀我的
以前的文章提供了有關(guān)數(shù)據(jù)庫表和PrimeUtil類的外觀的更多信息，如以下示例所示。

在默認(rèn)的ForkJoinPool上執(zhí)行

這是我在上一篇文章中討論的應(yīng)用程序，它將并行掃描數(shù)據(jù)庫表以查找未確定的素?cái)?shù)候選者，然后將確定它們是否為素?cái)?shù)，并相應(yīng)地更新表。 看起來是這樣的：

Manager<PrimeCandidate> candidatesHigh = app.configure(PrimeCandidateManager.class).withParallelStrategy(ParallelStrategy.computeIntensityHigh()).build();candidatesHigh.stream() .parallel() // Use a parallel stream.filter(PrimeCandidate.PRIME.isNull()) // Only consider nondetermined prime candidates.map(pc -> pc.setPrime(PrimeUtil.isPrime(pc.getValue()))) // Sets if it is a prime or not.forEach(candidatesHigh.updater()); // Apply the Manager's updater

首先，我們使用stream().filter(PrimeCandidate.PRIME.isNull())方法在所有候選對(duì)象上創(chuàng)建流（使用名為ParallelStrategy.computeIntensityHigh（）的并行策略），其中“ prime” stream().filter(PrimeCandidate.PRIME.isNull()) null 。 然后，對(duì)每個(gè)這樣的總理候選人PC，我們無論是“黃金”列設(shè)置為true ，如果pc.getValue()是一個(gè)主要的或false ，如果pc.getValue()是不是一個(gè)素?cái)?shù)。 有趣的是， pc.setPrime()方法返回實(shí)體pc本身，使我們可以輕松地在多個(gè)流操作上進(jìn)行標(biāo)記。 在最后一行，我們通過應(yīng)用candidatesHigh.updater()函數(shù)以檢查結(jié)果更新數(shù)據(jù)庫。

同樣，請(qǐng)確保查看我以前的文章 ，了解并行策略的詳細(xì)信息和好處。 簡而言之，Java的默認(rèn)并行策略可以很好地滿足較低的計(jì)算需求，因?yàn)樗诿總€(gè)線程上放置了大量初始工作項(xiàng)。 Speedment的并行策略對(duì)于中等到較高的計(jì)算需求更好地工作，從而將少量工作項(xiàng)布置在參與線程上。

該流將確定完全并行的質(zhì)數(shù)，并且執(zhí)行線程將使用公共的ForkJoinPool如該圖所示（我的筆記本電腦具有4個(gè)CPU核心和8個(gè)CPU線程）：

使用自定義執(zhí)行器服務(wù)

正如我們?cè)诒疚拈_頭所了解的那樣，并行流是由Common執(zhí)行的。
默認(rèn)情況下， ForkJoinPool 。 但是，有時(shí)我們想使用我們自己的執(zhí)行器，也許是因?yàn)槲覀兒ε路簽E成災(zāi)
ForkJoinPool ，以便其他任務(wù)無法正常運(yùn)行。 可以為Speedment（和其他流庫）輕松地定義我們自己的執(zhí)行程序，如下所示：

final ForkJoinPool forkJoinPool = new ForkJoinPool(3);forkJoinPool.submit(() -> candidatesHigh.stream() .parallel().filter(PrimeCandidate.PRIME.isNull()).map(pc -> pc.setPrime(PrimeUtil.isPrime(pc.getValue()))).forEach(candidatesHigh.updater()); );try {forkJoinPool.shutdown();forkJoinPool.awaitTermination(1, TimeUnit.HOURS);} catch (InterruptedException ie) {ie.printStackTrace();}

該應(yīng)用程序代碼未經(jīng)修改，但被包裝到一個(gè)我們可以控制自己的自定義ForkJoinPool中。 在上面的示例中，我們?cè)O(shè)置了一個(gè)只有三個(gè)工作線程的線程池。 工作線程不與公共ForkJoinPool的線程共享。

使用自定義執(zhí)行程序服務(wù)的線程如下所示：

這樣，我們既可以控制實(shí)際的ThreadPool本身，也可以使用并行策略來精確控制工作項(xiàng)在該池中的布局方式！

保持游泳池中的熱量！

翻譯自: https://www.javacodegeeks.com/2016/11/work-parallel-database-streams-using-custom-thread-pools.html

并行流 線程池

總結(jié)

以上是生活随笔為你收集整理的并行流 线程池_使用自定义线程池处理并行数据库流的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。