Django实现异步定时任务

Posted 2021-12-04 dream-huang

Celery（芹菜）

 　　这个翻译之后居然叫芹菜~~~~好吧~~挺补的~~~最近Django框架需要涉及到执行定时周期任务~~之前想用线程方法的~发现好像不会尴尬~~~然后又想用“django_crontab”来执行的毕竟这个学习周期短~~但是不好用啊发现，最后还是决定花时间来学学~~芹菜。

　　Celery是Python语言实现的分布式队列服务，除了支持即时任务，还支持定时任务。

Celery中的五个核心角色

• Task

　　任务(Task)就是你要做的事情，例如一个注册流程里面有很多任务，给用户发验证邮件就是一个任务，这种耗时任务可以交给Celery去处理，还有一种任务是定时任务，比如每天定时统计网站的注册人数，这个也可以交给Celery周期性的处理。

• Broker

　　Broker 的中文意思是经纪人，指为市场上买卖双方提供中介服务的人。在Celery中它介于生产者和消费者之间经纪人，这个角色相当于数据结构中的队列。例如一个Web系统中，生产者是处理核心业务的Web程序，业务中可能会产生一些耗时的任务，比如短信，生产者会将任务发送给 Broker，就是把这个任务暂时放到队列中，等待消费者来处理。消费者是 Worker，是专门用于执行任务的后台服务。Worker 将实时监控队列中是否有新的任务，如果有就拿出来进行处理。Celery 本身不提供队列服务，一般用 Redis 或者 RabbitMQ 来扮演 Broker 的角色

• Worker

　　Worker 就是那个一直在后台执行任务的人，也称为任务的消费者，它会实时地监控队列中有没有任务，如果有就立即取出来执行。

• Beat

　　Beat 是一个定时任务调度器，它会根据配置定时将任务发送给 Broker，等待 Worker 来消费。

• Backend

　　Backend 用于保存任务的执行结果，每个任务都有返回值，比如发送邮件的服务会告诉我们有没有发送成功，这个结果就是存在Backend中，当然我们并不总是要关心任务的执行结果。

                                                 

 

 

 

 

 

 

 

基本使用

 

• 如果出现：AttributeError: ‘str‘ object has no attribute ‘items‘，则是redis版本过高（我安装的是3.x），安装2.10.6即可~（但是在我安装2.10之后又说我版本太低，但是当我重新装回3.x的时候又能用了~~~我只能说玄学好吧）

•  如果出现下面这个报错解决办法是，在实例化Celery时添加下面配置项，这个好像是新协议问题，使用指定到旧协议。

CELERY_TASK_PROTOCOL = 1  # Django中使用这个
app.conf.task_protocol = 1

哇~~~~想哭一波三折终于看到输出了（这一步就代表初步使用没问题了，接下来我终于可以做周期任务了）

获取返回结果 

#!/usr/bin/env python3
# -*- encoding:utf-8 -*-
"""
author:Egon
mail:[email protected]
"""
from celery import Celery

# 创建Celery实例
app = Celery(‘tasks‘, broker=‘redis://127.0.0.1:6379‘,
                      backend=‘redis://127.0.0.1:6379‘,
             )
app.conf.task_protocol = 1

# 创建任务
@app.task
def add(x):
    print("开始任务了:%s"%x)
    return x

task1.py

#!/usr/bin/env python3
# -*- encoding:utf-8 -*-
"""
author:Egon
mail:[email protected]
"""
import sys
dir = r"D:\\today\\celery_task1"
sys.path.append(dir)  # 我的任务文件不在环境变量里，IDLE找不到
from task1 import add
reslut = add.delay(1)
print(reslut, type(reslut))
# 查看返回值,a702b8c6-3ac0-48ab-a0de-f21b458ddab9 <class ‘celery.result.AsyncResult‘>

r = reslut.get()
print(r) # 1
# 会报错这里是实例化的时候，没有定义backend，就是保存任务结果的位置。

cmd.py

 

如果想获取返回结果直接用get方式会报错，需要指定保存任务结果的位置。

reslut.get()  # 会报错这里是实例化的时候，没有定义backend，就是保存任务结果的位置。

指定输出到，我这里还是指定到redis中。

报错

好吧我算是把各种报错都玩了个遍~~~~

• 在执行时又出现下面这个报错了，解决步骤：

pip install eventlet # 安装一下
celery -A task1 worker -l info -P eventlet # 将启动命令改成这个
再去执行cmd.py就没问题了~~

Traceback (most recent call last):
  File "d:\\programmingsoftware\\python35\\lib\\site-packages\\billiard\\pool.py", line 358, in workloop
    result = (True, prepare_result(fun(*args, **kwargs)))
  File "d:\\programmingsoftware\\python35\\lib\\site-packages\\celery\\app\\trace.py", line 525, in _fast_trace_task
    tasks, accept, hostname = _loc
ValueError: not enough values to unpack (expected 3, got 0)