devbox
Monodyee
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flask框架配置celery-[1]：flask工厂模式集成使用celery，可在异步任务中使用flask应用上下文，即拿即用，无需更多配置_celery flask

作者：Monodyee | 2024-04-13 03:47:20

赞

踩

celery flask

一、概述

1、celery框架和flask框架在运行时，是在不同的进程中，资源是独占的。

2、celery异步任务如果想使用flask中的功能，如orm，是需要在flask应用上下文管理器中执行orm操作的

3、使用celery是需要使用到中间件的，简单点就使用redis做中间件

注意：

在flask工厂模式集成celery异步框架，在celery的异步任务中能够获取到flask的应用上下文管理器，也就是说在celery异步任务中你可以去调用flask项目中功能，如orm操作等。

使用本文配置，可以无需修改flask创建app应用的程序，直接将celery相关包创建，运行就可以使用，且能够在异步任务使用flask的功能。

二、项目结构

依赖环境：

  1. celery==4.4.7
  2. eventlet==0.33.3
  3. Flask==2.1.3
  4. Flask-Caching==1.10.1
  5. Flask-Cors==3.0.10
  6. Flask-Migrate==2.7.0
  7. Flask-RESTful==0.3.9
  8. Flask-SocketIO==5.1.1
  9. Flask-SQLAlchemy==2.5.1
  10. PyMySQL==1.0.2
  11. redis==3.5.3
  12. SQLAlchemy==1.4.0
  13. Werkzeug==2.0.2

目录结构：

flask-project

        |--apps

                |-- user

                        |-- models

                        |--views.py

                        |--urls.py

                |--__init__.py

        |--ext

                |--__init__.py

                |--config.py

        |--celery_task

                |--__init__.py

                |--async_task.py

                |--celery.py

                |--celeryconfig.py

                |--check_task.py

                |--scheduler_task.py

        app.py

三、flask工厂模式下各模块功能

1、apps/user/models.py ： 写了一个user表

2、apps/user/views.py:写了测试调用celery异步任务的接口

3、apps/user/urls.py: 注册路由的

4、ext/__init__.py：cache、db、cors的拓展

5、ext/config.py : cache和cors使用到的配置

6、apps/__init__.py： 一个函数create_app,生成flask应用对象

7、app.py: 启动flask应用对象的模块

本文重点不在flask工厂模式，默认看官都懂如何创建flaks工厂模式的项目了。

在视图中在执行异步任务，并获取异步任务的id：

  1. from celery_task.async_task import send_email_task,cache_user_task
  2. #用户资源：get\put\delete, 对单个进行操作
  3. class UserOneResource(ResourceBase):
  4.     def put(self,id):
  5.         #测试异步发邮件
  6.         email = request.args.get('email')
  7.         code = request.args.get('code')
  8.         res = send_email_task.delay(email,code)
  9.         print(res.id)
  10.         return NewResponse(msg='put',data={'task_id':res.id})
  11.  
  12.     def patch(self,id):
  13.         #测试异步操作flask的orm和cache
  14.         p = request.args.get('p')
  15.         if p=='set':
  16.             res = cache_user_task.delay()
  17.             print(res,type(res))
  18.             return NewResponse(msg='patch',data={'task_id':res.id})
  19.         else:
  20.             from ext import cache
  21.             data = cache.get('all-user-data')
  22.             return NewResponse(msg='patch',data=data)

res = 异步函数.delay(函数需要的参数)

task_id = res.id

注意：task_id 可以知道对应的任务的完成情况，获取任务的返回值等。

四、celery项目的配置

1、celery的配置

将celery的配置都放到一个py文件中，方便后期的维护和使用

celeryconfig.py

  1. from celery.schedules import crontab
  2. from datetime import timedelta
  3. '''
  4. 参数解析：
  5. accept_content：允许的内容类型/序列化程序的白名单，如果收到不在此列表中的消息，则该消息将被丢弃并出现错误，默认只为json；
  6. task_serializer：标识要使用的默认序列化方法的字符串，默认值为json；
  7. result_serializer：结果序列化格式，默认值为json；
  8. timezone：配置Celery以使用自定义时区；
  9. enable_utc：启用消息中的日期和时间，将转换为使用 UTC 时区，与timezone连用，当设置为 false 时，将使用系统本地时区。
  10. result_expires： 异步任务结果存活时长
  11. beat_schedule：设置定时任务
  12. '''
  13. #手动注册celery的异步任务：将所有celery异步任务所在的模块找到，写成字符串
  14. task_module = [
  15.     'celery_task.async_task',  # 写任务模块导入路径，该模块主要写异步任务的方法
  16.     'celery_task.scheduler_task',  # 写任务模块导入路径，该模块主要写定时任务的方法
  17. ]
  18.  
  19. #celery的配置
  20. config = {
  21.     "broker_url" :'redis://127.0.0.1:6379/0',   #'redis://:123456@127.0.0.1:6379/1' 有密码时，123456是密码
  22.     "result_backend" : 'redis://127.0.0.1:6379/1',
  23.     "task_serializer" : 'json',
  24.     "result_serializer" : 'json',
  25.     "accept_content" : ['json'],
  26.     "timezone" : 'Asia/Shanghai',
  27.     "enable_utc" : False,
  28.     "result_expires" : 1*60*60,
  29.     "beat_schedule" : { #定时任务配置
  30.             # 名字随意命名
  31.             'add-func-30-seconds': {
  32.                 # 执行add_task下的addy函数
  33.                 'task': 'celery_task.scheduler_task.add_func',  # 任务函数的导入路径，from celery_task.scheduler_task import add_func
  34.                 # 每10秒执行一次
  35.                 'schedule': timedelta(seconds=30),
  36.                 # add函数传递的参数
  37.                 'args': (10, 21)
  38.             },
  39.             # 名字随意起
  40.             'add-func-5-minutes': {
  41.                 'task': 'celery_task.scheduler_task.add_func',  # 任务函数的导入路径，from celery_task.scheduler_task import add_func
  42.                 # crontab不传的参数默认就是每的意思，比如这里是每年每月每日每天每小时的5分执行该任务
  43.                 'schedule': crontab(minute='5'),  # 之前时间点执行，每小时的第5分钟执行任务, 改成小时，分钟，秒 就是每天的哪个小时哪分钟哪秒钟执行
  44.                 'args': (19, 22)  # 定时任务需要的参数
  45.             },
  46.             # 缓存用户数据到cache中
  47.             'cache-user-func': {
  48.                 'task': 'celery_task.scheduler_task.cache_user_func',
  49.                 # 导入任务函数：from celery_task.scheduler_task import cache_user_func
  50.                 'schedule': timedelta(minutes=1),  # 每1分钟执行一次，将用户消息缓存到cache中
  51.             }
  52.         }
  53. }

2、创建celery对象

celery.py

  1. from celery import Celery,Task
  2. from .celeryconfig import config,task_module
  3. import sys
  4. import os
  5. '1、把flask项目路径添加到系统环境变量中'
  6. project_path = os.path.dirname(os.path.dirname(__file__))
  7. sys.path.append(project_path)
  8.  
  9. '''
  10. 2、创建celery应用对象
  11.   'task'可以任务是该celery对象名字，用于区分celery对象
  12.   broker是指定消息中间件
  13.   backend是指定任务结果存储位置
  14.   include是手动指定异步任务所在的模块的位置
  15. '''
  16. #创建celery异步对象
  17. celery = Celery('task', broker=config.get('broker_url'), backend=config.get('result_backend'), include=task_module)
  18. #导入一些基本配置
  19. celery.conf.update(**config)
  20.  
  21. '3、给celery所有任务添加flask的应用上下文，在celery异步任务中就可以调用flask中的对象了'
  22. class ContextTask(celery.Task):
  23.     def __call__(self, *args, **kwargs):
  24.         from apps import create_app
  25.         app = create_app()
  26.         with app.app_context():
  27.             return self.run(*args, **kwargs)
  28. celery.Task = ContextTask

注意：

1、第一步很关键，设置到python项目运行时，加载环境变量的问题。这一步是将flask项目的根目录加载环境变量中，这样第3步才能从apps中导入create_app函数。

2、第二步是创建celery通用的方法了，没什么好说的。

3、第三步很关键，涉及到celery异步任务能否在flask应用上下文管理器运行，从而可以调用flask中的功能，例如orm操作，cache操作.。（在执行任务时，先套上flask的应用上下文管理器）

3、异步任务模块

将所有异步任务相关的函数都集中到一个模块中，方便维护和使用。

async_task.py

  1. # 导入celery对象app
  2. from celery_task.celery import celery
  3. from ext import cache
  4. import time
  5.  
  6.  
  7. '''
  8. 1、没有返回值的，@app.task(ignore_result=True)
  9. 2、有返回值的任务，@app.task 默认就是(ignore_result=False)
  10. '''
  11.  
  12.  
  13. # 没有返回值，禁用掉结果后端
  14. @celery.task
  15. def send_email_task(receiver_email,code):  # 此时可以直接传邮箱，还能减少一次数据库的IO操作
  16.     '''
  17.     :param email: 接收消息的邮箱，用户的邮箱
  18.     :return:
  19.     '''
  20.     # 模拟邮件发送验证码
  21.     time.sleep(5)
  22.     return {'result':'邮件已经发送',receiver_email:'2356'}
  23.  
  24. @celery.task
  25. def cache_user_task():
  26.     #orm查询数据，放到cache中
  27.     from apps.user.models import UserModel
  28.     user = UserModel.query.all()
  29.     lis = []
  30.     for u in user:
  31.         id = u.id
  32.         name = u.name
  33.         dic = {'id':id,'name':name}
  34.         lis.append(dic)
  35.         print(dic)
  36.     cache.set('all-user-data',lis)
  37.     return {'code':200,'msg':'查询数据成功'}

4、定时任务模块

将所有定时任务相关的函数都集中到一个模块中，方便维护和使用。

schedulser_task.py

  1. from celery_task.celery import celery
  2. import time
  3.  
  4.  
  5. # 有返回值，返回值可以从结果后端中获取
  6. @celery.task
  7. def add_func(a, b):
  8.     print('执行了加法函数',a+b)
  9.     return a + b
  10.  
  11.  
  12. # 不需要返回值，禁用掉结果后端
  13. @celery.task(ignore_result=True)
  14. def cache_user_func():
  15.     print('all')
  16.  
  17.

5、检测任务id获取任务状态和返回值

check_task.py:

  1. from celery.result import AsyncResult
  2. from celery_task.celery import celery
  3.  
  4. '''验证任务的执行状态的'''
  5.  
  6.  
  7. def check_task_status(task_id):
  8.     '''
  9.     任务的执行状态：
  10.         PENDING :等待执行
  11.         STARTED :开始执行
  12.         RETRY   :重新尝试执行
  13.         SUCCESS :执行成功
  14.         FAILURE :执行失败
  15.     :param task_id:
  16.     :return:
  17.     '''
  18.     result = AsyncResult(id=task_id, app=celery)
  19.     dic = {
  20.         'type': result.status,
  21.         'msg': '',
  22.         'data': None,
  23.         'code': 400
  24.     }
  25.     if result.status == 'PENDING':
  26.         dic['msg'] = '任务等待中'
  27.     elif result.status == 'STARTED':
  28.         dic['msg'] = '任务开始执行'
  29.     elif result.status == 'RETRY':
  30.         dic['msg'] = '任务重新尝试执行'
  31.     elif result.status == 'FAILURE':
  32.         dic['msg'] = '任务执行失败了'
  33.     elif result.status == 'SUCCESS':
  34.         result = result.get()
  35.         dic['msg'] = '任务执行成功'
  36.         dic['data'] = result
  37.         dic['code'] = 200
  38.         # result.forget() # 将结果删除
  39.         # async.revoke(terminate=True)  # 无论现在是什么时候，都要终止
  40.         # async.revoke(terminate=False) # 如果任务还没有开始执行呢，那么就可以终止。
  41.     return dic
  42.

在视图函数中调用该方法，通过task_id ,返回任务的运行结果。

五、测试

1、运行项目

flask项目（在项目根目录下执行）：

        flask run --host 0.0.0.0 --port 5000

celery项目（在项目根目录下执行）:

启动celery进程：

windows系统：

        celery -A celery_task.celery worker -l info  -P  eventlet

linux系统：

        celery -A celery_task.celery worker -l info 

启动定时任务（先启动celery进程在启动定时任务）：

celery -A celery_task.celery beat -l info

2、运行结果

1、执行异步任务中，将orm数据存到cache中

2、执行定时任务了

六、注意事项

1、在系统中要先安装好redis和mysql，并都启动了

2、在测试异步操作orm时，会使用到flask的cache存数据，注意flask的cache不能配置内存模式，不然celery进程存到cache中的数据，flask进程中取不到的。

3、当前的配置下，celery的目录必须是在flask根目录下

七、拓展-改变celery_task的位置

如果你想将celery_task包移动到apps包下，此时你需要修改什么？

1、apps/celery_task/celery.py：将flask项目根目录加载到系统环境变量中的路径有变

  1. '1、把flask项目路径添加到系统环境变量中'
  2. project_path = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))

2、apps/celery_task/celeryconfig.py: 注册异步任务的模块，定时任务的模块的位置变化

  1. '1、加上apps.'
  2. task_module = [
  3.     'apps.celery_task.async_task',  # 写任务模块导入路径，该模块主要写异步任务的方法
  4.     'apps.celery_task.scheduler_task',  # 写任务模块导入路径，该模块主要写定时任务的方法
  5. ]
  6.  
  7.  
  8. '2、task参数对应的字符串，加上apps.'
  9. config = {
  10.     "broker_url" :'redis://127.0.0.1:6379/0',   #'redis://:123456@127.0.0.1:6379/1' 有密码时，123456是密码
  11.     "result_backend" : 'redis://127.0.0.1:6379/1',
  12.     "task_serializer" : 'json',
  13.     "result_serializer" : 'json',
  14.     "accept_content" : ['json'],
  15.     "timezone" : 'Asia/Shanghai',
  16.     "enable_utc" : False,
  17.     "result_expires" : 1*60*60,
  18.     "beat_schedule" : { #定时任务配置
  19.             # 名字随意命名
  20.             'add-func-30-seconds': {
  21.                 # 执行add_task下的addy函数
  22.                 'task': 'apps.celery_task.scheduler_task.add_func',  # 任务函数的导入路径，from celery_task.scheduler_task import add_func
  23.                 # 每10秒执行一次
  24.                 'schedule': timedelta(seconds=30),
  25.                 # add函数传递的参数
  26.                 'args': (10, 21)
  27.             },
  28.             # 名字随意起
  29.             'add-func-5-minutes': {
  30.                 'task': 'apps.celery_task.scheduler_task.add_func',  # 任务函数的导入路径，from celery_task.scheduler_task import add_func
  31.                 # crontab不传的参数默认就是每的意思，比如这里是每年每月每日每天每小时的5分执行该任务
  32.                 'schedule': crontab(minute='5'),  # 之前时间点执行，每小时的第5分钟执行任务, 改成小时，分钟，秒 就是每天的哪个小时哪分钟哪秒钟执行
  33.                 'args': (19, 22)  # 定时任务需要的参数
  34.             },
  35.             # 缓存用户数据到cache中
  36.             'cache-user-func': {
  37.                 'task': 'apps.celery_task.scheduler_task.cache_user_func',
  38.                 # 导入任务函数：from celery_task.scheduler_task import cache_user_func
  39.                 'schedule': timedelta(minutes=1),  # 每1分钟执行一次，将用户消息缓存到cache中
  40.             }
  41.         }
  42. }

3、在视图函数导入异步任务的路径也变了

  1. #异步任务
  2. from apps.celery_task.async_task import send_email_task,cache_user_task

4、启动celery和定时任务的命令变量【在项目根目录下执行命令】

启动celery：

windows启动命令: celery  -A  apps.celery_task.celery worker -l info  -P  eventlet

linux启动命令: celery  -A  apps.celery_task.celery worker -l info 

启动定时任务：

celery -A apps.celery_task beat -l info

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Monodyee/article/detail/414256
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号