最新Python APScheduler 定时任务详解

from datetime import datetime
from datetime import date
from apscheduler.schedulers.blocking import BlockingScheduler
def job(text):
    print(text)
scheduler = BlockingScheduler()
# 在 2019-8-30 运行一次 job 方法
scheduler.add_job(job, 'date', run_date=date(2022, 4, 9), args=['text1'], id="1", coalesce=True, max_instances=1)
# 在 2019-8-30 01:00:00 运行一次 job 方法
scheduler.add_job(job, 'date', run_date=datetime(2022, 4, 9, 17, 40, 58), args=['text2'], id="2", coalesce=True, max_instances=1)
# 在 2019-8-30 01:00:01 运行一次 job 方法
scheduler.add_job(job, 'date', run_date='2022-4-9 17:41:00', args=['text3'], id="3", coalesce=True, max_instances=1)
scheduler.start()

@sched.scheduled_job(
    "interval", id=spider_job_name + "_bg_data", coalesce=True, max_instances=1, minutes=20
)
def tick_rzjg_detail_xq():
    """
    快速完成
    :return:
    """
    each = "rzjg_bg_data"
    cmd_str = f"cd {ROOT} && bash run_spider.sh {each} --loglevel=INFO"
    print(cmd_str)
    os.system(cmd_str)
def func():
    print("Press Ctrl+C to exit")
    # 直接触发一次
    tick_rzjg_detail_xq()
    try:
        sched.start()
    except (KeyboardInterrupt, SystemExit):
        pass

if __name__ == "__main__":
    func()

 import time
    from apscheduler.schedulers.blocking import BlockingScheduler

    def job(text):
        t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
        print('{} --- {}'.format(text, t))

    scheduler = BlockingScheduler()
    # 在每天22点，每隔 1分钟 运行一次 job 方法
    scheduler.add_job(job, 'cron', hour=22, minute='*/1', args=['job1'])
    # 在每天22和23点的25分，运行一次 job 方法
    scheduler.add_job(job, 'cron', hour='22-23', minute='25', args=['job2'])
    # 在每天 8 点，运行一次 job 方法
    scheduler.add_job(job, 'cron', hour='8', args=['job2'])
    # 在每天 8 点 20点，各运行一次 job 方法    设置最大运行实例数
    scheduler.add_job(job, 'cron', hour='8, 20', minute=30, max_instances=4)
    scheduler.start()