如何在akka流地圖中使用異步驅動程序vs mapAsync

Question

我剛開始使用異步數據庫驅動程序reactivecouchbase，但遇到了一些基本的設計問題。 在傳統的方法中，我會通過限制連接數來限制數據庫施加的壓力。然而，使用異步驅動程序，我可以用新的查詢來淹沒數據庫？

如果這已成爲重要的是在一個例子如下。

可以說我有兩種不同的方式調用數據庫。

我的函數調用DB：

asyncCallDB: Future[DBResponse] 
blockingCallDB: DBResponse 

現在我想分貝呼籲過流，我可以使用兩種不同的功能映射：

Flow.map() 
Flow.mapAsync(numberOfConcurrentCalls)() 

現在我的問題是你會如何選擇調用數據庫：

Flow.map(blockingCallDB) //One call at a time with back preassure 
Flow.map(asyncCallDB) //Unlimited calls floods db no back pressure? 

Flow.mapAsync(numberOfConcurrentCalls)(blockingCallDB) //Up to numberOfConcurrentCalls at the same time with back pressure 
Flow.mapAsync(numberOfConcurrentCalls)(asyncCallDB) //Unlimited calls floods db no back pressure? 

我覺得我understading缺少這裏，想下適應這種類型的決定。

Answer 1

ReactiveCouchbase使用AsyncHttpClient與Couchbase服務器進行通信。正如你可以see in the source code它調用setMaximumConnectionsTotal，這限制了併發連接的數量。實際值取決於您在couchbase.http.maxTotalConnections中配置的內容。

您創建的每個CouchbaseBucket都有一個AsyncHttpClient。所以每個CouchbaseBucket最多有maxTotalConnections連接。

從Couchbase documentation on N1QL REST API：

的REST API運行同步，所以一旦在 語句的執行請求被啓動，結果流返回給客戶端， 終止語句完成的執行。

因此，在實踐中，每個存儲桶的併發查詢數限制爲maxTotalConnections。

因此，DB上的反壓總是以某種方式受到限制。因爲您將maxTotalConnections設置爲非負數，或者因爲您的客戶端由於RAM或文件描述符的數量有限而無法創建更多連接。

但是，仍然有可能創建太多Future s，以至於您的客戶端將耗盡內存。無論何時您認爲可能出現這種情況，您應該使用mapAsync，如in this answer或"Buffers and working with rate" (Akka documentation)中提到的其他技術之一。

有一個good description of mapAsync in the Akka documentation：

通行證進入的元素，返回一個Future結果的功能。 當 未來到達時，結果傳遞到下游。最多n個元素可以同時處理 ...

記住Flow.mapAsync將不會運行本身任何東西，它只是返回Flow，你必須一個Source和Sink之間的連接，然後run 。 Akka Quick Start Guide以非常理解的方式描述了這一點。