問題タブ [spring-cloud-stream]

Spring Cloud Dataflow を使用してストリームを作成しようとしましたが、うまくいきませんでした (主にドキュメントがないため)。

問題 1: Docker 化された Spring Cloud Dataflow の Web GUI へのアクセス 基本的な Ubuntu コンテナーで Kafka を使用して実行されている Docker 化された Spring Cloud サーバーがあります。何らかの理由で、Windows の Web GUI (< docker-machine ip >:9393/dashboard) にアクセスできません。ただし、Nginx リバース プロキシを実行する別の Docker Ubuntu コンテナーがあり、< docker-machine ip >/index.html などに移動すると表示されます。ポートの問題ではないと思います。Spring -p 9393:9393 を使用したクラウド コンテナーのセットアップ。それ以外の場合、ポートは使用されません。

問題 2: JSON ヘッダーによるルーティング 私の最終的な目標は、Nginx から読み込まれたファイルを取得し、その JSON ヘッダー(2 つの異なる JSON ヘッダーがあります) に基づいてルーティングし、クエリを Cassandra に取り込むことです。JSONヘッダーによるソートを除いて、これをすべて行うことができます。どのアプリを使用することをお勧めしますか?

Spring Cloud Streamを使用して、メッセージング駆動型のマイクロサービスを作成しています。

• プロジェクト A: プロデューサー (Spring Cloud Stream、統合された tomcat とそのポート: 8080)
• プロジェクト B: コンシューマー (Spring Cloud Stream、統合された tomcat とそのポート: 8081)
• メッセージ ブローカー: RabbitMQ (amqp ポート: 5672)

私の質問は: プロジェクト A と B の場合、統合された Web サーバー/ポート (Tomcat) が必要ですか? それとも、一般的な Java アプリケーションで十分でしょうか?

私は英語のネイティブ スピーカーではありませんが、質問をできるだけ明確に表現するように努めています。この問題に遭遇したため、2 日間混乱しましたが、まだ解決策が見つかりません。

Hadoop YARN の Spring Could Data Flow で実行されるストリームを構築しました。

ストリームは、HTTP ソース、プロセッサ、およびファイル シンクで構成されます。

1.Http ソース
HTTP ソース コンポーネントには、application.properties で定義された dest1 と dest2 という 2 つの異なる宛先にバインドする 2 つの出力チャネルがあります。

spring.cloud.stream.bindings.output.destination=dest1 spring.cloud.stream.bindings.output2.destination=dest2

以下は、参考用の HTTP ソースのコード スニペットです。

2. プロセッサ
プロセッサには、2 つの複数入力チャネルと、異なる宛先にバインドする 2 つの出力チャネルがあります。宛先バインディングは、プロセッサ コンポーネント プロジェクトの application.properties で定義されます。

以下は、プロセッサのコード スニペットです。

3. ファイル シンク コンポーネント。

Spring の公式の fil sink コンポーネントを使用します。maven://org.springframework.cloud.stream.app:file-sink-kafka:1.0.0.BUILD-SNAPSHOT

そして、applicaiton.properties ファイルに宛先バインディングを追加するだけです。spring.cloud.stream.bindings.input.destination=fileSink

4.発見：

私が期待したデータフローは次のようになります。

Source.handleRequest() -->Processor.handleRequest()

Source.handleRequest2() -->Processor.handleRequest2() --> Sink.fileWritingMessageHandler();

"processed by transform2" という文字列のみがファイルに保存されます。

しかし、私のテストの後、データフローは実際には次のようになります。

Source.handleRequest() -->Processor.handleRequest() --> Sink.fileWritingMessageHandler();

Source.handleRequest2() -->Processor.handleRequest2() --> Sink.fileWritingMessageHandler();

"processed by transform1" と "processed by transform2" の両方の文字列がファイルに保存されます。

5.質問:

Processor.handleRequest() の出力チャネルの宛先は、fileSink ではなく hdfsSink にバインドされますが、データは引き続きファイル Sink に流れます。私はこれを理解できず、これは私が望むものではありません。Processor.handleRequest2() からのデータのみが、両方ではなくファイル シンクに流れます。私が正しく行わない場合、誰かがそれを行う方法と解決策を教えてもらえますか? 2日間私を混乱させました。

ご親切にありがとうございました。

アレックス

Spring Cloud Data Flow で DLQ を構成しようとしています。これがストリームの定義とそれを展開する方法です

custom-transform - プロセッサ コードで、私は言及しました

つまり、メッセージに ERROR が含まれている場合、RunTimeException が発生し、これらのメッセージを DLQ でキャプチャしたいということです。しかし、コードを実行しているとき、test-tran という名前の Kafka DL キューを取得していないようです。

DLQ を有効にするにはさらにプロパティを設定する必要がありますか、それとも DLQ を適切に使用するためにコードを変更する必要がありますか?

カスタム変換コード

TransformationServiceApplication.java

TransformationMessageEndPoint.java

pom.xml

モジュールの追加

ストリームの追加

ストリームのデプロイ

spring-cloud-stream kafka の BinderAwareChannelResolver を使用して、イベントに基づいてチャネルを動的に解決しています。多くのプロデューサー チャネルがあり、すべてのチャネル出力の kafka メッセージが非 Spring アプリケーションによって消費されるため、headerMode を設定する必要があります。私の設定を乱雑にするapplication.propertiesで各チャンネルをrawとして指定する方法はありますか？

RabbitMQ を使用して単純な Spring Cloud Stream アプリケーションを構成しようとしています。私が使用するコードは、主にspring-cloud-stream-samplesから取得されます。エントリポイントがあります：

例からの単純なメッセージ プロデューサー:

さらに、ここに application.yml 構成があります。

この例を実行すると、Rabbit に接続され、test という交換が作成されます。しかし、私の問題は、キューとバインドが自動的に作成されないことです。Rabbit でトラフィックが発生しているのを確認できますが、すべてのメッセージが消えてしまいます。消費者が読み取らない限り、キューに留まる必要があります。

何かを誤解しているのかもしれませんが、私が読んだすべてのトピックから、Spring Cloud Stream はキューとバインディングを自動的に作成する必要があるようです。そうでない場合、メッセージが永続化されるように構成するにはどうすればよいですか?

Spring Cloud Brixton.SR5 と Spring Boot 1.4.0.RELEASE を使用しています。