Flask中怎么用celery做定时任务

朱雀 2023-06-18 13:52 75阅读 0赞

#### Celery ####

Celery 是一个异步任务队列，一个Celery有三个核心组件：

*  Celery 客户端
 *  Celery workers: 运行后台作业的进程
 *  消息代理，最常用的代理就是 RabbitMQ 和 Redis

#### Flask中怎么使用celery做定时任务 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzI3ODA4OQ_size_16_color_FFFFFF_t_70]

config.py

# coding: utf-8
    import os
    from datetime import timedelta
    from celery.schedules import crontab
    
    
    ROOT_FOLDER = os.path.dirname(os.path.abspath(__file__))
    
    
    # flask 和 celery 共用配置文件
    class Config:
    
        INFO_LOG = "/var/log/analyse_info.log"
        ERROR_LOG = "/var/log/analyse_error.log"
    
        HOST_IP = "10.132.1.211"
        # STATIC_FOLDER = os.path.join(ROOT_FOLDER, '../static')
        STATIC_FOLDER = './static'
    
        DB_CONNECTION = 'mysql://root:admin123@mysql.mysqlnet:3306/tempest'
        SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://root:admin123@mysql.mysqlnet:3306/tempest?charset=utf8'
        SQLALCHEMY_POOL_RECYCLE = 7200
        MAIL_SERVER = '200.200.0.11'
        MAIL_PORT = 25
        MAIL_USE_TLS = False
    
        MAX_CONTENT_LENGTH = 9999999
    
        TEMPEST_DATA_DIR = '/report/analyse/data'
    
        # 初始化celery
        CELERY_BROKER_URL = 'redis://localhost:6379/0',
        CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
    
        CELERY_IMPORTS = [
            'analyse.celery_task.schedules',
        ]
    
        CELERYBEAT_SCHEDULE = { 
            'schedule_to_save_bugs': { 
                'task': 'analyse.celery_task.schedules.schedule_to_save_bugs',
                'schedule': timedelta(minutes=2),
                'args': () # 任务函数的参数 
            },
        }
    
        @staticmethod
        def init_app(app):
            pass

app.py

# -*- coding: utf-8 -*-
    import sys
    import logging
    from flask import Flask, jsonify, url_for, request
    from analyse.config import config
    from analyse.extensions import db
    from analyse.extensions import migrate
    from analyse.extensions import api
    from analyse.extensions import mail
    from analyse.views.index import index_blueprint
    from analyse.views import api_blueprint
    from analyse import utlis
    
    reload(sys)
    sys.setdefaultencoding('utf-8')
    
    def create_app(config_name='default'):
        
        app = Flask(__name__)
        config_obj = config[config_name]()
        app.config.from_object(config_obj)
        config_obj.init_app(app)
        app.static_folder = app.config.get('STATIC_FOLDER', './static')
        register_logger(app)
        db.init_app(app)
        migrate.init_app(app, db)
        api.init_app(api_blueprint)
        app.register_blueprint(api_blueprint)
        app.register_blueprint(index_blueprint)
        mail.init_app(app)
        configure_errorhandlers(app)
    
        return app

celery\_task

*  manager.py
    
        #!/usr/bin/env python
        from celery import Celery
        from analyse.app import create_app
        
        
        def make_celery(app):
            celery = Celery(
                app.import_name,
                backend=app.config['CELERY_RESULT_BACKEND'],
                broker=app.config['CELERY_BROKER_URL']
            )
            celery.conf.update(app.config)
        	
            # 将定时任务执行包装在app应用程序上下文中，这样就可以调用db，models
            class ContextTask(celery.Task):
                def __call__(self, *args, **kwargs):
                    with app.app_context():
                        return self.run(*args, **kwargs)
        
            celery.Task = ContextTask
            return celery
        
        
        flask_app = create_app()
        celery = make_celery(flask_app)
 *  task.py
    
        # 使用你的定时任务
        
        from analyse.celery_task.manager import celery
        
        @celery.task()
        def add_together(a, b):
            return a + b

#### 启动服务 ####

1.  先启动celery beat
    
        celery -A analyse.celery_task.manager.celery beat
2.  在启动celery worker
    
        celery -A analyse.celery_task.manager.celery worker -l info

ry -A analyse.celery\_task.manager.celery beat

2. 在启动celery worker
    
    ```python
    celery -A analyse.celery_task.manager.celery worker -l info

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzI3ODA4OQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191202193451249.PNG?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzI3ODA4OQ==,size_16,color_FFFFFF,t_70