問題タブ [spring-cloud-stream]

最近、Spring Cloud Stream と RabbitMQ バインダーをいじり始めました。

2 つのサービスがメッセージを渡したいときにすべてを正しく理解していれば、一方はメッセージを送信するようにソースを構成し、もう一方はメッセージを受信するようにシンクを構成する必要があります。両方が同じチャネルを使用する必要があります。

という名前のチャンネルtestchannelがあります。ただし、ソースが RabbitMQ バインディングを作成したことに気付きました。

• 交換testchannel、
• ルーティングキーtestchannel,
• キューtestchannel.default(持続可能)、

シンクがRabbitMQバインディングを作成している間:

• 交換testchannel、
• ルーティングキー#,
• キューtestchannel.anonymous.RANDOM_ID(排他的)。

簡潔にするために、接頭辞をスキップしました。

両方のアプリケーションを実行したとき。最初に交換にメッセージを送信しtestchannel、それが両方のキューにルーティングされます (ルーティング キーは であると仮定しますtestchannel)。2 番目のアプリケーションはランダム キューからメッセージを消費しますが、デフォルト キューからのメッセージは消費されません。

私の他の問題は - 2 番目のアプリはsinkoutputのみを使用していますが、何も指定していないため、デフォルトである出力チャネルのバインディングも作成します。

同じ Gradle スクリプトを使用して両方のアプリをビルドします。

最初のアプリ プロパティ:

最初のアプリのソース コード:

2 番目のアプリのプロパティ:

2 番目のアプリのソース コード:

だから私の質問はです。

• ソースとシンクが同じチャネルを使用して、結果として同じブローカー キューを使用するべきではありませんか? それを達成するための適切な構成は何ですか？(私の目標は複数のシンクサービス インスタンスを持つことですが、メッセージを消費するのは 1 つだけです。)
• シンクのみを使用している場合、フレームワークは出力バインディングを作成する必要がありますか? はいの場合、それを無効にする方法。

構成の変更が発生したときにバスを使用してクライアントを更新する信頼できる構成サービスを構成しようとしています。ローカル ファイル システムを監視する 2 つの構成サーバーを開始しました。クライアントが起動時に構成サービスを検出できるようにするための 2 つの eureka サーバー (つまり、eureka fist 構成タイプ)。私は amqp バスとして rabbitMQ を使用しました。現在の動作は次のとおりです。構成ファイルを更新してhttp://config-server1/bus/refreshへの投稿を要求すると、構成サーバーが通知を送信し、1 つのクライアントのみがそれを選択します。そのため、3 つのクライアントを更新するには、3 つの投稿を作成する必要があります。質問: /bus/refresh への 1 回の投稿ですべてのクライアントが更新されるようにバスを構成するにはどうすればよいですか。

前もって感謝します。

私はRabbitMQバインダーでSpring Cloud Streamを使用しています。ペイロードと Java ネイティブのシリアル化ではうまくbyte[]機能しますが、JSON ペイロードで作業する必要があります。

これが私のプロセッサクラスです。

InputDtoOutputDtoJackson アノテーション付きの POJO です。

• JSON 変換戦略を構成するにはどうすればよいですか?
• 受け入れられて処理されるメッセージ ヘッダーはどのように見えるべきですか?

CloudFoundry で実行されている Spring Cloud Dataflow の最も基本的な例を取得しようとしています。ここの手順に従いました: http://docs.spring.io/spring-cloud-dataflow-admin-cloudfoundry/docs/current-SNAPSHOT/reference/htmlsingle/#getting-startedで管理アプリを利用できるようにします組織/スペース。次に、 http://cloud.spring.io/spring-cloud-dataflow/から最も基本的な例、つまり「ticktock」ストリームを作成しようとしました。

ticktock-time と ticktock-log の両方のアプリがスペースに作成され、必要なサービス「redis」がこれらのアプリにバインドされ、起動しようとしていることがわかります。残念ながら、「redis」サービスへのアクセスに問題があるため、完全には起動しません。ログには次のものがあります。

コンテキストの初期化中に例外が発生しました - 更新試行のキャンセル: org.springframework.context.ApplicationContextException: Bean 'outputBindingLifecycle' の開始に失敗しました。ネストされた例外は org.springframework.context.ApplicationContextException: Bean 'inputBindingLifecycle' の開始に失敗しました。ネストされた例外は org.springframework.data.redis.RedisConnectionFailureException: Cannot get Jedis connection; です。ネストされた例外は redis.clients.jedis.exceptions.JedisConnectionException です: プールからリソースを取得できませんでした

それは最終的にによって引き起こされます

原因: redis.clients.jedis.exceptions.JedisConnectionException: java.net.ConnectException: 接続が拒否されました

間にいくつかの構成手順がありませんか?

アレクサンダー

spring-cloud-stream 1.0.0.M4 をさまざまなバインダーで使用しようとしています。プロデューサーとコンシューマーの間でデータ転送オブジェクトを共有していないので (誰が共有するのでしょうか?)、バインディングにコンテンツ タイプの構成を含める必要が生じました。

プロデューサー構成:

生産者コード:

消費者構成:

消費者コード:

私はこれをウサギ、redis、および kafka バインダーで試しましたが、json ハックはウサギでのみ機能することがわかりました。kafka と redis を使用すると、コンシューマーで次のエラーが発生します。

カフカ消費者エラー:

redis コンシューマー エラー:

spring-cloud-stream で使用する必要がある要求応答パターンはありますか? spring-cloud-stream で見つけることができるすべてのドキュメントは、MessageChannel.send ファイア アンド フォーゲット タイプのプロデューサーを対象としており、Spring 統合の @MessagingGateway には精通していますが、それがどのようになるかはわかりませんspring-cloud-stream を使用します。これは、割り当てられた識別子でエンティティを保存する REST POST エンドポイントがあり、割り当てられた識別子を呼び出し元に返す必要があるが、メッセージングを使用したい場合に役立ちます。

以前の質問Spring Cloud Stream message from/to JSON conversion configurationに従い、説明どおりにストリームを構成しましたが、正しく動作させることはできません。

私のセットアップは次のとおりです。私は2つのアプリAとB. アプリAは入力チャネルone、出力を使用しますtwo。アプリBは input を使用しますtwo。チャネルtwoは content type で構成されていますapplication/json。

アプリ A. プロパティ。

リスナーメソッド。

アプリ B. プロパティ。

リスナーメソッド。

適切な JSON 文字列を含むメッセージを channel に手動で送信すると、メッセージoneは正しく処理さtwoれ、JSON メッセージとしてチャネルに送信されます (ヘッダーは上記の質問で説明したものとまったく同じです)。その後、twoアプリ B によってチャネルで受信され、例外がスローされます。Method handle(java.lang.String) cannot be found

もちろん、Dto と String を入力として処理する両方のメソッドを作成すると機能しますが、常に String メソッドが呼び出され、自分でペイロードを逆シリアル化する必要があります。

私はどこか間違っていますか？このような署名を使用してメソッドをセットアップするにはどうすればよいpublic Dto handle(Dto incoming)ですか?