必收藏技能！為Java多線程應用程序優化數據存儲庫
數據存儲庫通常是超高要求系統的瓶頸。在這些系統中，正在執行的查詢數量非常大。DelayedBatchExecutor是一個用于減少所需查詢數量的組件，通過在Java多線程應用程序中對所需查詢進行批處理。
1個參數的n個查詢Vs. n個參數的1個查詢
假設有一個對關系數據庫執行查詢的Java應用程序，以便在給定其唯一標識符(id)的情況下檢索Product實體(row)。
查詢如下所示：
SELECT * FROM PRODUCT WHERE ID =
現在，檢索n個Products，有如下兩種方法：
- 執行1個參數的n個獨立查詢：
- SELECT * FROM PRODUCT WHERE ID =
- SELECT * FROM PRODUCT WHERE ID =
- ...
- SELECT * FROM PRODUCT WHERE ID =
- 使用IN運算符或ORs的串聯，對n個參數執行1個查詢以便同時檢索n個 Products
- -- Example using IN OPERATOR
- SELECT * FROM PRODUCT WHERE ID IN (, , ..., )
后者在網絡流量和數據庫服務器資源(CPU和磁盤)方面更為有效，因為：
- 往返數據庫的次數為1，而不是n。
- 數據庫引擎優化了n個參數的數據遍歷過程，即每個表格可能只需要掃描1次，而不是n次。
這不僅適用于SELECT操作，而且適用于其他操作，例如 INSERTs，UPDATEs和DELETEs。實際上，JDBC API包括上述操作的批量處理操作。
同樣的情況也適用于NoSQL存儲庫，其中大多都明確提供BULK操作。
DelayedBatchExecutor
需要從數據庫中檢索數據的Java應用程序，如REST微服務或異步消息處理器，通常以多線程應用程序(*1)實現，其中：
- 每個線程在其執行的某個時刻執行相同的查詢(每個查詢具有不同的參數)。
- 并發線程數很高(每秒數十或數百)。
在這種場景下，數據庫很可能在較短的時間間隔內多次執行相同的查詢。
如前所述，如果將1個參數的n個查詢替換為具有n個參數的單個等效查詢，那么則應用程序將使用較少的數據庫服務器和網絡資源。
好消息是它可以通過timewindows(時間窗口)的機制來實現，如下所示：
第一個嘗試執行查詢的線程會打開一個時間窗口，因此其參數被存儲在一個列表中，同時該線程被暫停。在時間窗口內執行相同查詢的其余線程會將其參數添加到列表中，并且也會被暫停。此時，數據庫上未執行任何查詢。
時間窗口結束或列表已滿(先前已定義最大容量限制)后，將使用列表中存儲的所有參數執行單個查詢。最后，一旦數據庫提供了該查詢的結果，每個線程將接收相應的結果，同時所有線程將自動恢復。
筆者構建了一個簡單而輕量級的應用機制(DelayedBatchExecutor)，很容易在新的或現有的應用程序中使用。它基于Reactor庫，并且為參數列表使用超時的Flux緩沖發布器。
運用DelayedBatchExecutor的吞吐量和延遲分析
假設針對Products的REST微服務公開了一個端點，用于檢索數據庫中給定的 productId的Product數據。在沒有DelayedBatchExecutor的情況下，如果每秒對端點命中200次，則數據庫每秒執行200個查詢。如果端點使用的DelayedBatchExecutor 配置了50毫秒的時間窗口且最大容量=10個參數，數據庫每秒鐘將只執行10個參數的20個查詢，代價是每執行一個線程，最多在50毫秒內增加延時(*2)。
換句話說，為了將延時增加50毫秒(* 2)，數據庫每秒接收的查詢減少了10倍，然而保持了系統的整體吞吐量。還不錯!!
其他有趣的配置：
- 窗口時間= 100毫秒，最大容量= 20個參數→20個參數的10個查詢(查詢減少20倍)
- 窗口時間= 500毫秒，最大容量= 100個參數→2個查詢100個參數(查詢減少100倍)
執行中的DelayedBatchExecutor
深入研究Product微服務示例。假設對于每個傳入的HTTP請求，微服務的控制器都要求檢索已有id的Product(Java Bean)，因此將調用以下方法：
DAO組件(ProductDAO)的public Product getProductById(IntegerproductId) .
以下分別是有和沒有 DelayedBatchExecutor的DAO執行。
沒有 DelayedBatchExecutor
- public classProductDAO {
- public Product getProductById(Integer id) {
- Product product= ...// execute the query SELECT * FROM PRODUCT WHERE ID=
- // using your favourite API: JDBC, JPA, Hibernate...
- return product;
- }
- ...
- }
有DelayedBatchExecutor
- // Singleton
- publicclass ProductDAO {
- DelayedBatchExecutor2 delayedBatchExecutorProductById =
- DelayedBatchExecutor.define(Duration.ofMillis(50), 10, this::retrieveProductsByIds);
- public Product getProductById(Integer id) {
- Product product = delayedBatchExecutorProductById.execute(id);
- return product;
- }
- private List retrieveProductsByIds(List idList) {
- List productList = ...// execute query:SELECT * FROM PRODUCT WHERE ID IN (idList.get(0), ..., idList.get(n));
- // using your favourite API: JDBC, JPA, Hibernate...
- // The positions of the elements of the list to return must match the ones in the parameters list.
- // For instance, the first Product of the list to be returned must be the one with
- // the Id in the first position of productIdsList and so on...
- // NOTE: null could be used as value, meaning that no Product exist for the given productId
- return productList;
- }
- ...
- }
首先，必須在DAO中創建一個DelayedBatchExecutor實例，在本例中為 delayedBatchExecutorProductById。需要以下三個參數：
- 時間窗口(在此示例中為50毫秒)
- 參數列表的最大容量(在此示例中為10個參數)
- 將使用參數列表調用的方法(詳細信息見后文)。在此示例中，方法為retrieveProductsByIds
其次，已經重構了DAO方法 publicProduct getProductById(Integer productId)，以簡單調用delayedBatchExecutorProductById 實例的execute 方法。所有的“magic”都是由 DelayedBatchExecutor完成的。
之所以delayedBatchExecutorProductById是DelayedBatchExecutor2
如果execute方法需要接收兩個參數(例如，一個 Integer和一個String)并返回Product實例，則定義為 DelayedBatchExecutor3
最終，retrieveProductsByIds方法必須返回List
如果使用的是 DelayedBatchExecutor3
就是這樣。
一旦運行，執行控制器邏輯的并發線程會在某時刻調用方法 getProductById(Integerid) ，并且此方法將返回對應的Product。并發線程不知自己已經被 DelayedBatchExecutor暫停并恢復了。
由數據存儲庫延伸的“題外話”
盡管本文與數據存儲庫有關，但 DelayedBatchExecutor也可以用在其他地方，例如：對REST進行微服務請求。再說，用1個參數啟動n個GET請求要比使用n個參數啟動1個GET昂貴得多。
DelayedBatchExecutor的優化
筆者創建了 DelayedBatchExecutor并使用了一段時間，有效地解決了個人項目中并發線程啟動的多個查詢的執行問題。因此相信它對其他人可能也有用處，所以決定將其公開。
話雖如此，DelayedBatchExecutor改進和功能擴展的空間還很大。最有趣的是能夠根據執行的特定條件動態更改DelayedBatchExecutor參數(窗口時間和最大容量)的功能，以便在利用帶有n個參數的查詢時很大程度地減少延時。
