Microservices Engine:MSE XXL-JOB を使用したグレースフルシャットダウンの実装

オープンソース XXL-JOB の内部ジョブ処理フロー

オープンソース XXL-JOB におけるジョブのディストリビューションと実行プロセスには、XXL-JOB Admin と XXL-Job Executor の 2 つのモジュールが関わります。

エグゼキュータの登録：

1. XXL-JOB SDK が起動すると、ビジネスアプリケーションは ExecutorRegistryThread スレッドを初期化し、XXL-JOB Admin に継続的にハートビートメッセージを送信します。

2. ハートビートメッセージを受信すると、XXL-JOB Admin は JobRegistryHelper を通じてエグゼキュータ情報を xxl_job_registry データベーステーブルに書き込みます。

3. JobRegistryHelper 内のスレッドは、登録済みエグゼキュータのリストを提供する xxl_job_group テーブルの address_list フィールドを定期的にクエリして更新します。

オンラインエグゼキュータの選択：

1. スケジューリングスレッドがジョブをトリガーすると、XxlJobTrigger がジョブを実行します。

2. ジョブを実行する前に、XxlJobTrigger は xxl_job_group テーブルの address_list フィールドからエグゼキュータのリストを読み取ります。

3. ExecutorRouter は、指定されたルーティングポリシーに基づいてリストからエグゼキュータを選択します。

4. XxlJobTrigger は RPC リクエストを送信し、選択されたノードにジョブをディスパッチします。選択されたノードがオフラインの場合、ジョブのトリガーは失敗します。

ジョブの実行と結果のフィードバック：

1. エグゼキュータがジョブリクエストを受信すると、ジョブ ID に基づいてジョブごとに JobThread スレッドを作成します。

2. ジョブリクエストがトリガーされると、現在の JobThread スレッドが処理するためにキューに追加されます。ジョブごとにブロッキングポリシーは異なります。

3. JobThread スレッドは、キュー内のトリガー結果を継続的に読み取り、対応する JobHandler を実行してビジネスロジック処理を完了します。

4. ジョブが終了すると、JobThread スレッドは実行情報を TriggerCallbackThread の実行応答キューに送信し、次のジョブに進みます。

5. エグゼキュータが停止すると、XxlJobExecutor.destroy メソッドを実行して JobThread スレッドを停止し、スケジューリングリクエストのキューをクリアします。

TriggerCallbackThread スレッドは継続的に実行され、現在の実行結果キューをロードし、それらをバッチで XXL-JOB Admin にディスパッチします。結果の送信に失敗した場合は、ローカルディスクに保存して後で再試行します。

オープンソース XXL-JOB でのグレースフルシャットダウンの実装

MSE SchedulerX プラグインなしでオープンソース XXL-JOB を使用する場合、トラフィックの削除、キュー内のジョブの完了待機、そしてアプリケーションのシャットダウンという 3 つのステップに従うことで、手動でグレースフルシャットダウンを実装できます。

com.xxl.job.core.executor.XxlJobExecutor#destroy メソッドは、Spring Boot モードでアプリケーションプロセスが終了する際に自動的にコールバックを実行します。しかし、デフォルトのロジックではグレースフルシャットダウンが完全には実装されていません。以下の変更が必要です。

ステップ 1: アプリケーションノードからのトラフィックの削除

XxlJobExecutor#destroy の stopEmbedServer() メソッドは、ハートビート登録メカニズムを停止させ、XXL-JOB Admin に registryRemove リクエストを送信することで、現在のエグゼキュータを削除します。しかし、xxl_job_group テーブルの address_list フィールドはリアルタイムで同期されないため、トラフィックは効果的に削除されません。

これを修正するには、次のいずれかの方法で XXL-JOB Admin サーバーを変更します。

• JobRegistryHelper.registryRemove メソッドに処理ロジックを追加して、xxl_job_group テーブルの address_list フィールドを更新します。freshGroupRegistryInfo メソッドで更新ロジックを実装することもできます。

• XxlJobTrigger#trigger() メソッドを変更して、自動登録プロセス中に xxl_job_registry テーブルから直接 address_list を読み取るようにします。

ステップ 2: キュー内のジョブの完了待機

XxlJobExecutor#destroy メソッドを変更して、キュー内のすべてのジョブが完了するのを待ちます。

public void destroy(){

    // エグゼキュータサーバーを破棄
    stopEmbedServer();

    // jobThreadRepository を破棄
    if (jobThreadRepository.size() > 0) {
        List keyList = new ArrayList(jobThreadRepository.keySet());
        for (int i=0; i < keyList.size(); i++) {
            JobThread jobThread = jobThreadRepository.get(keyList.get(i));
            // キュー内のすべてのジョブが完了するのを待ちます。
            while (jobThread != null && jobThread.isRunningOrHasQueue()) {
                try {
                    TimeUnit.SECONDS.sleep(1L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    jobHandlerRepository.clear();

    // JobLogFileCleanThread を破棄
    JobLogFileCleanThread.getInstance().toStop();

    // TriggerCallbackThread を破棄
    TriggerCallbackThread.getInstance().toStop();

}

TriggerCallbackThread.getInstance().toStop() メソッドは、TriggerCallbackThread スレッドが停止した後に実行結果を同期するため、結果のフィードバックに追加の処理は必要ありません。

ステップ 3: アプリケーションプロセスの停止

アプリケーションプロセスを停止するには、アプリケーションのデプロイメントスクリプトで kill -15 を使用して JVM シャットダウンフックをトリガーします。ビジネス要件に基づいて、タイムアウト時にアプリケーションプロセスを強制的に停止することもできます。または、Spring Boot Actuator の /actuator/shutdown エンドポイントを使用して、グレースフルシャットダウン機能を統合することもできます。