项目结构.+--core#逻辑运算|+--__init__.py+--www|+--模板#模板|+--意见#意见||+--任务.py||+--__init__.py|+--model.py#数据库模型|+--__init__.py+--application.py#程序入口+--config.py#配置文件+--celery_worker.py#celeryworker1.依赖安装pip3installceleryeventletredisflaskflask_restxflask_cors2.init#/www/__init__.pyfromflaskimportFlask,render_templatefromflask_corsimportCORSfromceleryimportCeleryfromconfigimportConfigcelery=Celery(__name__,broker="redis://username:password@localhost:6379/)definit_celery(app,celery):"""将烧瓶应用上下文添加到celery.Task"""TaskBase=celery.TaskclassContextTask(TaskBase):abstract=Truedef__call__(self,*args,**kwargs):withapp.app_context():returnTaskBase.__call__(self,*args,**kwargs)celery.Task=ContextTaskcelery.conf.update(app.config)defcreate_app():app=Flask(__name__)app.config.from_object(Config)from.viewsimportapiapi.init_app(app)celery.conf。更新（app.config）CORS（app，supports_credentials=True）返回app3。编写任务#/www/views/task.py@celery.task()deflong_time_task(config):returnfunc(config)4.定义视图#/www/views/task.py@api.route('/task')classTaskApi(Resource):def__init__(self,*args,**kwargs):super().__init__(*args,**kwargs)defput(self):config=request.get_json()task=long_time_task.apply_async(args=[config])defget(self):task_id=request.args.get('task_id',None)task=long_time_task.AsyncResult(task_id)return{'state':task.state,'current':0,'total':1,'result':task.result}5.定义CeleryWorker#celery_worker.pyfromwwwimportcelery,create_appapp=create_app()app.app_context().push()6。启动FlaskApp&Celery#!/usr/bin/envbashpython3application.py&celeryworker-Acelery_worker.celery--loglevel=info--pool=eventlet参考https://github.com/miguelgrin...https://github.com/miguelgrin...
