問題タブ [spring-cloud-stream]

私が見てきたことに基づいて、Spring Cloud Dataflow (SCDF) でストリームを作成すると、基盤となるアプリケーションがデプロイされ、通信サービス (RabbitMQ など) がバインドされ、Spring Cloud Stream 環境変数が設定され、アプリケーションが開始されます。これはすべて、cf push コマンドを使用して手動で簡単に実行できます。

その間、私は Spring Cloud Dataflow でいくつかの欠点に直面しています。

• SCDF サーバーは PCF のメモリを大量に消費します (6 つのアプリケーションしかないストリームがありますが、サーバーに約 10GB 必要です)
• アプリケーションの命名、メモリ、インスタンスなどに柔軟性がない (通常は manifest.yml で設定するすべてのもの)
• ビルド ツール (Bamboo など) との統合には、PCF CLI だけでなく SCDF CLI を使用する必要があるため、追加の作業が必要になります。
• 既存のストリームは変更できません。Blue-Green デプロイを行うには、アプリケーションを手動でデプロイする必要があります (サービスをバインドし、環境変数を手動で設定します)。その後、Blue-Green デプロイが完了すると、SCDF はストリームを Failed として表示します。これは、基盤となるアプリケーションの 1 つが変更されたことを認識していないためです。
• 失敗したストリームを再デプロイしようとしたときの MySQL Primary Key Constraint エラーなど、私が遭遇したさまざまなエラー

それで、私は何が欠けていますか？アプリケーションを手動でデプロイするだけで、Spring Cloud Dataflow を使用することが有益なのはなぜですか?

AggregateApplication私の知る限り、またはを使用して、Spring Cloud Stream のいくつかのコンポーネントを 1 つのアプリケーションとして使用するオプションがありますAggregateApplicationBuilder。

私が理解したことから、春はこの状況でステップ間の通信にブローカー (Rabbit または Kafka) を使用せず、前のステップの結果を引数として次のステップにほぼ直接渡すだけです。

私の場合、ブローカーを使用してアプリケーションの 1 つのインスタンスでより多くのコンポーネントを実行する別の方法はありますか? これが Cloud Stream に最適なアーキテクチャではないことは承知していますが、現在、Dataflow を実行できるインフラストラクチャがなく、ブローカーの耐久性も使用したいと考えています。

次のシナリオがあります。

• Producer は、 http://docs.confluent.io/3.0.0/kafka-rest/docs/intro.html#produceで説明されているように、Confluent の REST プロキシ (Confluent のスキーマ レジストリにスキーマを登録する) を介して、Avro でエンコードされたメッセージを Kafka トピックに送信します。 -and-consume-avro-メッセージ
• Spring Cloud Stream 対応のメッセージは、トピックで新しいメッセージをリッスンします

私のアプリケーションは次のようになります。

一方、MyMessage は、Avro スキーマから Avro によって作成されたクラスです。

私のapplication.propertiesは次のようになります：

私の問題は、新しいメッセージを受信するたびに、次の例外がスローされることです。

私が理解していることから、問題は、Confluent スタックがメッセージ ペイロードの一部としてメッセージのスキーマの ID を含み、クライアントがスキーマ ID の後に実際の Avro メッセージの読み取りを開始することが期待されることです。Confluent の KafkaAvroDeserializer を使用するように Kafka バインディングを構成する必要があるようですが、これを実現する方法がわかりません。

(Confluent の avro コンソール コンシューマを使用してメッセージを完全に取得できるため、Avro エンコーディングの問題ではないようです)

また、@EnableSchemaRegistry アノテーションをいじって ConfluentSchemaRegistryClient Bean を構成しようとしましたが、これはスキーマが格納/取得される場所のみを制御し、実際の逆シリアル化は制御しないように見えます。

これはどういうわけか機能しているはずですか？

AggregateApplicationBuilderのセクション 3.1.6 で説明されている Cloud Stream で遊んでいます。セクションに従ってアプリケーションを実行すると、以下のエラーが表示されます。これには、ブローカーの構成が必要です。

リファレンス ガイドで説明されているように、情報を渡しながらブローカーを排除する方法を理解するために、ここで私を助けてくれませんか?

ありがとうカーシック

spring bus にその rabbitmq キューの名前を変更するように指示する方法はありますか? 起動時には、次のようなランダムな値のように見えます。

springCloudBus.anonymous.4zzIP0z-TH6oIza5mCun7Q

これをより人間が読める予測可能なキュー名に変更するためにSpring Busを取得しようとしています。例えば：

testQueue

または、メッセージを保持しているサービスに関する知識を持つもの。

bootRun の application.yml に以下を追加してみました。

役に立たない。助けてください！！

プライベート Cloud Foundry スタックで FRP (AWS Lambda に似たもの) のラインでオンデマンド アプリケーションを開発する必要があります。これは、他の方法では常に少量のアプリケーションを実行するコストを節約するためです。

メッセージが JMS ベースのメッセージング システム (IBM MQ など) に届いた場合にのみ、アプリケーションをトリガー/開始する必要があります。少量のアプリケーションであるため、それ以外の場合は停止したままにする必要があります。

詳細な調査と Spring ドキュメントの検索に基づいて、メッセージがソース キューに表示されるたびにマイクロサービスをトリガー/開始し、完了時にシャットダウンする望ましいソリューションには、次の組み合わせが必要になるように思われます。Spring Cloud Data Flow + Spring Cloud Stream + Spring Cloud Task をテクノロジースタックとして。

さまざまなサンプルのうち、以下の組み合わせは、ローカル マシンで PoC を開始するために必要なもののようです。

https://github.com/spring-cloud/spring-cloud-task/tree/master/spring-cloud-task-samples/tasksink https://github.com/spring-cloud/spring-cloud-stream-samples /tree/master/sink https://github.com/spring-cloud/spring-cloud-dataflow/tree/master/spring-cloud-dataflow-server-local

ただし、すべてをプラグインして結果を達成する方法はまだわかりません。

ある程度の進歩はありますが、このソリューションのアーキテクチャを完全に明確にすることはまだ明確ではありません。

私が理解している限りでは、私たちは...

1. Spring Cloud DataFlow サーバーを実行する
2. Spring Cloud DataFlow Shell または UI を実行する
3. Spring Cloud Stream シンク アプリケーションを作成して実行します (Maven にもインストールする必要がありますか?)
4. Spring Cloud Task シンク アプリケーションを作成し、Maven Repo にインストールします
5. Spring Cloud Task も作成する必要がありますか?
6. モジュールを登録します (しかし、この場合のモジュールとは何ですか?)
7. DF サーバーでストリームを作成し、その上に展開します
8. もっと何か？

...結果を達成するために。

誰かがこれで私を助けて、ここから逃したビルディングブロックを提案できますか? そして、上記の質問に対する答えがわかっている場合。

また、私の理解にギャップがあり@EnableTaskますか?アプリケーションがトリガーされたときにのみ実行され、それ以外の場合は停止したままになるのに十分なほど正しく想定していますか?

SpringCloudStream の Brooklyn.Release バージョンに取り組んでいます。私のユースケースには、複数のシンクを備えた HttpSource があります。Starter App の依存関係をアプリケーションに追加し、以下のように使用すると:

私の集計アプリは

以下のように常に次の応答を取得します。

HttpSourceConfiguration の ComponentScan を追加しました (すぐに使用できます)。しかし、成功しません。

Rabbit Binder で同じ SourceApplication を使用すると、期待どおりに動作します。これを機能させるために私を案内してもらえますか？

あなたの助けと時間を感謝します。

よろしくカーシック