問題タブ [poller]

こんにちは、xml 構成を解決するのに苦労しています。

ここに私の春の統合構成xmlがあります:

ここに私のメインクラスがあります::

メインクラスを使用して実行しているだけで、この例外が発生しています::

私は春の統合を学ぶのが初めてで、構成を修正する方法も知りません..

私の要件は、一定の時間間隔、たとえば 9 (時間) 後にポーラーを停止したいということです。今のところ、1 分後にポーラーを停止しようとしています。以下は私のコードです：

しかし、今私が観察しているのは、プログラムを起動すると、起動時にすぐにコンソールに次のようなメッセージが表示されることです。

起動ログの最初のどこかで、次のようになります。

2014 年 5 月 28 日 10:27:55 AM org.springframework.integration.endpoint.AbstractE ndpoint 開始情報: 開始された incomingFiles.adapter

起動ログの最後のどこかに次のように記録されています。

2014 年 5 月 28 日 10:27:55 AM org.springframework.integration.endpoint.AbstractE ndpoint 停止情報: 停止した incomingFiles.adapter

fixed-delay="#{10 * 1000}" が 10 秒であるのに、incomingFiles.adapter がすぐに停止したのはなぜですか。時間はまったく同じで、遅延はまったくありません。したがって、理想的には、ポーラーは 10 秒後に停止する必要があります。すぐにではありません。また、ディレクトリには4つのファイルがあり、1つだけをピックアップしています。

何が悪いのか提案してください。

変更、つまりファイルの追加、変更、削除についてフォルダーをポーリングする必要があります。

上記のさまざまなタイプのイベントを区別したい場合は、カスタム ポーラーを実装する必要があります。つまり、AbstractPoller を実装します。別のプロジェクトでこれを行うポーラーを既に実装していますが、他の機能を使用する必要があるため、統合とバッチを実行したいと考えています。

これを行う最善の方法は何ですか？

ありがとう

Spring Integration を利用して、実際に転送することなく、リモートの場所に表示されるファイルに関するメッセージを開始したいと思います。必要なのは、たとえば、ファイルへのパスとファイル名を示すヘッダー値を含むメッセージの生成だけです。

これを達成するための最良の方法は何ですか? FTP インバウンド チャネル アダプターとサービス アクティベーターを組み合わせて、必要なヘッダー値を書き込もうとしましたが、これにより、ファイルがローカルの一時ディレクトリに転送され、サービス アクティベーターがそれを確認するまでに、メッセージは次のようになります。ローカルファイルとリモート パス情報を参照する java.io.Fileがなくなりました。このローカル転送が発生する前にメッセージを変換することは可能ですか?

私は NetMQ 3.3.0.11 の調査を開始しましたが、Poller の使用に関する問題に遭遇しました。

私は、ポーラーが約 1 秒間ポーリングしてから停止し、1 秒間のポーリングを再開する前に何か他のことを実行できるようにしようとしています。

次のコードがあります。

私が直面している問題は、メッセージを受け取る準備ができていても、そのポーリング期間中に Client.ReceiveReady イベントが発生しないことです。そして、InvalidOperationException格言Poller is startedが上げられます。

私が間違っていることは何か分かりますか？

Magento API と通信してデータを抽出する Windows サービスがあります。Windows サービスは、初めて起動したときはエラーや問題なく正常に実行されます。Windowsサービスの次回実行時刻を計算するポーラーが組み込まれており、configファイルにキー値「pollerExecutionTime」が指定されています。私が気付いたのは、Windowsサービスを自動開始タイプに設定しているにもかかわらず、スケジュールされた時間に開始されないことです。ポーラーのコードは次のとおりです。Windowsサービスのデバッグに精通していないので、誰かが私が間違っていることを教えてくれるなら、それは非常に役に立ちます。

ポーラーを使用してチャネルからメッセージをプルするサービスアクティベーターがあります。チャネルには、データベースへの永続ストアによってサポートされるキューがあります。ポーラーは、チャネルからのメッセージ処理にある程度の同時実行性を追加するために、タスク エグゼキューターでも構成されます。

task-executor は queue-capacity で構成されます。

ポーラーはデータベースからチャネルからメッセージを取得し、これはトランザクション対応になるように構成されているため、タスク エグゼキューターで使用可能なスレッドがなくなった場合、タスク エグゼキューターでキューに入れられるメッセージのトランザクションはどうなりますか。スレッドに対する task-executor の要求はキューに入れられます。これらのメッセージには独自のスレッドがないため、トランザクションはどうなりますか? タスク エグゼキューターによってキューに入れられたポーラーによる永続的なチャネル ストアからのメッセージの削除がコミットされると仮定します。では、サーバーがタスク エグゼキューターでキューに入れられたランナブルをキューに入れている間にサーバーに障害が発生した場合、それらは失われますか?

トランザクションの永続的なチャネル キューの考え方は、サーバーがダウンした場合にメッセージが失われないようにすることなので、キューに入れられたメッセージ (タスク エグゼキュータ内) は、チャネル データベースでバックアップされた queue/store でアクティブなトランザクションに関してどのように処理されますか?

以下の Spring Integration doc の 7.1.7 Asynchronous polling セクションでは、Poller と TaskExecutor が調整されていない場合にメモリ リークが発生する可能性があると説明されています。しかし、私はそれを理解していませんでした。

http://docs.spring.io/autorepo/docs/spring-integration/3.0.x/reference/html/messaging-endpoints-chapter.html#async-polling

上記の構成は、調整されていない構成の 1 つを示しています。

すべてのスレッドがブロックされ、新しいメッセージが到着するか、タイムアウトが期限切れになるのを待っている場合でも、ポーラーは新しいタスクをスケジュールし続けます。5 秒のタイムアウトでタスクを実行する 20 のスレッドがあるとすると、それらは 1 秒あたり 4 の割合で実行されます (5000/20 = 250 ミリ秒)。ただし、新しいタスクは 1 秒あたり 20 の割合でスケジュールされているため、タスク エグゼキューターの内部キューは (プロセスがアイドル状態の間) 1 秒あたり 16 の割合で増加するため、基本的にメモリ リークが発生します。

これを処理する方法の 1 つは、Task Executor の queue-capacity 属性を 0 に設定することです。

誰か詳しく教えてください。

上記のコードに対する私の理解は次のとおりです。

pool-size は 20 であるため、20 個のスレッドが実行されます。

receive-timeout は 5 秒です。したがって、タスクを完了するために 20 のスレッドに 5 秒が与えられます。

固定レートは 50 ミリ秒です。したがって、新しいタスクは 1 秒あたり 20 のレートでスケジュールされます。

いくつか質問があります:

Q. 20 スレッドの実行に 5 秒以上かかるとどうなりますか?

Q. ドキュメントでは、前述のタスクは 1 秒あたり 4 つの割合で実行されます。ただし、4 秒未満で実行されるタスクもあれば、それ以上かかるタスクもあります。

Q. メモリ リークはどのように発生しますか? 使用可能なスレッドとキューがない場合、Executor は代わりに拒否ポリシーに従ってそれを拒否します。

Q. queue-capacity を 0 に設定すると、どのように役立ちますか? 私の理解によると、すべてのスレッドがビジーである場合、エグゼキューターはキュー容量に達するまでそれらをキューに入れます。