Celery异步任务框架(简单使用)

2024-03-06 13:19:43

一：简介

1.Celery是什么

Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统专注于实时处理的异步任务队列同时也支持任务调度

Celery 官网：http://www.celeryproject.org/

Celery 官方文档英文版：http://docs.celeryproject.org/en/latest/index.html

Celery 官方文档中文版：http://docs.jinkan.org/docs/celery/

PS：Celery在Windows系统上会出现不兼容的情况

2.Celery异步任务框架

1.可以不依赖任何服务器，通过自身命令，启动服务(内部支持socket) 2.celery服务为为其他项目服务提供异步解决任务需求的注：会有两个服务同时运行，一个是项目服务，一个是celery服务，项目服务将需要异步处理的任务交给celery服务，celery就会在需要时异步完成项目的需求

人是一个独立运行的服务 | 医院也是一个独立运行的服务正常情况下，人可以完成所有健康情况的动作，不需要医院的参与；但当人生病时，就会被医院接收，解决人生病问题人生病的处理方案交给医院来解决，所有人不生病时，医院独立运行，人生病时，医院就来解决人生病的需求

3.Celery架构

Celery的架构由三部分组成，消息中间件（message broker）、任务执行单元（worker）和任务执行结果存储（task result store）组成

消息中间件

Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成。包括，RabbitMQ, Redis等等

任务执行单元

Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中。

任务结果存储

Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, redis等

使用场景

异步执行：解决耗时任务,将耗时操作任务提交给Celery去异步执行，比如发送短信/邮件、消息推送、音视频处理等等延迟执行：解决延迟任务

定时执行：解决周期(周期)任务,比如每天数据统计

二：Celery的安装和使用

pip install celery

消息中间件：RabbitMQ/Redis

app=Celery(‘任务名’, broker=’xxx’, backend=’xxx’)

两种celery任务结构：提倡用包管理，结构清晰

# 如果 Celery对象:Celery(...) 是放在一个模块下的 # 1）终端切换到该模块所在文件夹位置：scripts # 2）执行启动worker的命令：celery worker -A 模块名 -l info -P eventlet # 注：windows系统需要eventlet支持，Linux与MacOS直接执行：celery worker -A 模块名 -l info # 注：模块名随意

# 如果 Celery对象:Celery(...) 是放在一个包下的 # 1）必须在这个包下建一个celery.py的文件，将Celery(...)产生对象的语句放在该文件中 # 2）执行启动worker的命令：celery worker -A 包名 -l info -P eventlet # 注：windows系统需要eventlet支持，Linux与MacOS直接执行：celery worker -A 模块名 -l info # 注：包名随意

windows上启动celery正常，执行任务时报错

Traceback (most recent call last):
  File "c:\users\circle\appdata\local\programs\python\python37-32\lib\site-packages\billiard\pool.py", line 358, in workloop
    result = (True, prepare_result(fun(*args, **kwargs)))
  File "c:\users\circle\appdata\local\programs\python\python37-32\lib\site-packages\celery\app\trace.py", line 546, in _fast_trace_task
    tasks, accept, hostname = _loc
ValueError: not enough values to unpack (expected 3, got 0)

解决办法

安装 pip install eventlet
重新输入如下名命令
# celery_task是包名,包下必须要有一个叫celery的文件
celery worker -A celery_task -l info -P eventlet

方式一：在一个文件夹内的三个页面

worker执行页面

import celery
# broker存储的位置

broker = 'redis://127.0.0.1:6379/1'

# backend存储的位置

backend ='redis://127.0.0.1:6379/2'
# 实例化的celery对象

app=celery.Celery(name,broker=broker,backend=backend)
# 需要添加的任务

@app.task

defadd(x,y):

print(x*y)

return x+y

broker提交任务的页面

from celery_test import add
# 执行这个文件，就是把这个任务添加到数据库中，只要worker在工作
# 就会把这个任务从数据库1中拿出来执行，并把结果放到数据库2中
ret = add.delay(3,4)
# ret 是这个任务的uuid，用于获取任务结果

backend获取任务结果的页面

from celery_test import app
from celery.result import AsyncResult

# 任务对象的唯一标识：uuid

id = '19dc2faa-39f9-47b6-af77-e9d3a4d05d2e'

if name == 'main':

async1 = AsyncResult(id=id, app=app)

if async1.successful():

result = async1.get()

print(result)

elif async1.failed():

print('任务失败')

elif async1.status == 'PENDING':

print('任务等待中被执行')

elif async1.status == 'RETRY':

print('任务异常后正在重试')

elif async1.status == 'STARTED':

print('任务已经开始被执行')

方式二：worker单独做一个项目文件，添加任务和获取结果分离出来（执行异步任务）

创建一个celery项目（包），内部必须含有名字为celery的py文件，在内部创建celery对象

celery.py

import celery
broker = 'redis://127.0.0.1:6379/1'

backend ='redis://127.0.0.1:6379/2'
app=celery.Celery(name,broker=broker,backend=backend,include=['celery_task.task1','celery_task.task2'])

编写需要添加的任务也在这个包内，可创建不同的任务文件，可添加多个

task1.py

from .celery import app
@app.task

defadd(x,y):

print(x,y)

return x+y

执行延迟任务

添加任务页面

# 执行延迟任务就是多个一个时间参数
# 这里注意，时间参数是根据utc时间，并不是中国时间
from datetime import datetime, timedelta

# 时间对象必须和时间对象相加

eta=datetime.utcnow() + timedelta(seconds=10)

add.apply_async(args=(200, 50), eta=eta)

添加定时任务

celery页面

# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
# 任务的定时配置

from datetime import timedelta

from celery.schedules import crontab

app.conf.beat_schedule = {

'low-task': {

'task': 'celery_task.tasks.low',

'schedule': timedelta(seconds=3),

# 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点

'args': (300, 150),

}

}
# 定时任务的添加必须要新启动一个beat命令去工作

# celery beat -A celery_task -l info

三：Django中使用Celery

1.celery.py

# 加载django配置环境
import os
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffyapi.settings.dev')
# 实例化Celery，获取worker对象，include添加可处理的任务函数
from celery import Celery
broker = 'redis://127.0.0.1:6379/0'  # 数据存储位置
backend = 'redis://127.0.0.1:6379/1' # 返回值存储位置
app = Celery(__name__, broker=broker, backend=None, include=['celery_task.tasks'])
# 时区

app.conf.timezone = 'Asia/shanghai'

# 是否使用utc时间

app.conf.enable_utc = False
# 任务的定时配置

from datetime import timedelta

from celery.schedules import crontab
app.conf.beat_schedule = {

'update-banner-list': {

'task': 'celery_task.tasks.update_banner_list',

'schedule': timedelta(seconds=5),

'args': (),

}

}

2.task.py

from .celery import app
from django.core.cache import cache
from django.conf import settings
from home import serializer, models
@app.task

defupdate_banner_list():

queryset = models.Banner.objects.filter(is_delete=False, is_show=True).order_by('-display_order')[:settings.BANNER_COUNT]

ser = serializer.BannerModelSerializer(instance=queryset, many=True)

banner_list = ser.data

for banner in banner_list:

banner['banner_url'] = 'http://127.0.0.1:8000%s' % banner['banner_url']

cache.set('banner_list', banner_list, 606024)

return True

码农公寓