Python scheduleモジュールを使用しての定周期処理

Question

実現したいこと

任意の間隔での関数の定周期実行
前回処理開始時 - (インターバル) - 今回処理開始

発生している問題・分からないこと

指定したインターバル間隔が処理終了時～次回処理開始となってしまう。
前回処理開始時 - 前回処理終了時 - (インターバル) - 今回処理開始

該当のソースコード

import time
import schedule
import datetime as dt

INTERVAL = 10

def main():
    scheduler = schedule.Scheduler()
    scheduler.every(INTERVAL).seconds.do(
        heavy_function
    )

    while True:
        scheduler.run_pending()
        time.sleep(0.1)
    

def heavy_function():
    start_time = dt.datetime.now(tz=dt.timezone.utc)
    print(f'start: {start_time}')
    time.sleep(5)
    end_time = dt.datetime.now(tz=dt.timezone.utc)
    print(f'end: {end_time}')
    

if __name__ == '__main__':
    main()

start: 2024-12-05 03:22:14.116578+00:00
end: 2024-12-05 03:22:19.121740+00:00
start: 2024-12-05 03:22:29.145224+00:00
end: 2024-12-05 03:22:34.150440+00:00
start: 2024-12-05 03:22:44.174894+00:00
end: 2024-12-05 03:22:49.180033+00:00

試したこと・調べたこと

• teratailやGoogle等で検索した
• ソースコードを自分なりに変更した
• 知人に聞いた
• その他

上記の詳細・結果

実現したいことを実施する方法が見つかりませんでした。

補足

特になし

Comments

[poto568]

私の環境では「python schedule」でgoogle検索すると以下の記事が
6番目に表示されています。(私が書いた記事ではありません。)
https://progzakki.sanachan.com/program-lang/python/schedule-job-periodically/
続きはこちら。
https://progzakki.sanachan.com/program-lang/python/signal-handler-periodically/
python単体で完結できませんが、cronとかタスクスケジューラなんかと
併用しても良いかもしれません。

[bsdfan]

「scheduleモジュールを使用して」というのが必須なら難しいと思います。jobを別スレッドにする方法もありますが、起動時間のズレが積み重なっていくのは消せません。(それぐらいのズレが許容できるならscheduleでスレッドなりを使えばいいと思います)
他のライブラリでもいいなら、apscheduler でご希望の動作ができると思います。

Answer 1

指定したインターバル間隔が処理終了時～次回処理開始となってしまう。

スケジューラーでは呼び出しのみに留めて、実際の処理は別スレッドで行うようにします。
schedule ライブラリでは、（補足追記: スケジューラーと同一スレッドで実行される）タスク内で time.sleep のようなスケジューラーを阻害するようなことはしてはいけません。

schedule は元々そのような設計で、
ドキュメントの一番上に注意書きがあります。DeepL翻訳

Scheduleを使用しない場合:

正直に言うと、Scheduleは'万能'なスケジューリングライブラリではありません。
このライブラリは単純なスケジューリング問題のための単純なソリューションとして設計されています。
もし必要であれば他をあたるべきでしょう：

• ジョブの永続性（再スタート時にスケジュールを覚えておく）
• 正確なタイミング（秒以下の精度での実行）
• 同時実行（複数のスレッド）
• ローカライゼーション（平日や休日）

スケジュールはジョブ機能の実行時間を考慮しません。
安定した実行スケジュールを保証するには、
長時間実行するジョブをメインスレッド（スケジューラが実行されるスレッド）から移動させる必要があります。> 実装のサンプルは 並列実行 を参照してください。

※ 注意すべきは、リストとして挙がっている４項目に該当する場合が、「使用しない場合」
つまり、schedule ライブラリ側では対応していない項目です。
追記: 翻訳で意図が伝わりにくくなってるかもしれないので、原文も参考にしてください。

解決策(簡易版)

def as_background(func):
    from threading import Thread
    from functools import wraps

    @wraps(func)
    def _func():
        thread = Thread(target=func, daemon=True)
        thread.start()
    
    return _func


@as_background
def heavy_function():
    ...

この実装では毎回スレッドを生成しますが、秒単位の制度が必要な場合は問題になるかもしれません。
予め複数のスレッドを稼働させておき、同期queue等でメッセージを送ることにより、
遅延をさらに少なくすることができます。⇒ 上記リンクにqueueを用いた実装例もあり

別ライブラリの APScheduler なら、デフォルトでそのような実装になってます。
（メインスレッドでスケジューラー稼働、別スレッドでジョブ実行）

他に、精度を確保するアイデアとしては、折角のコードの可読性が損なわれてしまいますが、
定期的にタスク登録用のタスクを用いることで、遅延をリセットします
例えば、毎時 0分に なったら、10秒後20秒後...にジョブが実行されるように複数登録。

解決策2: スレッドプールを用いた実装 (apscheduler と同様の挙動・精度)

import time
from datetime import datetime
from threading import Thread
from queue import Queue
from concurrent.futures import ThreadPoolExecutor
import logging
import schedule

logger = logging.getLogger(__name__)


class BackgroundWorker:
    def __init__(self, *args, **kw):

        # NOTE: daemon-thread, will not wait task completed
        queue = self.queue = Queue()
        thread = self.thread = Thread(
                        target=self.run,
                        args=(queue,),
                        name="worker",
                        daemon=True)
        thread.start()

    def __call__(self, func):
        def _job(*args, **kw):
            self.queue.put((func, args, kw))
        return _job

    def __del__(self):
        if self.thread:
            self.queue.put(None)
            self.thread.join()

    def run(self, queue):
        logger.info("start")
        with ThreadPoolExecutor(
                max_workers=5,
                thread_name_prefix="worker") as executor:
            for func, args, kw in iter(queue.get, None):
                executor.submit(func, *args, **kw)
        logger.info("end")


def main():
    worker = BackgroundWorker()

    @worker
    def job():
        logger.debug(f"I'm working... {datetime.now()}")
        time.sleep(5)
        logger.debug(f"end            {datetime.now()}")

    schedule.every(10).seconds.do(job)

    try:
        while True:
            schedule.run_pending()
            time.sleep(1)
    except (KeyboardInterrupt, SystemExit) as e:
        pass


if __name__ == '__main__':
    logging.basicConfig(
        level=logging.DEBUG,
        format="[%(threadName)-10s][%(levelname)-8s] %(message)s",

    )
    main()

Comments

[keruuuu]

丁寧なご回答ありがとうございます。
APSchedulerを使用することで解決できました。