問題タブ [consumer]

Mule ESB 3.5.0 Community Runtime のサンプルプロジェクト「tshirt-service-consumer」のフロー「listInventory」を自作の Web サービス SOAP で再現してみます。しかし、プロジェクトを実行すると、コンソールにエラーが表示されます

別のクライアント SOAP を使用して Web サービスを呼び出すと、問題はありません。

wsdl :

フロー

私はC++カフカの消費者を探しています。私は以下に出くわしましたが、これは C++ kafka 用ですが、コンシューマーはありません。

https://github.com/adobe-research/libkafka (サンプル プロデューサーのみ)

https://github.com/edenhill/librdkafka/tree/master/src-cpp

上記の作業に基づくC++カフカコンシューマー、またはC++カフカコンシューマーの新しい方法を持っている人はいますか?

JMS トピックにサブスクライブするアプリケーション (Scala 2.10.3、Akka 2.3.1、Camel 2.13.0) があり、特定のファイルがダウンロード可能になると JMS メッセージ経由で通知されます。各 JMS メッセージには、SFTP 経由で収集できるファイルの名前とパスが含まれています。

次に、SFTP 経由でファイルをフェッチできるようにしたいのですが、JMS メッセージを受信したファイルのみをフェッチします(書き込み中のファイルをフェッチする可能性がある問題を回避するため)。

Akka Camel と Consumer モデルに適合するソリューションが必要です。SFTP エンドポイントに使用されるfile2とftp2の Camel オプションを読みましたが、次の点で助けが必要です。

• &filter=... パラメータを介して endpointUri 文字列で参照できるクラス/オブジェクトを定義するにはどうすればよいですか? コンシューマーがファイルのリストをポーリングするたびに、更新されたフィルター リストが適用されるように、フィルター オブジェクトを更新できるようにしたいと考えています。

• カスタム IdempotentRepository を定義して、デフォルトの 1000 より大きいキャッシュ サイズを許可するにはどうすればよいですか?

現在、私の SFTP Consumer Actor は次のようになっています (いくつかの値は編集されています ...):

I am very much new to Kafka and we are using Kafka 0.8.1.

What I need to do is to consume a message from topic. For that, I will have to write one consumer in Java which will consume a message from topic and then save that message to database. After a message is saved, some acknowledgement will be sent to Java consumer. If acknowledgement is true, then next message should be consumed from the topic. If acknowldgement is false(which means due to some error message,read from the topic, couldn't be saved into the database), then again that message should be read.

I think I need to use Simple Consumer,to have control over message offset and have gone through the Simple Consumer example as given in this link https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example.

In this example, offset is evaluated in run method as 'readOffset'. Do I need to play with that? For e.g. I can use LatestTime() instead of EarliestTime() and in case of false, I will reset the offset to the one before using offset - 1.

Is this how I should proceed?

複数の activemq jms キューからのメッセージを消費する方法が必要です。

activemq のドキュメントに従って、ワイルドカード コンシューマーをサポートしています。

メッセージング バスとしてキャメルを使用しています。以下の名前付きキューを確認できますか

エンドポイントを見るようにキャメルルートを設定することによってactivemq:*.processQueue ？

また、これに代わるよりクリーンな方法があれば教えてください。

この場合（任意のバージョン）、プロデューサーが「永久に」実行する必要があるExecutorServiceを使用する単一のプロデューサー複数のコンシューマーを実装するためのJavaでの最良のアプローチを見つけようとしています（必要ではありません）。すべてが完全に処理されるまで待機する必要があります。たとえば、すべてのコンシューマー スレッドが終了し、キューが空になり、生成されるアイテムが残っていません。プロデューサは、データ ソースも一定の間隔でのみポーリングする必要があります。

例として、30 分ごとにプロデューサーがレコードのデータ ソースをポーリングし、それらをキューに送信するようにします。コンシューマーが処理に 30 分以上かかっている場合は、データ ソースを再度ポーリングする前に、すべてのアイテムが処理されるまでプロデューサーに待機してもらいます (30 分が経過するとすぐにポーリングします)。

私のコードを書いてくれる人を探しているわけではありません。いくつかの基本的なヒント/ガイダンスをいただければ幸いです。

これは、私が作業しようとしている短縮された実装例です。私は問題を解決するための恐ろしい試みをすべてやめました。ThreadPoolExecutor を構築するためにハードコードされたパラメーターは、最終的に動的になることに注意してください。

私の Java プログラムには、クライアント サーバー アーキテクチャがあります。クライアント側で1つのメッセージを作成し、10000のスレッドで10000のクライアントが同時に1つのメッセージを送信するため、サーバーに送信します。サーバークラス側には、固定サイズのキューがあり、すべてのメッセージを最初にこのキューに入れる必要があります。次に、複数のチャネル (または複数のクライアント) を使用して、同時にメッセージを読み取ります。1 つのチャネル (またはコンシューマー) が 1 つのメッセージを読み取った場合、このメッセージはキューから削除する必要があります。しかし、複数のチャネルでは、1 つのチャネルよりも十分なパフォーマンスが得られません。スレッドの同期に問題があると思います。プログラムを最適化するにはどうすればよいですか? これが私のプログラムです。

クライアント側には、メッセージを送信するための実行可能なクラスが 1 つあります (クラスの重要な部分を示します)。

そしてメインクライアントクラスで

サーバー側では、キューに書き込みます

そして、このバッファを読み取り、Oracle DB メッセージに送信します。

私が作成する複数のチャネル

そして最後にサーバークラスで、バッファに書き込む2つのスレッドと、キューから読み取ってDBに書き込むためのチャネルを作成するスレッドを開始しています。

C# で RabbitMQ をリッスンする標準のスタンドアロン スレッドをセットアップしています。スレッドでリッスンするメソッドが次のようになっているとします。

RabbitMQ の C# クライアントでメッセージの消費を適切に停止するエレガントな方法は何ですか? RabbitMQ の例やドキュメント、またはこれらの SO の質問には何の役にも立たないことに注意してください。

• 選択キューからのメッセージの消費を停止する方法 - RabbitMQ
• rabbitmq、ピカパイソンで優雅に消費を一時停止および再開する方法
• RabbitMQ コンシューマーを実行して Java アプリケーションを安全に終了する最善の方法は何ですか?

ここでの問題consumer.Queue.Dequeue()は、ブロッキング コールです。これらのオプションを試しました：

• 呼び出しchannel.BasicCancel(string tag)ます。これによりSystem.IO.EndOfStreamException、ブロッキング呼び出しで a が発生します。明らかな理由から、この例外を制御フローの一部として使用したくありません。

• consumer.Queue.Dequeue(int millisecondsTimeout, out T result)ループの繰り返しの間にフラグを呼び出してチェックします。これは機能しますが、ハッキーに思えます。

スレッドを正常に終了させ、管理されていないリソースをクリーンアップして、スレッドが中止されないようにしたい.

どんな助けでも大歓迎です。ありがとう

私は現在、スレッド化プログラムの概念を紹介されており、スレッドとセマフォを使用して株式市場をシミュレートする任務を与えられています。コードは次のとおりです。

これにより、スレッドの作成により、main() が開始されるとすぐにセグメンテーション違反が発生します...ルートから作成したコードではないため、エラーをバックトレースして修正するのに問題があります-見てくれてヒントくれる人がいたらいいのに

ありがとうございました！

編集

あなたの助けのおかげで問題を解決しました:)

init() で在庫リストを初期化しました。

main() の 2 番目の cicle を変更し、

ありがとう

アプリケーションで ActiveMQ ブローカーの 3 つのネットワークを使用しています。メッセージを消費する順序を維持するために、JMSXGroupId を使用しました。次の構成を使用してブローカーのネットワークを構成しました。

このように、3 つのブローカーすべてに対して構成しました。私のポリシー エントリは次のとおりです。

3 つの異なる JVM に 3 つのコンシューマがあり、エンドポイントは concurrentConsumers=5 で構成されています。コンシューマーのブローカー URL は次のように構成されます。

消費者-1 :

消費者-2 :

消費者-3 :

ユースケース: 15 分ごとに 3000 件のメッセージを公開しています。JMSXGroupId はメッセージごとに異なり、一意です。最後の 2 つのメッセージは、同じ JMSXGroupId を持ちます。

問題: コンシューマー側では、メッセージの ~95% が適切に消費され、メッセージの ~5% がブローカーで発生します (ブローカーで保留中)。

ノート：

1. コンシューマーがメッセージを処理するのに 1000 ミリ秒かかります。
2. メッセージは、メッセージ発行の 4 回目または 5 回目の反復でブローカーに配置され始めます。

maxPageSizeをデフォルト値の 200 から 1000に増やすと、攻撃は @12 回目または 15 回目の繰り返しから始まります。

誰でもこの問題を解決するのを手伝ってくれませんか...これについてさらに詳細が必要な場合は、コメントして共有してください。

追加の詳細

• メッセージは処理されずに保持されます
• 私には 5 つのコンシューマーがあります (各ブローカーには 5 つの実際のコンシューマーと 2 つの転送コンシューマーがあります)
• テスト中にブローカーとコンシューマーが再起動されなかった
• しかし、パブリッシャーは、プロセスを開始してメッセージをパブリッシュするとダウンし、15 分後に再び開始してパブリッシュします。
• パブリッシャーはブローカーを DNS に接続し、コンシューマーはフェイルオーバー付きのコンマ区切り TCP で接続されます。
• Postgres DB を使用してメッセージを保存しています。ジャーナルとストアの詳細設定は次のとおりです。

注:構成は 3 つのブローカーすべてで同じです 。 PersistenceFactory

収納詳細