并行數據庫流

在我以前的文章中 ，我寫了關于使用并行流和Speedment并行處理數據庫內容的信息。 在許多情況下，并行流可能比通常的順序數據庫流快得多。

線程池

Speedment是一個開源的Stream ORM Java工具包和Runtime Java工具，它將現有的數據庫及其表包裝到Java 8流中。 我們可以使用現有的數據庫并運行Speedment工具，它將生成與我們使用該工具選擇的表相對應的POJO類。 Speedment的一個獨特功能是它支持并行數據庫流，并且可以使用不同的并行策略來進一步優化性能。默認情況下，并行流是在公共ForkJoinPool上執行的，它們可能會與其他任務競爭。 在本文中，我們將學習如何根據自己的習慣執行并行數據庫流。
ForkJoinPool，可以更好地控制我們的執行環境。

加速入門

前往GitHub上的開放源Speedment ，學習如何開始Speedment項目。 將工具連接到現有數據庫確實很容易。 閱讀我的
前一篇文章提供了有關數據庫表和PrimeUtil類的外觀的更多信息，如以下示例所示。

在默認的ForkJoinPool上執行

這是我在上一篇文章中討論的應用程序，它將并行掃描數據庫表以查找不確定的素數候選者，然后將確定它們是否為素數，并相應地更新表。 它是這樣的：

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())方法在所有候選對象上創建一個流（使用名為ParallelStrategy.computeIntensityHigh（）的并行策略），其中“ prime” stream().filter(PrimeCandidate.PRIME.isNull()) null 。 然后，對每個這樣的總理候選人PC，我們無論是“黃金”列設置為true ，如果pc.getValue()是一個主要的或false ，如果pc.getValue()是不是一個素數。 有趣的是， pc.setPrime()方法返回實體pc本身，使我們能夠輕松地標記多個流操作。 在最后一行，我們通過應用candidatesHigh.updater()函數以檢查結果更新數據庫。

同樣，請確保查看我以前的文章 ，了解并行策略的詳細信息和好處。 簡而言之，Java的默認并行策略可以很好地滿足較低的計算需求，因為它在每個線程上放置了大量初始工作項。 Speedment的并行策略對于中等到較高的計算需求更好地工作，從而將少量工作項布置在參與的線程上。

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

使用自定義執行器服務

正如我們在本文開頭所了解的那樣，并行流是由Common執行的。
默認情況下為ForkJoinPool 。 但是，有時我們想使用我們自己的執行器，也許是因為我們害怕泛濫成災
ForkJoinPool ，以便其他任務無法正常運行。 可以為Speedment（和其他流庫）輕松地定義我們自己的執行程序，如下所示：

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();}

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

使用自定義執行程序服務的線程如下所示：

這樣，我們既可以控制實際的ThreadPool本身，又可以使用并行策略精確控制工作項在該池中的布局方式！

保持游泳池中的熱量！

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

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。