當執(zhí)行的查詢數(shù)量很大時，數(shù)據(jù)存儲庫通常是高要求系統(tǒng)的瓶頸。延遲批處理執(zhí)行器（DelayedBatchExecutor）是一個組件，可通過在Java多線程應用程序中對所需查詢進行批處理來減少所需查詢的數(shù)量。

n個查詢1個參數(shù)與1個查詢n個參數(shù)

讓我們假設一個Java應用程序執(zhí)行對關系數(shù)據(jù)庫的查詢，以在給定其唯一標識符（id）的情況下檢索Product實體（行）。

查詢看起來像這樣：

現(xiàn)在，要檢索n種產(chǎn)品，可以通過兩種方法進行：

對一個參數(shù)執(zhí)行n個獨立查詢：

使用IN運算符或OR的組合，對n個參數(shù)執(zhí)行一次查詢以同時檢索n個產(chǎn)品

后者在網(wǎng)絡流量和數(shù)據(jù)庫服務器資源（CPU和磁盤）方面更為有效，因為：

到數(shù)據(jù)庫的往返次數(shù)為1，而不是n。

數(shù)據(jù)庫引擎針對n個參數(shù)優(yōu)化了其數(shù)據(jù)遍歷過程，即，它可能只需要對每個表進行一次掃描而不是n次掃描。

這不僅適用于SELECT操作，而且適用于其他操作，例如INSERT，UPDATE和DELETE，實際上，JDBC API包括這些操作的批處理操作。

這同樣適用于NoSQL存儲庫，其中大多數(shù)都顯式提供BULK操作。

延遲批處理執(zhí)行器

需要從數(shù)據(jù)庫檢索數(shù)據(jù)的Java應用程序（如REST微服務或異步消息處理器）通常實現(xiàn)為多線程應用程序（* 1），其中：

每個線程在其執(zhí)行的某個時刻執(zhí)行相同的查詢（每個查詢具有不同的參數(shù)）。

并發(fā)線程數(shù)很高（每秒數(shù)十或數(shù)百）。

在這種情況下，數(shù)據(jù)庫很可能在很短的時間間隔內(nèi)多次執(zhí)行相同的查詢。

如前所述，如果將這1個參數(shù)的n個查詢替換為具有n個參數(shù)的單個等效查詢，則應用程序?qū)⑹褂幂^少的數(shù)據(jù)庫服務器和網(wǎng)絡資源。

好消息是，可以通過以下涉及時間窗口的機制來實現(xiàn)它 ：

第一個嘗試執(zhí)行查詢的線程將打開一個時間窗口，因此其參數(shù)存儲在列表中，并且該線程已暫停。在時間窗口內(nèi)執(zhí)行相同查詢的其余線程會將其參數(shù)添加到列表中，并且也會被暫停。此時，尚未在數(shù)據(jù)庫上執(zhí)行任何查詢。

時間窗口結束或列表已滿（預先定義了最大容量限制）后，便會使用列表中存儲的所有參數(shù)執(zhí)行單個查詢。最后，一旦數(shù)據(jù)庫提供了該查詢的結果，每個線程將接收其相應的結果，并且所有線程將自動恢復。

我為自己（延遲批處理執(zhí)行器）構建了此機制的簡單輕便的實現(xiàn)，可以輕松在新的或現(xiàn)有的應用程序中使用。它基于 Reactor庫，并且使用帶有通量的Flux緩沖發(fā)布者作為參數(shù)列表。

使用延遲批處理執(zhí)行器的吞吐量和延遲分析

讓我們假設一個針對產(chǎn)品的REST微服務，它公開了一個端點，用于從給定的數(shù)據(jù)庫中檢索產(chǎn)品數(shù)據(jù) productId。如果不使用 延遲批處理執(zhí)行器，則說到端點每秒有200次命中，則數(shù)據(jù)庫每秒執(zhí)行200個查詢。如果端點使用的 時間窗口延遲批處理執(zhí)行器配置為50毫秒，最大容量 = 10個參數(shù)，則數(shù)據(jù)庫每秒僅執(zhí)行20個查詢，每個參數(shù)10個參數(shù)，但代價是最多在50毫秒內(nèi)增加延遲（* 2）對于每個線程執(zhí)行。

換句話說，為了將等待時間增加50 ms（* 2），在保持系統(tǒng)整體吞吐量的同時，數(shù)據(jù)庫每秒收到的查詢減少了10倍。

其他有趣的配置：

窗口時間 = 100毫秒，最大容量 = 20個參數(shù)→20個參數(shù)的10個查詢（查詢減少20倍）

窗口時間 = 500毫秒，最大容量 = 100個參數(shù)→2個查詢，共100個參數(shù)（查詢減少100倍）

延遲批處理執(zhí)行器在行動

深入研究Product微服務示例，假設對于每個傳入的HTTP請求，微服務的控制器都要求我們檢索提供其ID的Product（Java Bean），因此它將調(diào)用該方法：

public Product getProductById(Integer productId) DAO組件的ProductDAO。

讓我們看看不帶和帶的DAO的實現(xiàn) 延遲批處理執(zhí)行器。

沒有延遲批處理執(zhí)行器

使用延遲批處理執(zhí)行器

首先，延遲批處理執(zhí)行器必須在DAO中創(chuàng)建的實例，在本例中為delayedBatchExecutorProductById。它需要以下三個參數(shù)：

時間窗口（在此示例中為50毫秒）

參數(shù)列表的最大容量（在此示例中為10個參數(shù)）

將使用參數(shù)列表調(diào)用的方法（我們將在后面詳細介紹）。在此示例中，方法是retrieveProductsByIds

注意：我們將在后面看到為什么延遲批處理執(zhí)行器的標識（delayedBatchExecutor ProductById）是該類的實例DelayedBatchExecutor2

其次，DAO方法public Product getProductById(Integer productId)已經(jīng)過重構，可以簡單地調(diào)用實例的execute方法，僅此delayedBatchExecutor ProductById而已。所有的“魔術”都是由DelayedBatchExecutor。

之所以delayedBatchExecutor ProductById是的實例，DelayedBatchExecutor2是因為其execute方法返回一個Product實例并接收一個Integer實例作為其參數(shù)。因此，我們有： DelayedBatchExecutor2.

如果execute方法需要接收兩個參數(shù)（例如an Integer和a String）并返回的實例Product，則定義為DelayedBatchExecutor3 ，依此類推。

最后，該 retrieveProductsByIds方法必須返回a List并接收a List作為參數(shù)。

如果我們使用DelayedBatchExecutor3，則retrieveProductsByIds必須是List retrieveProductsByIds(List productIdsList, List stringList)

就是這樣。

一旦運行，執(zhí)行控制器邏輯的并發(fā)線程將getProductById(Integer id)在某個時候調(diào)用該方法，并且該方法將返回相應的乘積。他們不會知道他們實際上可能已經(jīng)被暫停并恢復了延遲批處理執(zhí)行器.

超越數(shù)據(jù)倉庫

盡管本文與數(shù)據(jù)存儲庫有關， 延遲批處理執(zhí)行器 但是可以在其他上下文中使用，例如，在對REST微服務的請求中。同樣，用一個參數(shù)啟動n個GET請求要比使用n個參數(shù)啟動1個GET要昂貴得多。

延遲批處理執(zhí)行器的改進

我創(chuàng)建 延遲批處理執(zhí)行器并使用了一段時間，以有效地處理由個人項目中的并發(fā)線程啟動的多個查詢的執(zhí)行。我相信它對其他人也可能有用，所以我決定將其公開。

話雖如此，仍有很大的改進空間并可以擴展所提供的功能 延遲批處理執(zhí)行器。最有趣的是能夠根據(jù)執(zhí)行的特定條件動態(tài)更改參數(shù) 延遲批處理執(zhí)行器（窗口時間和最大容量），以最大程度地減少等待時間，同時利用具有n個參數(shù)的查詢。