ホームページ >バックエンド開発 >Python チュートリアル >Python でサードパーティモジュール apscheduler をインストールして使用する方法は?

Python でサードパーティモジュール apscheduler をインストールして使用する方法は?

王林転載: 2023-05-08 10:04:171594ブラウズ

apscheduler モジュール

apscheduler モジュールのインストール

pip install apscheduler

apscheduler モジュールの概要

APScheduler( Advanced Python Scheduler) は、軽量の Python スケジュールされたタスクスケジューリングフレームワーク (Python ライブラリ) です。

APScheduler には、次の 3 つの組み込みスケジューリングシステムがあります。

cron スタイルのスケジューリング (オプションの開始/終了時間)
間隔ベースの実行 (ジョブを等間隔で実行します。開始/終了時刻も選択できます)
タスクの 1 回遅延実行 (ジョブを 1 回だけ実行します)

サポートされるバックエンドストレージジョブ

APScheduler では、スケジューリングシステムのバックエンドとジョブストレージを任意に組み合わせることができます。サポートされるバックエンドストレージジョブは次のとおりです。

メモリ
##SQLAlchemy
MongoDB
Redis
RethinkDB
ZooKeeper

APScheduler には 4 つのコンポーネントがあります

トリガー (トリガー) にはスケジュールロジックが含まれており、各ジョブには次回の実行時間を決定する独自のトリガーがあります。トリガー自体の初期設定を除けば、トリガーは完全にステートレスです。
ジョブストア (ジョブストレージ) は、スケジュールされたジョブを保存します。デフォルトのジョブストアは単にジョブをメモリに保存し、他のジョブストアはジョブをデータベースに保存します。ジョブが永続ジョブストアに保存されると、ジョブのデータはロード時にシリアル化および逆シリアル化されます。ジョブストレージはスケジューラを共有できません。
エグゼキュータはジョブの実行を処理します。通常、これは、指定された呼び出し可能オブジェクトをジョブ内のスレッドまたはプロセスプールに送信することによって行われます。ジョブが完了すると、実行者はスケジューラに通知します。
schedulers (スケジューラー) ジョブの追加、変更、削除など、ジョブストレージとエグゼキューターの構成はスケジューラーで実行できます。さまざまなアプリケーションシナリオに応じてさまざまなスケジューラを選択できます。BlockingScheduler、BackgroundScheduler、AsyncIOScheduler、GeventScheduler、TornadoScheduler、TwistedScheduler、および QtScheduler の 7 つのオプションが利用可能です。

各コンポーネントの紹介

トリガー

ジョブをスケジュールするときは、このジョブのトリガーを選択して、ジョブの実行時期を説明する必要があります。がトリガーされる場合、APScheduler には 3 つの組み込みトリガータイプがあります:

date: 1 回限りの指定された日付;
interval: 一定の間隔time 時間範囲内での実行頻度;
cron: Linux の crontab 形式は互換性があり、最も強力です。

date 最も基本的な種類のスケジューリングで、ジョブは 1 回だけ実行されます。そのパラメータは次のとおりです:

1.run_date

(datetime|str) – ジョブの実行日付または時刻

2.timezone

(datetime.tzinfo| str) &ndash ; タイムゾーンの指定

ジョブストレージ

アプリケーションが起動するたびにジョブを再作成する場合は、デフォルトのジョブストア (MemoryJobStore) を使用します。 )、ただし、スケジューラの再起動やアプリケーションのクラッシュにもかかわらずジョブを保持する必要がある場合は、アプリケーション環境に基づいて特定のジョブストアを選択する必要があります。例: Mongo または SQLAlchemy JobStore (ほとんどの RDBMS をサポートするために使用) を使用します。

executor

Executor の選択は、上記のフレームワークのどれを使用するかによって異なります。この場合、デフォルトの ThreadPoolExecutor を使用することで、すでにニーズを満たすことができます。アプリケーションに

CPU 集中型の操作が含まれる場合は、ProcessPoolExecutor を使用してより多くの CPU コアを使用することを検討してください。 ProcessPoolExecutor を 2 番目のエグゼキュータとして使用して、両方を同時に使用することもできます。

適切なスケジューラの選択

BlockingScheduler: アプリケーションで実行されているのがスケジューラだけである場合
BackgroundScheduler : 他のフレームワークを実行しておらず、アプリのバックグラウンドでスケジューラーを実行したい場合に使用します。
AsyncIOScheduler: プログラムで asyncio (非同期フレームワーク) を使用する場合に使用されます。
GeventScheduler: プログラムで gevent (高性能 Python 同時実行フレームワーク) を使用する場合に使用されます。
TornadoScheduler: プログラムが Tornado (Web フレームワーク) に基づいている場合に使用されます。
TwistedScheduler: プログラムで Twisted (非同期フレームワーク) を使用する場合に使用します。
QtScheduler: アプリケーションが Qt の場合に使用できます。。

apscheduler モジュールは

Add job

新しいジョブを追加するには 2 つの方法があります:

add_job でジョブを追加します。 ;

# -*- coding:utf-8 -*-
from apscheduler.schedulers.blocking import BlockingScheduler
import datetime

def my_job1():
    print(&#39;my_job1 is running, Now is %s&#39; % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

def my_job2():
    print(&#39;my_job2 is running, Now is %s&#39; % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

sched = BlockingScheduler()
# 每隔5秒运行一次my_job1
sched.add_job(my_job1, &#39;interval&#39;, seconds=5, id=&#39;my_job1&#39;)
# 每隔5秒运行一次my_job2
sched.add_job(my_job2, &#39;cron&#39;, second=&#39;*/5&#39;, id=&#39;my_job2&#39;)
sched.start()

デコレータモードではジョブが追加されます。

# -*- coding:utf-8 -*-
from apscheduler.schedulers.blocking import BlockingScheduler
import datetime

sched = BlockingScheduler()

# 每隔5秒运行一次my_job1
@sched.scheduled_job(&#39;interval&#39;, seconds=5, id=&#39;my_job1&#39;)
def my_job1():
    print(&#39;my_job1 is running, Now is %s&#39; % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

# 每隔5秒运行一次my_job2
@sched.scheduled_job(&#39;cron&#39;, second=&#39;*/5&#39;, id=&#39;my_job2&#39;)
def my_job2():
    print(&#39;my_job2 is running, Now is %s&#39; % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

sched.start()