全网最细之Celery 4.x动态添加定时任务

微信扫一扫,分享到朋友圈

全网最细之Celery 4.x动态添加定时任务

首先我们有一个这样的需求 我们为了能在Web端口动态的添加定时任务,本次来调研一下Celery 4.x 在Django框架下 要如何去动态添加定时任务。
我们新建一个Django项目
安装最新的Django
pip install django
复制代码
然后创建项目 django-admin startproject 项目名称 执行结果:
django-admin startproject django_con .
复制代码
安装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
复制代码
接下来安装相关依赖:
amqp==2.5.2
anyjson==0.3.3
asgiref==3.2.7
billiard==3.6.3.0
celery==4.4.2
cffi==1.14.0
cryptography==2.9.2
Django==3.0.6
django-celery==3.3.1
django-celery-beat==2.0.0
django-celery-results==1.2.1
django-timezone-field==4.0
dnspython==1.16.0
eventlet==0.25.2
greenlet==0.4.15
importlib-metadata==1.6.0
kombu==4.6.8
monotonic==1.5
pycparser==2.20
python-crontab==2.4.2
python-dateutil==2.8.1
pytz==2020.1
redis==3.5.1
six==1.14.0
sqlparse==0.3.1
vine==1.3.0
zipp==3.1.0
复制代码
测试celery
创建celery目录 第一步:在Django项目中创建一个celery_tasks文件夹,然后再创建tasks.py图下:

第二步:编写tasks.py内容为图下:
from celery import Celery

使用redis作为broker

app = Celery(‘celery_tasks.tasks’, broker=’redis://192.168.196.135:6379/8′)

创建任务函数

@app.task
def my_task():
print(“任务函数正在执行….”)
复制代码
第三步:celery第一个参数设定一个名字,第二个参数设定一个中介broker,在这我是用redis 座位中介。my_task函数是我编写的一个任务函数,通过加上装饰器app.task,让它注册道broker队列中。
我们进入项目的根目录,执行命令:celery -A celery_tasks.tasks worker -l info图下:
 
调用任务
我们接下来测试功能,创建一个任务,加入道任务队列中,提供worker执行。
进入python终端,执行下面代码:
[root@python_env django_cron]# python3 manage.py shell
from celery_tasks.tasks import my_task

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

my_task.delay()
输出结果: <AsyncResult: 647b2589-95d2-45c9-a9a7-0b5530caf249>
复制代码
然后返回worker终端界面,查看一下任务的执行,图下:
 
现在就可以看到已经收到了任务,并且打印出了信息。
储存结果
要是我们想跟踪任务的状态,Celery就需要将结果保存到一个地方。 有几种方式可以保存:SQLAlchemy、Django ORM、Memcached、 Redis、RPC (RabbitMQ/AMQP)
加入我们仍然使用Redis座位储存的结果,任务结果储存配置我们通过Celery的backend 参数来设定。我们要将tasks模块修改:
from celery import Celery

使用redis作为broker以及backend

app = Celery(‘celery_tasks.tasks’,
broker=’redis://192.168.196.135:6379/8′,
backend=’redis://192.168.196.135:6379/9′)

创建任务函数

@app.task
def my_task(a, b):
print(“任务函数正在执行….”)
return a + b
复制代码
我给Celery增加了backend参数,指定redis作为结果存储,并将任务函数修改为两个参数,并且有返回值。
借来下我们执行调用一下这个任务看看。
from celery_tasks.tasks import my_task

传递参数至任务中

ret = my_task.delay(10,20)

查询返回值的结果

ret.result
输出结果: 30

查看是否执行失败

ret.failed()
输出结果: False
复制代码
再来看worker的执行情况,图下:
 
可以看到celery任务已经执行成功了。
这只是一个开始,下一步要看看如何添加定时的任务。
优化Celery目录
上面直接将Celery的应用创建、配置、tasks任务全部写在了一个文件,这样在后面项目越来越大,也是不方便的。下面来拆分一下,并且添加一些常用的参数 图下:
 
创建Celery应用的文件 celery.py
​from celery import Celery
from celery_tasks import celeryconfig

import os

为celery设置环境变量

os.environ.setdefault(“DJANGO_SETTINGS_MODULE”, “django_con.settings”)

创建celery app

app = Celery(‘celery_tasks’)

从单独的配置模块中加载配置

app.config_from_object(celeryconfig)

设置app自动加载任务

app.autodiscover_tasks([
‘celery_tasks’,
])
复制代码
配置Celery的参数文件 celeryconfig.py

设置结果存储

CELERY_RESULT_BACKEND = ‘redis://192.168.196.135:6379/9’

设置代理人broker

BROKER_URL = ‘redis://192.168.196.135:6379/8’

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
复制代码
tasks 任务文件 tasks.py
from celery_tasks.celery import app

创建任务函数

@app.task
def my_task(a, b, c):
print(“任务函数正在执行….”)
return a + b + c
复制代码
接下来我们开始安装使用定时任务
安装 django-celery-beat
pip3 install django-celery-beat
复制代码
配置 django-celery-beat

安装应用 django_celery_beat

INSTALLED_APPS = [
‘django_celery_beat’, # 安装 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
复制代码
在 celerconfig.py 中配置 django_celery_beat:
from django.conf import settings

import os

为celery设置环境变量

os.environ.setdefault(“DJANGO_SETTINGS_MODULE”, “django_con.settings”)

celery beat配置

CELERY_ENABLE_UTC = False

CELERY_TIMEZONE = settings.TIME_ZONE
DJANGO_CELERY_BEAT_TZ_AWARE = False
CELERY_BEAT_SCHEDULER = ‘django_celery_beat.schedulers:DatabaseScheduler’
复制代码
创建 django-celery-beat 相关表
执行Django数据库迁移:python3 manage.py migrate图下:
 
配置Celery使用 django-celery-beat
配置celery.py:
from celery import Celery
from celery_tasks import celeryconfig
from django.utils import timezone

import os

为celery设置环境变量

os.environ.setdefault(“DJANGO_SETTINGS_MODULE”, “django_con.settings”)

创建celery app

app = Celery(‘celery_tasks’)

从单独的配置模块中加载配置

app.config_from_object(celeryconfig)

设置app自动加载任务

app.autodiscover_tasks([
‘celery_tasks’,
])

解决时区问题,定时任务启动就循环输出

app.now = timezone.now
复制代码
配置 celeryconfig.py:
from django.conf import settings
import os

为celery设置环境变量

os.environ.setdefault(“DJANGO_SETTINGS_MODULE”, “django_con.settings”)

设置结果存储

CELERY_RESULT_BACKEND = ‘redis://192.168.196.135:6379/9’

设置代理人broker

BROKER_URL = ‘redis://192.168.196.135:6379/8’

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 beat配置

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:
from celery_tasks.celery import app

创建任务函数

@app.task
def my_task1(a, b, c):
print(“任务1函数正在执行….”)
return a + b + c

@app.task def my_task2(): print(“任务2函数正在执行….”) 复制代码 启动定时任务work:  启动定时任务前提需要有一个work执行异步任务,然后再启动一个定时器触发任务   启动任务 work celery -A celery_tasks worker -l info 复制代码 启动定时器触发 beat celery -A celery_tasks beat -l info –scheduler django_celery_beat.schedulers:DatabaseScheduler 复制代码 创建定时任务说明  创建定时任务可以查看 django_celery_beat 的官网说明:pypi.org/project/dja…

下面先来翻译看看官网的示例,然后再实际应用一下。 
官网示例说明 
创建基于间隔时间的周期性任务 
初始化周期间隔对象 interval 对象 
创建一个基于间隔时间的周期性任务之前,首先需要创建一个 interval 对象,用于提供任务设置周期间隔:
from django_celery_beat.models import PeriodicTask, IntervalSchedule

executes every 10 seconds.

schedule, created = IntervalSchedule.objects.get_or_create(
every=10,
period=IntervalSchedule.SECONDS,
)
复制代码
可以看到上面固定间隔的时间是采用秒 period=IntervalSchedule.SECONDS,如果你还想要固定其他的时间单位,可以设置其他字段参数,如下:
  1,IntervalSchedule.DAYS 固定间隔天数2,IntervalSchedule.HOURS 固定间隔小时数3,IntervalSchedule.MINUTES 固定间隔分钟数4,IntervalSchedule.SECONDS 固定间隔秒数5,IntervalSchedule.MICROSECONDS 固定间隔微秒 
注意: 如果你有多个周期性任务都是间隔10秒,那么这些任务都应该设置同一个interval 对象 另外,可以如果不清楚有哪些固定的时间单位,可以这样查看,如下:
from django_celery_beat.models import PeriodicTask, IntervalSchedule

IntervalSchedule.PERIOD_CHOICES

输出结果:
((‘days’, ‘Days’),
(‘hours’, ‘Hours’),
(‘minutes’, ‘Minutes’),
(‘seconds’, ‘Seconds’),
(‘microseconds’, ‘Microseconds’))
复制代码
创建周期性间隔任务 
下面这种是无参数的创建方法:
PeriodicTask.objects.create(
interval=schedule, # we created this above.
name=’Importing contacts’, # simply describes this periodic task.
task=’proj.tasks.import_contacts’, # name of task.
)
复制代码
带参数的创建方法,如下:
import json
from datetime import datetime, timedelta

PeriodicTask.objects.create(
interval=schedule, # we created this above.
name=’Importing contacts’, # simply describes this periodic task.
task=’proj.tasks.import_contacts’, # name of task.
args=json.dumps([‘arg1’, ‘arg2’]),
kwargs=json.dumps({
‘be_careful’: True,
}),
expires=datetime.utcnow() + timedelta(seconds=30)
)
复制代码
创建基于 crontab 的周期性任务
初始化 crontab 的调度对象 
上面是创建基于固定周期的调度对象,那么 crontab 就是类似linux 中的crontab 定时方式。 crontab 调度对象有如下字段:minute, hour, day_of_week, day_of_month 、month_o对应配置30 * * * * 的 crontab 定时写法。
from django_celery_beat.models import CrontabSchedule, PeriodicTask
schedule, _ = CrontabSchedule.objects.get_or_create(
minute=’30’,
hour=’ ‘,
day_of_week=’

‘,
day_of_month=’ ‘,
month_of_year=’

‘,
timezone=pytz.timezone(‘Canada/Pacific’)
)
复制代码
注意上面的这个时区设置,中国的时区应该设置为 timezone=pytz.timezone(‘Asia/Shanghai’) 
创建基于 crontab 调度的定时任务
创建任务的方式跟创建固定间隔时间的周期性任务基本一致,只不过将 interval=schedule 改为了 crontab=schedule,有参数的写法也是一致:
PeriodicTask.objects.create(
crontab=schedule,
name=’Importing contacts’,
task=’proj.tasks.import_contacts’,
)
复制代码
暂时停止周期性任务
periodic_task.enabled = False
periodic_task.save()
复制代码
启动运行周期任务的示例 
执行周期性任务的前提是需要有 workers 去执行,那么首先需要已经安装好了 Celery,上面我们已经安装好了。也就是跟我前面说的,celery的 workers 和 beat 定时服务都需要同时开启。 
1,开启celery 的 worker 服务: $ celery -A [project-name] worker –loglevel=info 
2,作为一个单独的进程,启动beat服务:
$ celery -A [project-name] beat -l info –scheduler django_celery_beat.schedulers:DatabaseScheduler
复制代码
或者你也可以使用-S (scheduler flag) 标识符,更多参数说明查看 celery beat –help
$ celery -A [project-name] beat -l info -S django
复制代码
另外,作为替代方案,你也可以只使用一个命令运行上面的两个步骤(worker和beat服务)(建议只用于开发环境)
$ celery -A [project-name] worker –beat –scheduler django –loglevel=info
复制代码
3,现在启动完这两个服务,就可以开始添加周期性任务了。
具体操作演练
看完了上面官网的说明,下面拿我前面写好的两个task任务来创建一下周期性任务。
创建基于间隔时间的周期性任务
初始化周期间隔对象interval 对象
In [1]: from django_celery_beat.models import PeriodicTask, IntervalSchedule

创建一个间隔10秒钟的 interval 对象

schedule, created = IntervalSchedule.objects.get_or_create(
every=10,
period=IntervalSchedule.SECONDS,
)

查询已创建的 interval 对象

IntervalSchedule.objects.all()

输出结果: <QuerySet [<IntervalSchedule: every 10 seconds>]>
复制代码
创建周期性间隔任务
创建一个无参数的周期性间隔任务
PeriodicTask.objects.create(
interval=schedule, # 上面创建10秒的间隔 interval 对象
name=’my_task2′, # 设置任务的name值
task=’celery_tasks.tasks.my_task2′, # 指定需要周期性执行的任务
)

输出结果: <PeriodicTask: my_task1: every 10 seconds>

复制代码
创建之后,beat服务日志显示如下:
[2020-09-23 14:03:00,146: INFO/MainProcess] Scheduler: Sending due task my_task2 (celery_tasks.tasks.my_task2)
[2020-09-23 14:03:09,189: INFO/MainProcess] Scheduler: Sending due task my_task2 (celery_tasks.tasks.my_task2)
复制代码
worker服务日志显示如下:
[2020-09-23 14:02:30,161: INFO/MainProcess] Received task: celery_tasks.tasks.my_task2[43365c48-9b52-44ea-ba5f-d75cd7df49dd]

[2020-09-23 14:02:30,163: WARNING/ForkPoolWorker-1] 任务2函数正在执行….
[2020-09-23 14:02:30,166: INFO/ForkPoolWorker-1] Task celery_tasks.tasks.my_task2[43365c48-9b52-44ea-ba5f-d75cd7df49dd] succeeded in 0.003136722996714525s: None
[2020-09-23 14:02:70,161: INFO/MainProcess] Received task: celery_tasks.tasks.my_task2[770beba4-4c3f-4e2f-8d36-7ee0b90dd3b9]

[2020-09-23 14:02:70,164: WARNING/ForkPoolWorker-1] 任务2函数正在执行….
[2020-09-23 14:02:70,167: INFO/ForkPoolWorker-1] Task celery_tasks.tasks.my_task2[770beba4-4c3f-4e2f-8d36-7ee0b90dd3b9] succeeded in 0.0030223939975257963s: None
[2020-09-23 14:03:20,184: INFO/MainProcess] Received task: celery_tasks.tasks.my_task2[4b87f2f2-30d0-4c20-b028-601f28cb8193]

[2020-09-23 14:03:20,186: WARNING/ForkPoolWorker-1] 任务2函数正在执行….
复制代码
创建一个带参数的周期性间隔任务:
import json

from datetime import datetime, timedelta

PeriodicTask.objects.create(

interval=schedule, # 设置使用上面创建的 10 秒间隔 interval 对象
name=’my_task1′, # 设置周期性任务的名称
task=’celery_tasks.tasks.my_task1′, # 设置指定执行的task
args=json.dumps([10, 20, 30]), # 传递task需要的参数
expires=datetime.now() + timedelta(seconds=30) # 任务的执行超时时间,避免任务执行时间过长
)

输出结果: <PeriodicTask: my_task1: every 10 seconds>
复制代码
查看beat服务的日志:
 
 查看worker服务的日志:
 
周期性任务在worker是否串行执行还是并行?
这里有个疑问,如果只有一个worker,其中一个task执行时间比较长,例如:上面的两个任务都设置休眠10秒,确认是否可以同时执行,还是要开启多个worker执行。 
设置 task 任务进行休眠
from celery_tasks.celery import app
import time

创建任务函数

@app.task
def my_task1(a, b, c):
print(“任务1函数正在执行….”)
print(“任务1函数休眠10秒…”)
time.sleep(10)
return a + b + c

@app.task
def my_task2():
print(“任务2函数正在执行….”)
print(“任务2函数休眠10秒….”)
time.sleep(10)
复制代码
删除这两个周期性任务,然后再创建后查看beat服务以及worker服务日志
删除之前的两个周期性任务:

暂停执行两个周期性任务

PeriodicTask.objects.get(name=”my_task1″).enabled = False

PeriodicTask.objects.get(name=”my_task1″).save()

PeriodicTask.objects.get(name=”my_task2″).enabled = False

PeriodicTask.objects.get(name=”my_task2″).save()

删除任务

PeriodicTask.objects.get(name=”my_task1″).delete()
输出结果: (1, {‘django_celery_beat.PeriodicTask’: 1})
PeriodicTask.objects.get(name=”my_task2″).delete()
输出结果: (1, {‘django_celery_beat.PeriodicTask’: 1})
复制代码
重启beat服务、worker服务: 因为修改了task,需要重启服务才能重新加载。

启动beat进程

celery -A celery_tasks beat -l info –scheduler
django_celery_beat.schedulers:DatabaseScheduler

启动一个worker进程

celery -A celery_tasks worker -l info
复制代码
重新创建两个周期性任务:
from django_celery_beat.models import PeriodicTask, IntervalSchedule

创建一个间隔10秒钟的 interval 对象

schedule, created = IntervalSchedule.objects.get_or_create(
every=10,
period=IntervalSchedule.SECONDS,
)

创建无参数周期性任务

PeriodicTask.objects.create(
interval=schedule, # 上面创建10秒的间隔 interval 对象
name=’my_task2′, # 设置任务的name值
task=’celery_tasks.tasks.my_task2′, # 指定需要周期性执行的任务
)

输出结果: <PeriodicTask: my_task1: every 10 seconds>

创建带参数周期性任务

import json

from datetime import datetime, timedelta

PeriodicTask.objects.create(

interval=schedule, # 设置使用上面创建的 10 秒间隔 interval 对象
name=’my_task1′, # 设置周期性任务的名称
task=’celery_tasks.tasks.my_task1′, # 设置指定执行的task
args=json.dumps([10, 20, 30]), # 传递task需要的参数
expires=datetime.now() + timedelta(seconds=30)
)

输出结果: <PeriodicTask: my_task1: every 10 seconds>
复制代码
查看beat推送周期性任务的日志: 

查看单个worker的执行日志:
 
可以看到,因为worker不能并行执行任务,所以任务从beat发出来之后,在单个worker是串行执行的,所以如果想要并发执行worker,可以开启多线程的方式,或者开启多个进程。
开启2个worker来查看执行日志:
 
所以需要并行执行任务的时候,就需要设置多个worker来执行任务。
创建基于 crontab 的周期性任务
无限一直循环执行的BUG 
crontab周期性任务在使用的时候会出现beat服务一直不停发任务的情况,导致无法使用。目前尝试多种方式,仍未有解决的办法。 初始化 crontab 的调度对象:
import pytz

from django_celery_beat.models import CrontabSchedule, PeriodicTask schedule, _ = CrontabSchedule.objects.get_or_create(
minute=’ ‘,
hour=’

‘,
day_of_week=’ ‘,
day_of_month=’

‘,
month_of_year=’*’,
timezone=pytz.timezone(‘Asia/Shanghai’)
)

CrontabSchedule.objects.all()

输出结果: <QuerySet [<CrontabSchedule: * * * * * (m/h/d/dM/MY) Asia/Shanghai>, <CrontabSchedule: 0 4 * * * (m/h/d/dM/MY) Asia/Shanghai>]>
复制代码
创建一个无参数的定时任务:
PeriodicTask.objects.create(

crontab=schedule, # 上面创建的 crontab 对象 * * * * *,表示每分钟执行一次
name=’my_task2_crontab’, # 设置任务的name值
task=’celery_tasks.tasks.my_task2′, # 指定需要周期性执行的任务
)

输出结果: <PeriodicTask: my_task2_crontab: * * * * * (m/h/d/dM/MY) Asia/Shanghai>
复制代码
beat服务不停发任务的日志,如下:
 
周期性任务的查询、删除等操作
其实周期性任务也是存储在数据库的数据,基本上是基于ORM的操作的。 
周期性任务的查询

导入周期性任务

from django_celery_beat.models import PeriodicTask

查询目前所有的周期性任务

PeriodicTask.objects.all()

输出结果:<ExtendedQuerySet [<PeriodicTask: Importing contacts: every 10 seconds>, <PeriodicTask: my_task: every 10 seconds>, <PeriodicTask: celery.backend_cleanup: 0 4 * * * (m/h/d/dM/MY) Asia/Shanghai>]>

遍历周期性任务

for task in PeriodicTask.objects.all():
print(task.id)

1

2
3

通过id获取其中一个周期性任务

task1 = PeriodicTask.objects.get(id=1)

task1

输出结果: <PeriodicTask: Importing contacts: every 10 seconds>

PeriodicTask.objects.get(name=”my_task”)
输出结果: <PeriodicTask: my_task: every 10 seconds>

通过name获取其中的周期性任务

task2 = PeriodicTask.objects.get(name=”my_task”)

task2

输出结果: <PeriodicTask: my_task: every 10 seconds>
复制代码
周期性任务的删除
获取到了周期性任务之后,好奇的我尝试直接删除,发现直接死循环:

删除周期性的任务,千万不要这样做,会死循环

task1.delete()

输出结果: (1, {‘django_celery_beat.PeriodicTask’: 1})

task2.delete()

输出结果: (1, {‘django_celery_beat.PeriodicTask’: 1})
复制代码
如果要删除周期性任务,必须首先暂停任务,然后再删除,如下:

设置name为 my_taks1 的任务暂停执行

PeriodicTask.objects.get(name=”my_task1″).enabled = False
PeriodicTask.objects.get(name=”my_task1″).save()

删除该任务

PeriodicTask.objects.get(name=”my_task1″).delete()

输出结果: (1, {‘django_celery_beat.PeriodicTask’: 1})
复制代码
周期性任务暂停之后,重新启动 
当然会有暂停任务之后,重新开启任务的需求,如下:

设置任务的 enabled 为 True 即可:

PeriodicTask.objects.get(name=”my_task1″).enabled = True
PeriodicTask.objects.get(name=”my_task1″).save()
复制代码
以上就是小编今天为大家带来的内容 小编本身就是一名python开发工程师我自己花了三天时间整理了一套python学习教程从最基础的python脚本到web开发,爬虫,数据分析,数据可视化,机器学习,等,这些资料有想要的小伙伴 点击 即可领取

微信扫一扫,分享到朋友圈

全网最细之Celery 4.x动态添加定时任务

外媒分享技嘉RTX 3090 Turbo/3080 Vision/3070 Eagle Gaming OC产品图片

上一篇

预热视频汇总,暗示ColorOS 11更具个性?网友:期待

下一篇

你也可能喜欢

全网最细之Celery 4.x动态添加定时任务

长按储存图像,分享给朋友