akka-streamsマップで非同期ドライバを使用する方法mapAsync

Question

reactivecouchbase非同期データベースドライバを使用し始めましたが、いくつかの基本的な設計上の問題が発生しています。 従来のアプローチでは、データベースへの接続数を制限することでデータベースにかかる負荷を制限しました。しかし、私は非同期ドライバを使用して、新しいクエリでデータベースを変えることはできますか？

これが重要になったのは、次のような例です。

私はデータベースを呼び出す方法が2通りあると言います。

My機能は、DBへの呼び出し：

asyncCallDB: Future[DBResponse] 
blockingCallDB: DBResponse 

は、今私は、私は2つの異なる機能を使用することができ、ストリームを超えるデシベルの呼び出しをマップする：

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? 

私はここに不足していると感じていますこの種の死刑を立てる。

Answer 1

ReactiveCouchbaseは、Couchbaseサーバーとの通信にAsyncHttpClientを使用します。できるだけ早くsee in the source codeを呼び出して、同時接続数を制限するsetMaximumConnectionsTotalを呼び出します。実際の値は、couchbase.http.maxTotalConnectionsで設定した内容によって異なります。

作成するCouchbaseBucketごとにAsyncHttpClientが1つあります。したがって、CouchbaseBucketごとに最大でmaxTotalConnectionsの接続があります。 Couchbase documentation on N1QL REST APIから

：要求が開始され における文の実行後に、結果が戻ってクライアントにストリーミングされるように

REST APIは、 終端を同期を実行したときに、ステートメント終了の実行。

したがって、実際には、各バケットの同時クエリ数はmaxTotalConnectionsに制限されています。

したがって、DBのバックプレッシャは、常に何らかの方法で制限されています。 maxTotalConnectionsを負でない数値に設定するか、またはRAMが限られているかファイル記述子の数が多いため、クライアントが接続をさらに作成できないためです。

ただし、クライアントのメモリが不足するように、あまりにも多くのを作成することは可能です。Futureこれが当てはまると思われるときはいつも、in this answerのようにmapAsync、または"Buffers and working with rate" (Akka documentation)に記載されている他のテクニックの1つを使用してください。

good description of mapAsync in the Akka documentationあります：未来の結果を返す関数へ

パスの着信要素。 未来が到着すると、結果は下流に渡されます。n個の要素まで同時に処理することができます...

Flow.mapAsyncは、それ自体では何も実行されません、それはちょうどあなたが、その後、SourceとSink間を接続する必要がFlowを返すことに注意してくださいrun 。 Akka Quick Start Guideはこれを非常に分かりやすい方法で説明しています。