問題タブ [reactivex]

私は現在、.NET 用の Reactive Extensions フレームワークに取り組み始めており、見つけたさまざまな紹介リソース (主にhttp://www.introtorx.com )を調べています。

私たちのアプリケーションには、ネットワーク フレームを検出する多くのハードウェア インターフェイスが含まれます。これらは私の IObservable になります。その後、これらのフレームを消費したり、データに対して何らかの変換を実行したり、新しいタイプのフレームを生成したりするさまざまなコンポーネントがあります。たとえば、n フレームごとに表示する必要がある他のコンポーネントもあります。Rx は私たちのアプリケーションに役立つと確信していますが、IObserver インターフェイスの実装の詳細に苦労しています。

私が読んだリソースのほとんど (すべてではないにしても) は、IObservable インターフェイスを自分で実装するのではなく、提供されている関数またはクラスのいずれかを使用するべきだと述べています。私の調査によると、 を作成するSubject<IBaseFrame>と必要なものが得られるようです。ハードウェア インターフェイスからデータを読み取り、Subject<IBaseFrame>インスタンスの OnNext 関数を呼び出す単一のスレッドが作成されます。その後、さまざまな IObserver コンポーネントがそのサブジェクトから通知を受け取ります。

私の混乱は、このチュートリアルの付録にある次のアドバイスから来ています。

サブジェクト タイプの使用は避けてください。Rx は事実上、関数型プログラミングのパラダイムです。サブジェクトを使用するということは、潜在的に変化する状態を管理していることを意味します。変更状態と非同期プログラミングの両方を同時に処理することは、非常に困難です。さらに、多くの演算子 (拡張メソッド) は、サブスクリプションとシーケンスの正確で一貫した有効期間が維持されるように注意深く記述されています。主題を導入するとき、これを破ることができます。サブジェクトを明示的に使用すると、将来のリリースでもパフォーマンスが大幅に低下する可能性があります。

私のアプリケーションは非常にパフォーマンスが重要です。Rx パターンを使用してパフォーマンスをテストしてから、実稼働コードに入れる予定です。ただし、Subject クラスを使用して Rx フレームワークの精神に反することを行っていること、およびフレームワークの将来のバージョンでパフォーマンスが低下することを懸念しています。

私がやりたいことをするより良い方法はありますか？ハードウェア ポーリング スレッドは、オブザーバーがあるかどうかに関係なく継続的に実行されるため (それ以外の場合は HW バッファーがバックアップされます)、これは非常にホットなシーケンスです。次に、受信したフレームを複数のオブザーバーに渡す必要があります。

アドバイスをいただければ幸いです。

からすべての履歴データを取得する方法はPublishSubject?

それは印刷します：

b: 1印刷されていないことがわかります。

(複数の場所からオブザーバブルを更新する必要があるため) 使用するPublishSubject必要がある場合、後のサブスクライバーもすべての履歴データを取得できるようにするにはどうすればよいですか?

私のリアクティブ プログラムを rxscala でテストするには、次のようなものを作成する必要がありますObservable。

どれの

1. 発行番号1
2. それから待つ1s
3. 発行番号4
4. それから待つ3s
5. 発行番号2
6. それから待つ2s

私は醜い解決策を持っていますThreadとReplaySubject：

より良い解決策はありますか？

簡単なrxscalaコード:

実行すると、例外がスローされます。

どこが間違っていますか？

rxscala を使用すると、次のようにオブザーバブルをサブスクライブできます。

側からすべてのオブザーバー (4 つあります) のサブスクライブを解除することは可能streamですか?

私は持っていObservable<String>ます。Map<String, Int>これを、個別の文字列ごとの出現回数を示すに変えたいと思います。

オブザーバブルには約 10 億個の要素が含まれており、そのうち 1000 個が個別のものです (したがって、データセット全体を RAM に保存することはできません)。現在、 を反復処理してObservable更新しHashMapます。また、競合状態を回避するために、同じスレッドで必ず監視します。ただし、要素周波数の取得は本質的に簡単に並列化できるはずなので、それを利用すると便利です。

それを行う方法はありますか？

私は一般的にRxJavaとFPにかなり慣れていません。Observable2 つの sを結合するコードを書きたいと思います。整数のセットが 2 つあるとします。

• [0..4]のモジュロとしてキーセレクターを使用して2、(key, value) = {(0,0), (1,1), (0,2),...}
• [0..9]のモジュロとしてキーセレクターを使用して3、(key, value) = {(0,0), (1,1), (2,2), (0,3), (1,4),...}

それらに参加するための私の手順は次のとおりです。

1. 各セットをそのキーでグループ化します。1 番目のセットは、 キー0とで 2 つのグループを作成します1。2 つ目は、キー0、1およびで 3 つのグループを作成します2。
2. グループの 2 つのセットのデカルト積を作成し、合計で 6 つのグループのペアをキーとして与えます: 0-0、0-1、0-2、1-0、。1-11-2
3. 0-0と のみを残して、両側に同じキーを持つペアのみをフィルタリングします1-1。
4. 各ペア内で、左右のグループのデカルト積を作成します。

以下は、デカルト積を計算するためのヘルパー クラスです。

参加するコードは次のとおりです。

ただし、出力は正しくありません。

を含む行を削除するfilterと、結果はまったくありません。正しい出力は、次のように実行する必要があります。

与える：

誰かが動作を説明できますか? どうもありがとう。

注：org.apache.commons.lang3.tuple.ImmutablePair不思議に思う場合に使用します。

次のrxscalaコードが期待どおりに動作しない理由を理解できません:

をグループ化し、projectEvents各projectNameプロジェクトのアイテムを印刷したいと考えています。しかし、このコードを実行すると、次のように表示されます。

######### event in projectプリントはありません。

理由がわかりません。見逃したものはありますか？

RxJSでは、作成後に間隔設定を変更するにはどうすればよいですか?

これまでのところ私はこれを持っていますが、うまくいきません

「TypeError: observable.interval is not a function at sixage.js:10:14」と表示されます

jsbin

編集：

これは、受け入れられた回答を使用した後の最終製品でした。

jsbin