問題タブ [spring-batch-tasklet]

現在実行中の春のバッチを終了する際に問題が発生しています。私が書いた

春のドキュメントを読んだ後の私のコードで。

私が直面している問題は、ジョブの終了が期待どおりに行われることもあれば、ジョブの終了が行われないこともあります。実際、joboperator で stop を呼び出すたびに、BATCH_JOB_EXECUTION テーブルが更新されます。終了が正常に行われると、バッチ プロセスで jobExecution を強制終了することにより、ジョブのステータスがSTOPPEDに更新されます。それ以外の場合は、バッチの残りのさまざまなフローを完了し、BATCH_JOB_EXECUTION テーブルのステータスをFAILEDに更新します。

しかし、job オペレーターで stop を呼び出すたびに、コンソールにメッセージが表示されます

私のプロジェクトには、一連のフローとステップが含まれています。

すべてのバッチ プロセスは次のように
なります。 JobBuilderFactory には 3
つのフローがあります。各フローにはステップビルダーと 2 つのタスクレットがあります。
各ステップビルダーには、パーティショナーとチャンク (サイズは 100) ベースの itemReader、itemProcessor、および itemWriter があります。

jobBuilderFactory で最初のフローを実行しているときに、停止メソッドを呼び出しています。完了するまでの全プロセスには約 30 分かかります。そのため、停止メソッドを呼び出してから約 20 ～ 25 分かかり、すべてのフローでチャンク サイズが 100 になり、50 万件を超えるレコードを処理しています。

それで、私の質問は、なぜjobExecutionがstopメソッドと呼ばれたときに停止するのか（これが私が望んでいたことです）、なぜjobExecutionを残りの時間停止できないのかということです。

前もって感謝します

以下は、etl.xml のジョブの構成です。

<batch:job id="procuerJob">

</batch:ジョブ>

以下は、トピックにメッセージを送信するために使用されるコードです。

ListenableFuture<SendResult<String, message>> listenableFuture = kafkaTemplate.send(メッセージ);

listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, message >>() {

}

kafkaTemplate.send(message) が実行されると、リスナーが呼び出され、ジョブが完了します。ジョブが完了した後に onSuccess()、onFailure() が呼び出されるのがわかります。kafkaトピックから確認を受け取った後にリスナーが呼び出されるように、ジョブの構成を変更するにはどうすればよいですか?

私はSpringバッチを初めて使用し、DBから巨大なデータセットをプルしてファイルに書き込む必要があるSpringバッチジョブを実装しています。以下は、私にとって期待どおりに機能しているサンプルジョブ構成です。

データ セットが巨大であるため、タスク エグゼキューターのスレッド プール値を 10 に設定し、グリッド サイズを 50 に設定しました。このセットアップでは、10 個のスレッドが一度に 10 個のファイルに書き込み、リーダーはファイルをチャンク単位で読み取るため、リーダー プロセッサとライターフローは複数回反復しています (次のパーティションに移動する前に、10 のグループの場合)。

ここで、1 つのスレッドのすべての反復 (読み取り、処理、書き込み) が完了したら、つまり各パーティションの完了後にファイルを圧縮できるタスクレットを追加したいと思います。

最後に実行するクリーンアップ タスクレットがありますが、そこに圧縮ロジックがあるということは、最初に各パーティションから生成されたすべてのファイルを取得してから圧縮を実行することを意味します。提案してください。