Celery架构

Posted gaimo

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Celery架构相关的知识,希望对你有一定的参考价值。

cerely是什么?

  cerely被用来稍后执行某些代码,或者调度器调度这些代码。

Celery架构

  Celery的架构由三部分组成,消息中间件(message broker)、任务执行单元(worker)和 任务执行结果存储(backend - task result store)组成。

消息中间件

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

任务执行单元

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

任务结果存储

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

总结:

"""
1、celery框架自带socket,所以自身是一个独立运行的服务
2、启动celery服务,是来执行服务中的任务的,服务中带一个执行任务的对象,会执行准备就绪的任务,将执行任务的结果保存起来
3、celery框架由三部分组成:存放要执行的任务broker,执行任务的对象worker,存放任务结果的backend
4、安装的celery主体模块,默认只提供worker,要结合其他技术提供broker和backend(两个存储的单位)
"""

工作的基本原理

技术图片

 

1、准备配置了broker与backend的worker(任务的来源),并启动。

2、添加任务到broker,worker就会执行任务,将结果存储到backend中。

3、想查看任务的执行结果,根据任务的id去bckend中查询。

Celery的安装配置

pip install celery

消息中间件:RabbitMQ/Redis

app=celery.Celery(任务名, broker=xxx, backend=xxx, include=[xxx, xxx])

 

Celery执行异步任务

包架构封装

project
    ├── celery_task      # celery包
    │   ├── __init__.py # 包文件
    │   ├── celery.py   # celery连接和配置相关文件,且名字必须交celery.py
    │   └── tasks.py    # 所有任务函数
    ├── add_task.py      # 添加任务
    └── get_result.py   # 获取结果

 

基本使用

celery.py

# 1)创建app + 任务

# 2)启动celery(app)服务:
# 非windows
# 命令:celery worker -A celery_task -l info
# windows:
# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet

# 3)添加任务:手动添加,要自定义添加任务的脚本,右键执行脚本

# 4)获取结果:手动获取,要自定义获取任务的脚本,右键执行脚本


from celery import Celery

broker = redis://127.0.0.1:6379/1
backend = redis://127.0.0.1:6379/2
app = Celery(broker=broker, backend=backend, include=[celery_task.tasks])

 

tasks.py

from .celery import app
import time
@app.task
def add(n, m):
    print(n)
    print(m)
    time.sleep(10)
    print(n+m的结果:%s % (n + m))
    return n + m

@app.task
def low(n, m):
    print(n)
    print(m)
    print(n-m的结果:%s % (n - m))
    return n - m

 

add_task.py

from celery_task import tasks

# 添加立即执行任务
t1 = tasks.add.delay(10, 20)
t2 = tasks.low.delay(100, 50)
print(t1.id)


# 添加延迟任务
from datetime import datetime, timedelta
def eta_second(second):
    ctime = datetime.now()
    utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
    time_delay = timedelta(seconds=second)
    return utc_ctime + time_delay

tasks.low.apply_async(args=(200, 50), eta=eta_second(10))

 

get_result.py

from celery_task.celery import app

from celery.result import AsyncResult

id = 21325a40-9d32-44b5-a701-9a31cc3c74b5
if __name__ == __main__:
    async = AsyncResult(id=id, app=app)
    if async.successful():
        result = async.get()
        print(result)
    elif async.failed():
        print(任务失败)
    elif async.status == PENDING:
        print(任务等待中被执行)
    elif async.status == RETRY:
        print(任务异常后正在重试)
    elif async.status == STARTED:
        print(任务已经开始被执行)

 

 

高级使用

celery.py
# 1)创建app + 任务
?
# 2)启动celery(app)服务:
# 非windows
# 命令:celery worker -A celery_task -l info
# windows:
# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet
?
# 3)添加任务:自动添加任务,所以要启动一个添加任务的服务
# 命令:celery beat -A celery_task -l info
?
# 4)获取结果
?
?
from celery import Celery
?
broker = redis://127.0.0.1:6379/1
backend = redis://127.0.0.1:6379/2
app = Celery(broker=broker, backend=backend, 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 = {
    low-task: {
        task: celery_task.tasks.low,
        schedule: timedelta(seconds=3),
        # ‘schedule‘: crontab(hour=8, day_of_week=1),  # 每周一早八点
        args: (300, 150),
    }
}

 

tasks.py

from .celery import app
?
import time
@app.task
def add(n, m):
    print(n)
    print(m)
    time.sleep(10)
    print(n+m的结果:%s % (n + m))
    return n + m
?
?
@app.task
def low(n, m):
    print(n)
    print(m)
    print(n-m的结果:%s % (n - m))
    return n - m

get_result.py

from celery_task.celery import app
?
from celery.result import AsyncResult
?
id = 21325a40-9d32-44b5-a701-9a31cc3c74b5
if __name__ == __main__:
    async = AsyncResult(id=id, app=app)
    if async.successful():
        result = async.get()
        print(result)
    elif async.failed():
        print(任务失败)
    elif async.status == PENDING:
        print(任务等待中被执行)
    elif async.status == RETRY:
        print(任务异常后正在重试)
    elif async.status == STARTED:
        print(任务已经开始被执行)
 

django中使用

celery.py

# 重点:要将 项目名.settings 所占的文件夹添加到环境变量
# import sys
# sys.path.append(r‘项目绝对路径‘)
?
# 开启django支持
import os
os.environ.setdefault(DJANGO_SETTINGS_MODULE, 项目名.settings)
import django
django.setup()
?
?
?
# 1)创建app + 任务
?
# 2)启动celery(app)服务:
# 非windows
# 命令:celery worker -A celery_task -l info
# windows:
# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet
?
# 3)添加任务:自动添加任务,所以要启动一个添加任务的服务
# 命令:celery beat -A celery_task -l info
?
# 4)获取结果
?
?
from celery import Celery
?
broker = redis://127.0.0.1:6379/1
backend = redis://127.0.0.1:6379/2
app = Celery(broker=broker, backend=backend, 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 = {
    django-task: {
        task: celery_task.tasks.test_django_celery,
        schedule: timedelta(seconds=3),
        args: (),
    }
}

 

tasks.py

from .celery import app
?
from home.models import Banner
from settings.const import BANNER_COUNT  # 轮播图最大显示条数
from home.serializers import BannerModelSerializer
from django.core.cache import cache
@app.task
def update_banner_list():
    # 获取最新内容
    banner_query = Banner.objects.filter(is_delete=False, is_show=True).order_by(-orders)[:BANNER_COUNT]
    # 序列化
    banner_data = BannerModelSerializer(banner_query, many=True).data
    for banner in banner_data:
        banner[image] = http://127.0.0.1:8000 + banner[image]
    # 更新缓存
    cache.set(banner_list, banner_data)
    return True
 

 

 

 


 

 

 

 

 

 

以上是关于Celery架构的主要内容,如果未能解决你的问题,请参考以下文章

celery:celery介绍架构基本使用,celery执行异步任务延迟任务定时任务,django中使用celery。

Celery+Rabbitmq实现异步任务

celery简单使用

Celery

Celery第一步

celery的使用