【python】使用APScheduler或Django-apscheduler实现定时任务

野性酷女 2022-12-20 08:54 789阅读 0赞

#### 1.APScheduler ####

之前在Django使用crontab来完成定时任务，还有一个更灵活的python工具APScheduler也可以完成

**特点**  
1）可以动态添加任务  
2）不依赖Linux的crontab系统定时  
3）可以对添加的定时任务做持久保存

#### 2.简单使用 ####

pip install apscheduler

from apscheduler.schedulers.background import BackgroundScheduler
    
    scheduler = BackgroundScheduler()  # 创建调度器对象
    
    def my_job(param1, param2):  # 任务
        print(param1, param2)
    
    scheduler.add_job(my_job, 'date', args=[100, 'python'])  # 添加任务
    
    scheduler.start()  # 启动调度器

#### 3.调度器、执行器、触发器 ####

##### 3.1 Scheduler 调度器 #####

调度器是管理定时任务的

当使用`BlockingScheduler`时作为独立进程使用，会阻塞主线程  
当使用`BackgroundScheduler`时会在后台运行，不会发生阻塞

from apscheduler.schedulers.blocking import BlockingScheduler
    from apscheduler.schedulers.background import BackgroundScheduler
    
    scheduler = BackgroundScheduler()
    scheduler.start()  # 此处程序不会发生阻塞
    
    scheduler = BlockingScheduler()
    scheduler.start()  # 此处程序会发生阻塞

##### 3.2 executors 执行器 #####

执行器控制执行方式

在定时任务该执行时，`ThreadPoolExecutor`以线程方式执行任务，`ProcessPoolExecutor`以进程方式执行任务

# 方式1： 线程
    from apscheduler.executors.pool import ThreadPoolExecutor
    executors = { 
        'default': ThreadPoolExecutor(20)   # 最多20个线程同时执行
    }
    scheduler = BackgroundScheduler(executors=executors)
    
    
    # 方式2： 进程
    from apscheduler.executors.pool import ProcessPoolExecutor
    executors = { 
        'default': ProcessPoolExecutor(3)  # 最多3个进程同时运行
    }
    scheduler = BackgroundScheduler(executors=executors)

##### 3.3 Trigger 触发器 #####

触发器控制的是什么时候会执行任务

###### 1）date 在特定的时间日期执行 ######

from datetime import date
    
    # 在2020年11月11日00:00:00执行
    sched.add_job(my_job, 'date', run_date=date(2020, 11, 11))
    
    # 在2020年11月1日16:30:05
    sched.add_job(my_job, 'date', run_date=datetime(2020, 11, 11, 16, 30, 5))
    sched.add_job(my_job, 'date', run_date='2009-11-06 16:30:05')
    
    # 立即执行
    sched.add_job(my_job, 'date')  
    sched.start()

###### 2）interval 经过指定的时间间隔执行 ######

from datetime import datetime
    
    # 每两小时执行一次
    sched.add_job(job_function, 'interval', hours=2)
    
    # 在2010年10月10日09:30:00 到2014年6月15日的时间内，每两小时执行一次
    sched.add_job(job_function, 'interval', hours=2, start_date='2010-10-10 09:30:00', end_date='2014-06-15 11:00:00')

时间间隔可选`seconds 、minutes、hours、days、weeks、start_date、end_date 、timezone`

###### 3）cron 按指定的周期执行 ######

# 在6、7、8、11、12月的第三个周五的00:00, 01:00, 02:00和03:00 执行
    sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
    
    # 在2014年5月30日前的周一到周五的5:30执行
    sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30')

可选周期`second、minute、hour、day、day_of_week、month、year、start_date、end_date、timezone`

#### 4.任务管理 ####

##### 4.1 添加、移除、暂停、恢复、停止 #####

# 方式1： 通过对象
    job = scheduler.add_job(myfunc, 'interval', minutes=2)  # 添加任务
    job.remove()  # 移除任务
    job.pause() # 暂停任务
    job.resume()  # 恢复任务
    
    # 方式2： 通过任务id
    scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')  # 添加任务 
    scheduler.remove_job('my_job_id')  # 移除任务
    scheduler.pause_job('my_job_id')  # 暂停任务
    scheduler.resume_job('my_job_id')  # 恢复任务
    
    scheduler.shutdown()  # 停止任务

##### 4.2 修改任务 #####

# 方式1： 通过对象
    job.modify(max_instances=6, name='Alternate name')
    
    # 方式2： 通过任务id
    scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')

#### 5.在Django使用APScheduler ####

pip install django-apscheduler

INSTALLED_APPS = [
        ......
        'django_apscheduler',# 定时执行任务
    ]

一定要先迁移

python manage.py migrate

迁移完了之后可以在任意一个APP的`views.py`写定时代码

使用装饰器写法

from apscheduler.schedulers.background import BackgroundScheduler
    from django_apscheduler.jobstores import DjangoJobStore, register_events, register_job
    
    # 1.实例化调度器
    scheduler = BackgroundScheduler()
    # 2.调度器使用DjangoJobStore()
    scheduler.add_jobstore(DjangoJobStore(), "default")
    try:
        # 3.设置定时任务
        # 另一种方式为每天固定时间执行任务，对应代码为
        @register_job(scheduler,"cron", hour=4)
        def my_job():
            # 这里写你要执行的任务
            pass
        # 4.注册定时任务(0.4.0版本之后不需要注册)
        # register_events(scheduler)
        # 5.开启定时任务
        scheduler.start()
    except Exception as e:
        print(e)
        # 有错误就停止定时器
        scheduler.shutdown()

普通写法

def my_job():
        print("do django-apscheduler job")
    
    scheduler = BackgroundScheduler()
    scheduler.add_jobstore(DjangoJobStore(), "default")
     # 此处触发器是每分钟执行一次，如果不知道ID它会随机生成一个32位ID
    scheduler.add_job(my_job, trigger=CronTrigger(minute="*/1"), id="my_job") 
    try:
        scheduler.start()
    except KeyboardInterrupt:
        scheduler.shutdown()

#### 6.报错 ####

`apscheduler.jobstores.base.ConflictingIdError: 'Job identifier (my_job) conflicts with an existing job'`  
如果你在启动任务的时候指定了任务ID，可能会出现这种ID冲突的情况，只需要进入数据库找到django\_apscheduler\_djangojob表，修改ID或者删掉记录即可

关于在Django使用crontab实现定时任务可以参考  
[【Django】页面静态化和crontab定时任务][Django_crontab]

[Django_crontab]: https://blog.csdn.net/qq_39147299/article/details/108533977