問題タブ [monix]

私はMonix 3このコードを使用しており、次のようなコードを持っています:

このコードは問題なく動作しますが、長時間の IO 操作が多数あるため、次のようにリファクタリングすることにしました.mapParallelUnordered。

最初の例と同じ結果を取得しようとしていますが、並列処理を行っています。問題はTask.raiseError、オブザーバブル全体を殺すため、 で停止しi = 2ます。

エラーを処理し、Observable を存続させる方法は?

私のソケット リアクティブ ライブラリの作業プロセスの問題については、少し長い背景話があります。

Monixソケット ライブラリは、主に(ReactiveX に類似した)という名前のリアクティブ ライブラリに基づいています。Monix にはAck、現在のメッセージが処理されたときに次のメッセージを発行する (Future を拡張する) 型によってバックプレッシャーを処理するベスト プラクティスがあります。これは、大量のメッセージが殺到した場合にシステムを保護するための優れたメカニズムです。

現在の設計では、すべてのソケット接続は Observable (または Stream) であり、Observable は TCP/IP ネットワーク バイト ストリームが解析されるときにプロトコル メッセージを作成し、プロトコル メッセージをサブスクライバーにプッシュします。

問題は、Monix ライブラリだけがすべての Observable に対してバックプレッシャーを実行できることです。何千ものクライアントが接続されていると考えると、クラウドが非常に多くのオブザーバブルになり、バックプレッシャーは無意味になります。

では、単一の Observable 以外のグローバル システムに関して、そのようなリアクティブ システムの背圧メカニズムを設計するにはどうすればよいでしょうか。

ありがとう

アトミック変数を設定する次のコードがあります（両方とも同じようjava.util.concurrent.atomicにmonix.execution.atomic動作します：

2 回目の比較では、値は更新されませんでしたが、それでも foo が呼び出されます。副作用があるため、これは問題を引き起こしてfooいます (私の実際のコードでは、Akka アクターを作成し、重複したアクターを作成しようとするため、エラーが発生します)。

実際に使用されない限り、2 番目のパラメーターが評価されないようにするにはどうすればよいですか? （できれば同期を使用しないでください）

lazy val が機能しないように、暗黙のパラメーターを foo に渡す必要があります。例えば