flask-web APScheduler 定时任务以及实际应用

申屠恺

2023-12-01

APScheduler使用

APScheduler （advanceded python scheduler）是一款Python开发的定时任务工具。

文档地址 https://apscheduler.readthedocs.io/en/latest/userguide.html#starting-the-scheduler

特点：

不依赖于Linux系统的crontab系统定时，独立运行
可以动态添加新的定时任务，如

下单后30分钟内必须支付，否则取消订单，就可以借助此工具（每下一单就要添加此订单的定时任务）
对添加的定时任务可以做持久保存

1 安装

pip install apscheduler

2 使用方式

from apscheduler.schedulers.background import BackgroundScheduler

# 创建定时任务的调度器对象
scheduler = BackgroundScheduler()

# 定义定时任务
def my_job(param1, param2):
    pass

# 向调度器中添加定时任务
scheduler.add_job(my_job, 'date', args=[100, 'python'])

# 启动定时任务调度器工作
scheduler.start()

3 调度器 Scheduler

负责管理定时任务

BlockingScheduler: 作为独立进程时使用
非动态定时任务，用BlockingScheduler ，程序会阻塞在这，防止退出

  from apscheduler.schedulers.blocking import BlockingScheduler

  scheduler = BlockingScheduler()
  scheduler.start()  # 此处程序会发生阻塞

BackgroundScheduler: 在框架程序（如Django、Flask）中使用
动态添加定时任务，用BackgroundScheduler，程序会立即返回，后台运行

  from apscheduler.schedulers.background import BackgroundScheduler

  scheduler = BackgroundScheduler()
  scheduler.start()  # 此处程序不会发生阻塞

4 执行器 executors

在定时任务该执行时，以进程或线程方式执行任务

ThreadPoolExecutor

  from apscheduler.executors.pool import ThreadPoolExecutor
  ThreadPoolExecutor(max_workers)  
  ThreadPoolExecutor(20) # 最多20个线程同时执行

使用方法

  executors = {
      'default': ThreadPoolExecutor(20)
  }
  scheduler = BackgroundScheduler(executors=executors)

ProcessPoolExecutor

  from apscheduler.executors.pool import ProcessPoolExecutor
  ProcessPoolExecutor(max_workers)
  ProcessPoolExecutor(5) # 最多5个进程同时执行

使用方法

  executors = {
      'default': ProcessPoolExecutor(3)
  }
  scheduler = BackgroundScheduler(executors=executors)

5 触发器 Trigger

指定定时任务执行的时机

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

from datetime import date

# 在2019年11月6日00:00:00执行
sched.add_job(my_job, 'date', run_date=date(2009, 11, 6))

# 在2019年11月6日16:30:05
sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 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 经过指定的时间间隔执行

weeks (int) – number of weeks to wait
days (int) – number of days to wait
hours (int) – number of hours to wait
minutes (int) – number of minutes to wait
seconds (int) – number of seconds to wait
start_date (datetime|str) – starting point for the interval calculation
end_date (datetime|str) – latest possible date/time to trigger on
timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations

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')

3） cron 按指定的周期执行

year (int|str) – 4-digit year
month (int|str) – month (1-12)
day (int|str) – day of the (1-31)
week (int|str) – ISO week (1-53)
day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)
hour (int|str) – hour (0-23)
minute (int|str) – minute (0-59)
second (int|str) – second (0-59)
start_date (datetime|str) – earliest possible date/time to trigger on (inclusive)
end_date (datetime|str) – latest possible date/time to trigger on (inclusive)
timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations (defaults to scheduler timezone)

# 在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')

6 配置方法

方法1

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.executors.pool import ThreadPoolExecutor

executors = {
    'default': ThreadPoolExecutor(20),
}
scheduler = BackgroundScheduler(executors=executors)

方法2

from pytz import utc

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ProcessPoolExecutor

executors = {
    'default': {'type': 'threadpool', 'max_workers': 20},
    'processpool': ProcessPoolExecutor(max_workers=5)
}

scheduler = BackgroundScheduler()
# .. 此处可以编写其他代码

# 使用configure方法进行配置
scheduler.configure(executors=executors)

7 启动

scheduler.start()

对于BlockingScheduler ，程序会阻塞在这，防止退出
对于BackgroundScheduler，程序会立即返回，后台运行

8 扩展

任务管理
方式1

job = scheduler.add_job(myfunc, 'interval', minutes=2)  # 添加任务
job.remove()  # 删除任务
job.pause() # 暂定任务
job.resume()  # 恢复任务

方式2

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')  # 恢复任务

调整任务调度周期

job.modify(max_instances=6, name='Alternate name')

scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')

停止APScheduler运行

scheduler.shutdown()

定时任务

crontab

Linux 本身自带的一个命令，由Linux操作系统维护定时任务
APScheduler

Python实现的，定时不是由Linux操作系统维护，是单独开启进程的方式，在进程中管理定时

定时任务有两种：

定时任务进行页面静态化在django运行起来之前，我们明确知道要有这个定任务

不是动态添加的
- 一般可以直接采用Linux crontab
- APScheduler
在django 程序已经运行的情况下

用户1下单判断在30分钟内必须支付，否则取消订单恢复库存

在用户下单的时刻起，创建一个30分钟的定时任务，任务的功能是到30分钟的时刻，判断订单状态，如果未支付，取消订单

(动态添加定时任务)

-> 支付
- 如果是用crontab 不是很方便，通常使用APScheduler

动态添加定时任务，用BackgroundScheduler，  程序会立即返回，后台运行
非动态定时任务，用BlockingScheduler ，程序会阻塞在这，防止退出

笔记：
任务存储后端是针对于动态添加定时任务，非动态定时任务没有必要用

定时修正统计数据

在toutiao-backend/toutiao/__init__.py中添加APScheduler调度器对象（两种方法：1）

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.executors.pool import ThreadPoolExecutor

def create_app(config, enable_config_file=False):
    ...

    # 添加定时任务APScheduler
    executors = {
        'default': ThreadPoolExecutor(10)
    }
    app.scheduler = BackgroundScheduler(executors=executors)

    from .schedule.statistic import fix_statistics

    # 每天3点执行
    app.scheduler.add_job(fix_statistics, 'cron', hour=3, args=[app])
    # 立即执行，用于测试
    # app.scheduler.add_job(fix_statistics, 'date', args=[app])

    app.scheduler.start()

    ...

在toutiao-backend/toutiao中新建schedule目录用于存放定时任务
toutiao-backend/toutiao/schedule/main.py 中添加APScheduler调度器对象（两种方法：2）
作为一个独立启动文件

import sys
import os

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.join(BASE_DIR, 'common'))
sys.path.insert(0, os.path.join(BASE_DIR))

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.executors.pool import ProcessPoolExecutor

# 创建一个apscheduler调度器对象

# 配置调度器对象使用的 任务存储后端  执行器（进程、线程）

executors = {
    # 表示默认到了时间，该执行的定时任务都是放到进程池中的一个子进程执行
    # 3表示进程池中最多有3个进程，也就说 在同一时刻，最多 有3个进程同时执行
    'default': ProcessPoolExecutor(3)
}

# Blocking 阻塞的调度
scheduler = BlockingScheduler(executors=executors)

import common

# 添加定时任务

# 在每天的凌晨3点执行
from statistic import fix_statistics
scheduler.add_job(fix_statistics, 'cron', hour=3)

# 测试使用这个触发时间，在scheduler启动的使用执行一次定时任务
# scheduler.add_job(fix_statistics, 'date')


# 启动scheduler
if __name__ == '__main__':
    # start()会阻塞当前文件退出
    scheduler.start()

toutiao-backend/toutiao/schedule/statistics.py

from sqlalchemy import func
from flask import current_app

from models.news import Article
from models import db
from cache import statistic as cache_statistic
from common import flask_app


def __fix_statistics(cls):
    # 进行数据库查询
    with flask_app.app_context():
        ret = cls.db_query()

        # 将数据库查询结果设置到redis中
        cls.reset(ret)


def fix_statistics():
    """
    修正redis中存储的统计数据 定时任务
    :return:
    """
    # 查询数据库得到统计数据
    # class UserArticlesCountStorage(CountStorageBase):
    #     """
    #     用户文章数量
    #     """
    #     key = 'count:user:arts'   zset  4,1

    # sql   分组聚合查询
    # select user_id, count(article_id)  from news_article_basic where status=2 group by user_id
    # ret = db.session.query(Article.user_id, func.count(Article.id))\
    #         .filter(Article.status == Article.STATUS.APPROVED)\
    #         .group_by(Article.user_id).all()

    # ret -> [
    # ( 1, 46141),
    # (2, 46357 ),
    # (3 ,46187)
    # ]

    # # 设置redis的存储记录
    # pl = current_app.redis_master.pipeline()
    # pl.delete('count:user:arts')
    #
    # # zadd(key, score1, val1, score2, val2, ...)
    # for user_id, count in ret:
    #     pl.zadd('count:user:arts', count, user_id)
    #
    # pl.execute()

    __fix_statistics(cache_statistic.UserArticlesCountStorage)

    __fix_statistics(cache_statistic.UserFollowingsCountStorage)

    __fix_statistics(cache_statistic.ArticleCollectingCountStorage)

    __fix_statistics(cache_statistic.UserArticleCollectingCountStorage)

common/cache/statistic.py

class CountStorageBase(object):
    """
    统计数量存储的父类
    """
    ...

    @classmethod
    def reset(cls, db_query_ret):
        """
        由定时任务调用的重置数据方法
        """
        # 设置redis的存储记录
        pl = current_app.redis_master.pipeline()
        pl.delete(cls.key)
        
        # ret -> [
        # ( 1, 46141),
        # (2, 46357 ),
        # (3 ,46187)
        # ]

        # zadd(key, score1, val1, score2, val2, ...)
        # 方式一
        # for data_id, count in db_query_ret:
        #     pl.zadd(cls.key, count, data_id)

        # 方式二
        redis_data = []
        for data_id, count in db_query_ret:
            redis_data.append(count)
            redis_data.append(data_id)

        # redis_data = [count1, data_id1, count2, data_id2, ..]
        pl.zadd(cls.key, *redis_data)
        # pl.zadd(cls.key, count1, data_id1, count2, data_id2, ..]
        # *redis_data 解析列表
        ## **redis_data  解析字典
        pl.execute()

class UserArticlesCountStorage(CountStorageBase):
    """
    用户文章数量
    """
    key = 'count:user:arts'

    @staticmethod
    def db_query():
        ret = db.session.query(Article.user_id, func.count(Article.id)) \
            .filter(Article.status == Article.STATUS.APPROVED).group_by(Article.user_id).all()
        return ret


class UserFollowingsCountStorage(CountStorageBase):
    """
    用户关注数量
    """
    key = 'count:user:followings'

    @staticmethod
    def db_query():
        ret = db.session.query(Relation.user_id, func.count(Relation.target_user_id)) \
            .filter(Relation.relation == Relation.RELATION.FOLLOW)\
            .group_by(Relation.user_id).all()
        return ret

pl.zadd(cls.key, *redis_data) 其中 *redis_data 是解析列表，**redis_data 是解析字典