問題タブ [project-reactor]

私は調査を行い、ドキュメントを読みましたが、それらはあまり理解できません。私が達成しようとしているのは、次の機能です。

Spring Reactor プロジェクトを使用し、eventBus を使用しています。イベント バスがモジュール A にイベントをスローしています。

モジュール A はイベントを受け取り、一意の値を保持する Hot Stream に挿入する必要があります。250 ミリ秒ごとに、ストリームはすべての値を取得して計算を行う必要があります。

例: eventBus は次の番号のイベントをスローしています: 1,2,3,2,3,2

ストリームは一意の値を取得して保持する必要があります-> 1、2、3 250ミリ秒後、ストリームは数値と空の値を出力する必要があります

開始方法を知っている人はいますか? 例を試してみましたが、実際には何も機能せず、何かを理解していないと思います。誰にも例がありますか？

TNX

編集：

次のことをしようとすると、常に例外が発生します:

例外：

ご覧のとおり、この問題に合格しない限り、試行を続けることはできません。

ところで：これは、私が使用するbuffer(1, TimeUnit.SECONDS)場合にのみ発生buffer(50)します。たとえば、使用すると機能します..

Spring 5 の新しいリアクティブ サポートを使い始めたばかりで、非同期データ生成をシミュレートしたいと考えていたところ、2 つの問題のある動作に気付きました。

1) s.onNext( String ) を複数回呼び出す:

この場合、スタックトレースは次のとおりです。

2) s.onNext( Alert.class -any DTO- ) を複数回呼び出す:

ログにエラーは表示されなくなりましたが、発信者は 500 応答コードとコンテンツ '[' を受け取ります。

ログ：

onNext() を複数回呼び出せないのはなぜですか? どうすればそれができますか？

注：私はonNext一度呼び出すだけで問題なく動作します：

また

Reactor Fluxの連載が可能です。たとえば、私の Flux は何らかの状態にあり、現在何らかのイベントを処理しています。そしていきなりサービス終了。Flux の現在の状態は、データベースまたはファイルに保存されます。そして、アプリケーションの再起動時に、そのファイル/テーブルからすべての Flux を取得し、それらをサブスクライブして、最後の状態から処理を再開します。これは原子炉で可能ですか？

Reactor Project を理解しようとしていますが、サブスクリプションをキャンセルする方法を探しています。たとえば Flux のサブスクリプションを作成した後、onCancel シグナルの送信に使用できる Cancellation オブジェクトへの参照を取得できることはわかっていますが、これはサブスクリプションを作成した後でのみであり、その参照をある種のコレクションに保持する必要があります。

Cancellation オブジェクトを取得するより良い方法はありますか? または、サブスクリプションをキャンセルするだけです。たぶん、すべてのアクティブなサブスクリプションへの参照を含むある種の場所-ええ、それは素晴らしいでしょう...

次のコードがあります。

コントローラ：

EventNotifier:

イベントコンシューマー:

メソッドsaveMessageが呼び出されていないことがわかります。サーバーの起動プロセス中に、一度呼び出されるのを見ました。

これがリアクター関連かスプリング関連かはわかりません。

編集：保存プロセスをサービスに移動しましたが、何も変わりませんでした

edit2: サービス アクションを記録して、何が起こっているかを確認しました。

コンソールの出力は次のとおりです。

JPA からのログはありません。