BlockingQueueで同期化されたSpringバッチ並列フローをプログラム的に停止

Question

基本的にファイルから読み込み、各行を処理して出力（別のファイル）に書き込むバッチジョブがあります。 処理ステップが高価なので、複数のスレッドで実行する必要がありますが、読み取りと書き込みのステップではファイルを使用しているため、これらのステップは1つのスレッドで実行する必要があります。 私は3つのフローを持っていました。それぞれ、2つのBlockingQueuesで同期され、それぞれが1つのステップで、それぞれ並列に実行されています。 読み取りステップはファイルから読み取り、1つのキューに書き込みます。 処理ステップはマルチスレッドであり、キューから読み取り、処理して別のキューに書き込みます。 書き込み手順は、2番目のキューから読み取り、出力を別のファイルに書き込みます。

すべてが完了したときに仕事を止めるクリーンで「速い」方法が見つからないことを除けば、それはかなりうまくいく。今、私は 'poll'を両方のキューでタイムアウトして使用しています。何秒間もアイテムが存在しなければ、完了していると仮定します。これにより、指定された秒数だけジョブの終了が遅れ、ジョブが遅延する可能性がある外部負荷（マシンの負荷など）によって非常に短い時間を使用することはできません。

私はPoison Pillのようなものを使用しようとしましたが、問題は、 'Null'（ファイルの終わりを示す）を取得したときにPoison Pillを返すFlatFileItemReaderのdoReadメソッドをオーバーライドすると問題ですこの読者は決して終わらず、仕事は決して終わらないでしょう。

誰かに提案がありますか？私が知っているドキュメントからは、読んだステップ（ファイル）と書き込みステップ（ファイル）のライターに「同期」を置くことができるかもしれないが、実際は別の解決策を好むだろう。

Answer 1

読者にステートフル変数を追加するだけで、ジョブの終了を追跡できます。だから、

public PoisoningReader<T> extends FlatFileItemReader<T> { 
    private boolean endJob = false; 

    @Override 
    public T doRead() { 
     if (endJob) { 
      return null; 
     } 

     T object = super.doRead(); 
     if (object == null) { 
      endJob = true; 
      return new PoisonPill(); 
     } 
     return item; 
    } 

Answer 2

、私は誰もが興味を持っているか、同様の問題に直面している場合には、私の解決策を投稿するつもりです。

私が要約したのは、Dean Clarkが示唆したようにPoison Pillを使い終わったことです。 私は最終的に1つのBlockingQueueだけを使用するようにジョブを簡素化しましたが、ステップ間で共有されるキューであるため、Poison Pillの注入方法についての問題が残りました..

基本的には、読者の周りにPoison Pillを返すためのプロセッサーとプロセッサーがそれを検出して無視するだけで、私はSpring Batchを正常に実行させ、私はPoison Pillの注入を担当するStepにリスナーを追加しました。このリスナーは「afterStep」をオーバーライドし、それを単にキューに追加します。 キューから読み取るステップは、ポイズンピルとキューの終わりを取得し、「これ以上行う作業はない」ことを示し、ヌルを返すことによって通常終了します。

もう1つのジョブでは、キューから読み取るステップには、アイテムを並列処理するThreadPoolが設定されているため、キューから読み取るすべてのスレッドを強制終了/ブロック解除する必要があります。素敵なトリックは、読者にキューから読んでもらっていました。毒薬であれば、それをキューに再注入してヌルを返します。このようにして各スレッドはPoison Pillを取得し、正しく終了します。