問題タブ [akka-camel]

ファイルの処理に akka-camel を使用しています。私の最初のテストはうまく機能していましたが、実際の xml ファイルを渡し始めたときに、型変換が発生しました。

これが私の消費者です（非常に単純ですが、msg.bodyAs [String]で突っ込んでいます）

エラー：

xml の実際の内容と関係があるかどうか疑問に思っています。それらはまったく大きくありません（約70kb）。XML 自体の実際の例を提供できるとは思えません。なぜ非常に小さく、文字列に変換されることに問題があるのか​​ について困惑しています。他のダミーのサンプル xml ファイルは問題なく動作しました。

編集: 私が持っていた提案の 1 つは、私が行った StreamCache を有効にすることでした。ただし、まだ機能していないようです。Ankush がコメントしたように、エラーは紛らわしいです。それが実際にストリームの問題なのか、それとも本当に変換の問題なのかはわかりません。

http://camel.apache.org/stream-caching.html

以下を追加しました

akka-camel を使用して rabbitmq 交換にサブスクライブしています。これらのアクターはいくつか作成されます... 要求ごとに 1 つroutingKeyです。交換とキューは変更されません。新しいものが要求されるたびroutingKeyに、新しいアクターを作成します。新しいチャネルが作成される代わりに、まったく新しい接続が作成されますが、これは望ましくありません。Consumer アクターが作成されるたびに新しい接続が作成される理由がよくわかりません。

アクター コードは次のとおりです。

私はこのようなアクターを作成しています:

アップデート

これがまさに私が達成する必要があるものです：

クライアント接続が受け入れられたときに、バックエンド アーキテクチャ内の他のコンポーネントからメッセージを受信する必要がある TCP/IP サーバーを作成しています。これを行うには、RabbitMQ を使用したいと思います。サーバーへの接続が成功すると、クライアントは ID を送信します。これは、ルーティング キーの一部として使用されます (例: command.<id>)。最初のクライアントが接続すると、RabbitMQ 接続とキューが作成され、ルーティング キーは次のようになりますcommand.first_id。次のクライアントが接続するときに、RabbitMQ への新しい接続を作成せずに、既に受け入れられているルーティング キーのリストにルーティング キーを追加 したいと考えています。command.second_d

akka + camel を使用して activemq からのメッセージを消費しています。メッセージを複製せずにこのコンシューマーを複数のマシンにデプロイする方法を見つけようとしています。この場合、私はトピックからのメッセージを消費しており、activemq は、さまざまな単一の独立したシステムではなく、さまざまなマシンに 1 つの akka システムがあることを認識している必要があります。

私はakkaクラスターを使用してそれを達成しようとしましたが、バックエンドのクラスターにサブスクライブするフロントエンドを使用した例は役に立ちません。私の「バックエンド」アクターはactivemqコンシューマー自体であり、activemqにクラスターにサブスクライブするように指示できないからです。

何か案は？

camel-jetty コンシューマーを使用した akka (Java) アプリケーションがあります。最小限の負荷 (約 10 TPS) で、クライアントに HTTP 503 エラーが表示され始めます。私たちのラボで問題を再現しようとしましたが、jetty は重複する HTTP リクエストを処理できないようです。以下は、Apache ベンチ (ab) からの出力です。

ab は、1 つのスレッドを使用して 10 個のリクエストを送信します (つまり、一度に 1 つのリクエスト)。

ab は、2 つのスレッドを使用して 10 個のリクエストを送信します (同時に最大 2 つのリクエスト)。

桟橋がこれほど悪いとは思いません。うまくいけば、それは単なる構成の問題です。これは私のキャメル コンシューマー URI の設定です。

akka 2.3.12 と camel-jetty 2.15.2 を使用しています

キャメルルートに次のものがありますが、ヘッダー値を取得していないようです。これは正しい方法ですか？

ZeroMQapache を使用してサーバーに接続しようとしていますcamel。camel-zeromqプロジェクトが提供するコンポーネントを使用していますが、これにはシステムにネイティブ ライブラリをcamel-extra事前にインストールする必要があるようです。ZeroMQ

誰かがcamel-zeromqコンポーネントを使用する方法を知っていますか?つまり、これらのネイティブ ライブラリをインストールする必要はありませんか?

これはなかなかありえないことだと思います。

そうだとすれば：

これらのライブラリをシステムに直接インストールするのを避けるために、何らかの方法でこれらのライブラリをプロジェクトに追加することは可能ですか?

このコンポーネントのもう 1 つの欠点camelは、あまり頻繁に更新されていないように見えることです。jeromqライブラリを使用してクライアントをサーバーに接続しました。これらのライブラリを使用するキャメルルートを作成する方法を知っている人はいますか?

誰かが私にアドバイスを教えてくれれば幸いです。私が直面しているこの問題を解決するための最良のアプローチを見つけようとして、ちょっと行き詰まっています。

私は Akka (最新の安定バージョン)akka-camelと JMS (この会話の目的のために、ActiveMQ としましょうが、理想的にはソリューションは汎用的である必要があります) を使用しています。

ユースケース

次のユースケースがあります。キューで次のQようなメッセージを受け取ります。

(A1,A2)私の最終目標は、 、 、などでそれらをペアにすること(B1,B2)です。重複や未配信メッセージなどの複雑な問題にもかかわらず、ペア全体が一致して処理されるまで、ブローカーがすべての未確認メッセージを保持し続けるようにしなければならないという複雑な問題があります。

例

で44 つのメッセージを受信して​​処理し、ペア を正常に処理しましたが、とはまだ一致せず保留中であり、JMS ではまでのすべてのメッセージを確認(C1, C2)することを意味するため、まだ何も確認できません。実際のところ、私が送り返すことができる最初の承認は、 が受信されたときです。この時点で、承認できます(保留中の のみ)。A1B1C2C25A2A1A1B1

問題

さて、私が理解できないように見えるのは、この種の遅延および非同期の確認応答をakka-camel. 私はオンラインで読んでいますが、メッセージを手動で確認する方法についての説明 (ドキュメントと例) を見つけることができますが、以前に処理されたメッセージをブローカーに確認する方法を示すものは何もありません。

この場合、Ackは でありobject、そのセマンティックは本当に単純です:現在のメッセージを確認しますが、Xは以前のメッセージですが、必ずしも現在のメッセージであるとは限りません。

このユースケースはサポートされていますか、それともサポート可能akka-camelですか、それとも自分で構築する必要がありますか?

ありがとう

私はakka camel-ftp消費者を持っています。すべての例外をコードで処理したいと考えています (認証例外、ftp 上のファイルを読み取れないなど)。スタックトレースはログにしか表示されず、処理できません。を使ったほうがいいかもしれませんscalaz-camel。

さらに、すべてのファイルがいつ処理され、次に ftp フォルダーを読み取るまでアクターがアイドル状態になるかを知りたいです。