flask使用celery

Posted 2021-04-12 小白码上说

1. 安装celery-sqlalchemy-scheduler==0.2.7 这个是一个跟django-celery-beat提取出来的插件包，能通过数据库管理celery的定时任务

2. 安装

   pip install celery-sqlalchemy-scheduler

3. 配置， 在统一配置的地方加上

   # 用数据库的配置# beat_dburi 是celery-sqlalchemy-scheduler模块读取的配置项'beat_dburi': 'mysql+pymysql://root:123456@127.0.0.1:3306/zero?charset=UTF8MB4',# 也可以这样更新配置from celery import Celery
   celery = Celery('tasks')
   beat_dburi = 'sqlite:///schedule.db'
   celery.conf.update( {'beat_dburi': beat_dburi})# 如果不用数据库，可以直接写在配置文件中'CELERYBEAT_SCHEDULE': { 'blog_views': { # 任务名 'task': 'blog.tasks.blog.test', # 任务路径 'schedule': crontab(), # crontab表达式，空为每分钟执行一次 'args': () # 参数 }, 'comment_views': { 'task': 'blog.tasks.comment.test1', 'schedule': crontab(), 'args': () }}

4. 启动命令

   # worker windows 启动，只能单线程elery -A [project_name] worker --pool=solo -l INFO # worker liunx启动 可以多线程celery worker -A [project_name] -l info -E
   # beat 启动 数据库模式启动celery -A [project_name] -S celery_sqlalchemy_scheduler.schedulers:DatabaseScheduler -l INFO
   # beat 启动，配置文件模式celery -A [project_name] -l INFO

‍




5. 如果数据模块启动时默如果数据没有对应的表会默认生成这5张表

   
   

• celery_crontab_schedule (crontab表达式存储表)

  
  

• celery_crontab_schedule (时间表达式存储表)

        如下图，为每5秒执行一次

• celery_periodic_task(任务信息表)

如下图，id 为2的任务关联了crontab_id为2的执行方式，哪当前任务会已每分钟一次去执行

• celery_periodic_task_changed(如果任务执行方式或者执行时间修改了，则需要修改这张表的last_update字段为当前最新的时间，下次beat监听的话就能拉取最新的任务配置)