您现在的位置是：网站首页> 编程资料编程资料

Python Celery动态添加定时任务生产实践指南_python_

2023-05-26 312人已围观

简介 Python Celery动态添加定时任务生产实践指南_python_

一、背景

实际工作中会有一些耗时的异步任务需要使用定时调度，比如发送邮件，拉取数据，执行定时脚本

通过celery 实现调度主要思想是通过引入中间人redis，启动 worker 进行任务执行，celery-beat进行定时任务数据存储

二、Celery动态添加定时任务的官方文档

celery文档：https://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html#beat-custom-schedulers

celery 自定义调度类说明:

自定义调度器类可以在命令行中指定（--scheduler参数）

django-celery-beat文档 : https://pypi.org/project/django-celery-beat/

关于django-celery-beat 插件的说明:

此扩展使您能够将定期任务计划存储在数据库中，可以从 Django 管理界面管理周期性任务，您可以在其中创建、编辑和删除周期性任务以及它们应该运行的频率

三、celery简单实用

3.1 基础环境配置

1. 安装最新版本的Django

pip3 install django #当前我安装的版本是 3.0.6

2. 创建项目

django-admin startproject typeidea django-admin startapp blog

3.安装 celery

pip3 install django-celery pip3 install -U Celery pip3 install "celery[librabbitmq,redis,auth,msgpack]" pip3 install django-celery-beat # 用于动态添加定时任务 pip3 install django-celery-results pip3 install redis

3.2 测试使用Celery应用

1. 创建blog目录、新建task.py

首先在Django项目中创建一个blog文件夹，并且在blog文件夹下创建tasks.py模块, 如下：

tasks.py代码如下:

#!/usr/bin/env python # -*- coding: UTF-8 -*- """ #File: tasks.py #Time: 2022/3/30 2:26 下午 #Author: julius """ from celery import Celery # 使用redis做为broker app = Celery('blog.tasks2',broker='redis://127.0.0.1:6379/0') # 创建任务函数 @app.task def my_task(): print('任务正在执行...')

Celery第一个参数是给其设定一个名字，第二参数我们设定一个中间人broker, 在这里我们使用Redis作为中间人。my_task函数是我们编写的一个任务函数，通过加上装饰器app.task, 将其注册到broker的队列中。

2. 启动redis、创建worker

现在我们在创建一个worker，等待处理队列中的任务。

进入项目的根目录，执行命令: celery -A celery_tasks.tasks worker -l info

3. 调用任务

下面来测试一下功能，创建一个任务，加入任务队列中，提供worker执行。

进入python终端, 执行如下代码:

$ python manage.py shell >>> from blog.tasks import my_task >>> my_task.delay()

调用一个任务函数，将会返回一个AsyncResult对象，这个对象可以用来检查任务的状态或者获得任务的返回值。

4. 查看结果

在worker的终端查看任务执行情况，可以看到已经收到83484dfe-f729-417b-8e51-6c7ae32a1377 任务，并打印了任务执行信息

5. 存储并查看任务执行状态

把任务执行结果赋值给ret，然后调用result() 会产生 DisabledBackend 报错，可见没有配置后端存储的时候并不能保存任务执行的状态信息，下一节我们会讲到如何配置backend保存任务执行结果

$ python manage.py shell >>> from blog.tasks import my_task >>> ret=my_task.delay() >>> ret.result()

四、配置backend存储任务执行结果

如果我们想跟踪任务的状态，Celery需要将结果保存到某个地方。有几种保存的方案可选:SQLAlchemy、Django ORM、Memcached、 Redis、RPC (RabbitMQ/AMQP)。

1. 添加backend参数

在本例中我们使用Redis作为存储结果的方案，通过Celery的backend参数来设定任务结果存储地址。我们将tasks模块修改如下:

from celery import Celery # 使用redis作为broker以及backend app = Celery('celery_tasks.tasks', broker='redis://127.0.0.1:6379/8', backend='redis://127.0.0.1:6379/9') # 创建任务函数 @app.task def my_task(a, b): print("任务函数正在执行....") return a + b

给Celery增加了backend参数，指定redis作为结果存储,并将任务函数修改为两个参数，并且有返回值。

2. 调用任务/查看任务执行结果

下面再来执行调用一下这个任务看看。

$ python manage.py shell >>> from blog.tasks import my_task >>> res=my_task.delay(10,40) >>> res.result 50 >>> res.failed() False

再来看看worker的执行情况，如下：

可以看到celery任务已经执行成功了。

但是这只是一个开始，下一步要看看如何添加定时的任务。

四、优化Celery目录结构

上面直接将Celery的应用创建、配置、tasks任务全部写在了一个文件，这样在后面项目越来越大，也是不方便的。下面来拆分一下，并且添加一些常用的参数。

基本结构如下

$ vim typeidea/celery.py (Celery应用文件)

#!/usr/bin/env python # -*- coding: UTF-8 -*- """ #File: celery.py #Time: 2022/3/30 12:25 下午 #Author: julius """ import os from celery import Celery from blog import celeryconfig project_name='typeidea' # set the default django setting module for the 'celery' program os.environ.setdefault('DJANGO_SETTINGS_MODULE','typeidea.settings') app = Celery(project_name) app.config_from_object('django.conf:settings') app.autodiscover_tasks()

vim blog/celeryconfig.py (配置Celery的参数文件)

#!/usr/bin/env python # -*- coding: UTF-8 -*- """ #File: celeryconfig.py #Time: 2022/3/30 2:54 下午 #Author: julius """ # 设置结果存储 from typeidea import settings import os os.environ.setdefault("DJANGO_SETTINGS_MODULE", "typeidea.settings") CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0' # 设置代理人broker BROKER_URL = 'redis://127.0.0.1:6379/1' # celery 的启动工作数量设置 CELERY_WORKER_CONCURRENCY = 20 # 任务预取功能，就是每个工作的进程／线程在获取任务的时候，会尽量多拿 n 个，以保证获取的通讯成本可以压缩。 CELERYD_PREFETCH_MULTIPLIER = 20 # 非常重要,有些情况下可以防止死锁 CELERYD_FORCE_EXECV = True # celery 的 worker 执行多少个任务后进行重启操作 CELERY_WORKER_MAX_TASKS_PER_CHILD = 100 # 禁用所有速度限制，如果网络资源有限，不建议开足马力。 CELERY_DISABLE_RATE_LIMITS = True CELERY_ENABLE_UTC = False CELERY_TIMEZONE = settings.TIME_ZONE DJANGO_CELERY_BEAT_TZ_AWARE = False CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

vim blog/tasks.py (tasks 任务文件)

import time from blog.celery import app # 创建任务函数 @app.task def my_task(a, b, c): print('任务正在执行...') print('任务1函数休眠10s') time.sleep(10) return a + b + c

五、开始使用django-celery-beat调度器

使用 django-celery-beat 动态添加定时任务 celery 4.x 版本在 django 框架中是使用 django-celery-beat 进行动态添加定时任务的。前面虽然已经安装了这个库，但是还要再说明一下。

官网的配置说明
https://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html#beat-custom-schedulers

1. 安装 django-celery-beat

pip3 install django-celery-beat

2.在项目的 settings 文件配置 django-celery-beat

INSTALLED_APPS = [ 'blog', 'django_celery_beat', ... ] # Django设置时区 LANGUAGE_CODE = 'zh-hans' # 使用中国语言 TIME_ZONE = 'Asia/Shanghai' # 设置Django使用中国上海时间 # 如果USE_TZ设置为True时，Django会使用系统默认设置的时区，此时的TIME_ZONE不管有没有设置都不起作用 # 如果USE_TZ 设置为False,TIME_ZONE = 'Asia/Shanghai', 则使用上海的UTC时间。 USE_TZ = False

3. 创建 django-celery-beat 相关表

执行Django数据库迁移: python manage.py migrate

4. 配置Celery使用 django-celery-beat

配置 celery.py

import os from celery import Celery from blog import celeryconfig # 为celery 设置环境变量 os.environ.setdefault("DJANGO_SETTINGS_MODULE","typeidea.settings") # 创建celery app app = Celery('blog') # 从单独的配置模块中加载配置 app.config_from_object(celeryconfig) # 设置app自动加载任务 app.autodiscover_tasks([ 'blog', ])

配置 celeryconfig.py

# 设置结果存储 from typeidea import settings import os os.environ.setdefault("DJANGO_SETTINGS_MODULE", "typeidea.settings") CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0' # 设置代理人broker BROKER_URL = 'redis://127.0.0.1:6379/1' # celery 的启动工作数量设置 CELERY_WORKER_CONCURRENCY = 20 # 任务预取功能，就是每个工作的进程／线程在获取任务的时候，会尽量多拿 n 个，以保证获取的通讯成本可以压缩。 CELERYD_PREFETCH_MULTIPLIER = 20 # 非常重要,有些情况下可以防止死锁 CELERYD_FORCE_EXECV = True # celery 的 worker 执行多少个任务后进行重启操作 CELERY_WORKER_MAX_TASKS_PER_CHILD = 100 # 禁用所有速度限制，如果网络资源有限，不建议开足马力。 CELERY_DISABLE_RATE_LIMITS = True CELERY_ENABLE_UTC = False CELERY_TIMEZONE = settings.TIME_ZONE DJANGO_CELERY_BEAT_TZ_AWARE = False CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

编写任务 tasks.py

import time from celery import Celery from blog.celery import app # 使用redis做为broker # app = Celery('blog.tasks2',broker='redis://127.0.0.1:6379/0',backend='redis://127.0.0.1:6379/1') # 创建任务函数 @app.task def my_task(a, b, c): print('任务正在执行...') print('任务1函数休眠10s') time.sleep(10) return a + b + c @app.tas
                提示：
                    本文由神整理自网络，如有侵权请联系本站删除！
                    

                    本站声明： 

                    1、本站所有资源均来源于互联网，不保证100%完整、不提供任何技术支持； 

                    2、本站所发布的文章以及附件仅限用于学习和研究目的;不得将用于商业或者非法用途；否则由此产生的法律后果，本站概不负责！
                
                
                
                                            上一篇：python3里gbk编码的问题解决_python_
                                                                下一篇：Python Flask框架使用介绍_python_