PythonでCeleryを使わずにタスクをスケジュールする方法と代替手段

すべてのバックエンド開発者がおそらく経験していることだと思います。FastAPIやFlaskでWebサービスを構築しており、プロジェクトはまだ小さく、単一のサーバーで動作しています。そうしたらプロダクトマネージャーから「毎朝9時に顧客にダイジェストを送って、期限切れのカートを1時間ごとにリセットしよう」と頼まれます。

最初の反応は、Celery BeatとRedisを使ってCeleryを追加することです。しかし、2つのバックグラウンドタスクのためだけにブローカーを立ち上げて、ワーカーを設定し、デーモンを監視するのはやりすぎです。2番目のオプションはシステムcronですが、それだとロジックがアプリケーションコードとサーバー設定に分散されてしまいます。スケジューラーをコードに埋め込んで、2つのタスクを処理でき、必要に応じて複数のノードにスケールできるものが欲しいところです。

ここでAPScheduler（Advanced Python Scheduler）ライブラリが便利です。

このライブラリでできること

APSchedulerはしばらく前から存在しており、現在バージョン4.0への移行期間中です。4番目のブランチはまだプレリリースの状態ですが、ライブラリアーキテクチャを大幅に再構築しています。

以前は主に単一プロセスのローカルスケジューラーでしたが、今は本格的な分散キューおよびスケジューリングシステムに進化しました。最もシンプルなユースケースでもわずか3行のコードで実行できます。

コアコンセプト：通常のPython関数またはコルーチンを宣言し、必要な実行条件を付けてトリガーをアタッチします。

from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler()

async def send_digest():
    print("Отправляем утренний дайджест...")

# Запуск каждый будний день в 9 утра
scheduler.add_job(send_digest, 'cron', day_of_week='mon-fri', hour=9, minute=0)
scheduler.start()

このライブラリは時間追跡、オフセット計算、適切なタイミングで関数を呼び出すことを担当します。

スケジュールオプション

4つの組み込みトリガータイプがあります：

• Cronトリガー。おなじみのLinuxスタイルの構文。曜日、月、特定の日時を柔軟に設定できます。
• 間隔実行。N秒、分、または時間ごとに実行。外部APIの定期的なポーリングに便利です。
• カレンダートリガー。間隔が月や年の長さに依存する場合に必要です。例えば、毎月1日の正午にタスクを実行する場合など。
• 1回限りの実行。指定された将来の時点で正確に1回だけ実行されます。

組み込みの条件が不十分な場合は、複合ルールを使用してトリガーを組み合わせたり、カスタムロジックで独自のクラスを作成したりできます。

一般的な障害からの保護

実際には、バックグラウンドタスクは定期的に過負荷や遅延に遭遇します。APSchedulerには、カスタムソリューションではよく見落とされたいくつかの有用なメカニズムが含まれています。

まず、并发実行数の制限です。重いレポートのエクスポートを5分ごとに設定했지만、前の実行が7分間スタックした場合、スケジューラーは2番目のインスタンスを并发で起動せず、データベースに過負荷をかけません。

次に、5パラメータです。これはタスクの開始時刻にランダムな遅延を追加します。100個のワーカーがあり、すべてが00:00に正確にキャッシュを更新する必要があるとします。ジャッターがなければ、データベースは即座に負荷の急上昇を経験します。数秒のランダムオフセットがあれば、負荷は 均等に分散されます。

3番目に、誤発火猶予時間の処理です。サーバーが負荷で停止したり、再起動したりした場合、スケジューラーはタスクがどれだけ遅れているかを確認します。遅延が許容範囲内であればタスクは実行され、範囲外であればキューの蓄積なしにスキップされます。

ストレージと分散モード

シンプルなスクリプトの場合、タスクはメモリに保持できます。しかし、サービスが再起動すると、スケジュールはリセットされます。これを防ぐため、ライブラリは永続ストレージをサポートしています：

• PostgreSQL
• MySQL
• SQLite
• MongoDB

バージョン4では、スケジューラーは分散クラスターで動作することを学びました。複数のアプリケーションインスタンスが共有データベースとイベントブローカー（Redis、PostgreSQL LISTEN/NOTIFY、MQTTがサポートされています）に接続します。

これにより水平スケーリングが実現されます：1つのノードが停止しても、他のノードがキューからタスクを取得します。

# Пример концепции работы с постоянным хранилищем
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

jobstores = {
    'default': SQLAlchemyJobStore(url='postgresql+asyncpg://user:pass@localhost/mydb')
}

scheduler = AsyncIOScheduler(jobstores=jobstores)

同期コードと非同期コード

このライブラリは最新のPythonスタックにうまく適合します。異なるスケジューラ実装が利用可能です：

• FastAPI、Litestar、Aiohttpなどの最新の非同期バックエンド向けのTrioとの統合。
• 同期スクリプト、Django、Flask向け。

9で非同期関数を同期的な回避策でラップする必要はありません—スケジューラーはメインイベントループ内でコルーチンをネイティブにawaitします。

これが役立つ場面

私は通常、3つの典型的な状況でAPSchedulerを使います：

1. 小さなマイクロサービスやボット。CeleryやRQのデプロイが過剰で、プロセス内で直接定期的なタスクが必要な場合。
2. 遅延ユーザーアクション。例えば、配達ちょうど24時間後に注文を評価するよう依頼するメールを送る場合。
3. 一時データのクリーンアップと、外部システムからの参照データの定期的な同期。

プロジェクトで使用する価値はあるか

Pythonで記述していて、予測可能な時間ベースのタスク実行が必要な場合、APSchedulerは最も成熟したオプションの1つです。

唯一のニュアンスは、現在のところバージョン3.xから4.0への移行期間であることです。3番目のブランチは長年にわたって本番環境でテストされており、最大限安定していますが、分散操作には制限があります。バージョン4.0は最新の 아키텍처とスケーリングをもたらしますが、著者は最終リリース前にAPIの破壊的変更の可能性について正直に警告しています。

現在の本番環境では、安定した10ブランチに留まる方が安全であり、バージョン4.0は今後の変更に注意しながら、ペットプロジェクトや新しいマイクロサービスで試してみる価値があります。