Celery:小试牛刀

Celery是如何工作的?

Celery 由于 其分布式体系结构,在某种程度上可能难以理解。下图是典型Django-Celery设置的高级示意图(FROM O’REILLY):

当请求到达时,您可以在处理它时调用Celery任务。调用任务的函数会立即返回,而不会阻塞当前进程。实际上,任务尚未完成执行,但是任务消息已进入任务队列(或许多可能的任务队列之一)。

workers 是独立的进程,用于监视任务队列中是否有新任务并实际执行它们,他们拿起任务消息、处理任务、存储结果。

一、安装一个broker

Celery需要一个发送和接收消息的解决方案,即一个消息代理(message broker)服务,常用的broker包括:

RabbitMQ功能齐全,稳定,耐用且易于安装,是生产环境的绝佳选择。

Ubuntu安装:

$ sudo apt-get install rabbitmq-server

Docker安装:

$ docker run -d -p 5672:5672 rabbitmq

Redis

Redis也具有完整的功能,但是在突然终止或电源故障的情况下更容易丢失数据。

Ubuntu安装:

$ sudo apt install redis-server

Docker安装:

$ docker run -d -p 6379:6379 redis

二、安装Celery

$ pip install celery

三、编写Celery任务代码

首先导入Celery,创建一个Celery对象,这个对象将作为一个操作 Celery 的入口,如创建任务,管理workers等。

以下示例会把所有东西都写在一个模块中,但是对于大型项目,您需要创建一个专用模块

# tasks.py
import time
from celery import Celery

app = Celery('tasks', broker='pyamqp://guest@localhost//')

@app.task
def add(x, y):
    print('--------start---------')
    for i in range(5):
        print(f'第{i}秒')
    print('--------over----------')
    return x + y

第一个参数是当前模块的名称,这是唯一的必需参数。

第二个参数指定要使用的消息代理的URL。这里使用RabbitMQ(也是默认选项)。

若使用Redis:

app = Celery('tasks', broker='redis://localhost:6379/0')

四、启动 worker 进程

$ celery -A tasks worker --loglevel=INFO

在生产环境中,需要在后台将工作程序作为守护程序运行。为此,需要使用 平台提供的工具 或 类似supervisord的工具

五、调用任务

调用任务需要导入带有celery示例的模块,这里没有重新创建一个模块导入,而是使用命令行模式。要调用我们定义的任务,可以使用delay()(详情参阅 调用任务):

>>> from tasks import add
>>> add.delay(4, 4)

调用任务将返回一个AsyncResult实例,这可用于检查任务的状态,等待任务完成或获取其返回值(或者如果任务失败,则获取异常和回溯)

默认情况下执行任务不返回结果。为了执行远程过程调用或跟踪数据库中的任务结果,需要配置result backend

六、获取运行结果

如果要跟踪任务的状态,Celery需要将状态存储或发送到某个地方。有多个result backend可供选择:SQLAlchemy / Django ORM, MongoDBMemcachedRedisRPCRabbitMQ / AMQP)等。

下面使用 RPC 作为result backend,该后端将状态作为瞬态消息发送回去。使用backend参数配置Celery对象的result backend

app = Celery('tasks', backend='rpc://', broker='pyamqp://')

或者,如果使用 Redis 作为result backend,但仍然使用 RabbitMQ 作为 broker(流行的组合):

app = Celery('tasks', backend='redis://localhost', broker='pyamqp://')

更多result backend配置参阅“result backend

我们再次调用该任务:

>>> result = add.delay(4, 4)
>>> result.ready() # 检查是否完成任务,返回布尔值

通过设置@app.task(ignore_result=True)选项,也可以禁用单个任务返回结果,详情见有关celery.result的完整参考

七、配置Celery

对于大多数使用情况,默认配置就够了,但是可以配置更多选项使Celery根据需要工作。详细配置见“配置和默认值”

可以直接在应用程序上设置配置,也可以使用专用的配置模块设置配置。例如配置用于序列化任务负载的默认序列化器:

# 配置一个设置:
app.conf.task_serializer = 'json'

# 一次配置许多设置,则可以使用update
app.conf.update(
    task_serializer='json',
    accept_content=['json'],  # Ignore other content
    result_serializer='json',
    timezone='Europe/Oslo',
    enable_utc=True,
)

对于较大的项目,建议使用专用的配置模块。

app.config_from_object('celeryconfig')

celeryconfig.py必须可用于从当前目录或Python路径中加载

celeryconfig.py

broker_url = 'pyamqp://'
result_backend = 'rpc://'

task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = 'Europe/Oslo'
enable_utc = True

参考

给TA买糖
共{{data.count}}人
人已赞赏
经验教程

漫画 | 公司前端把后端告上县衙,还列了 5 宗罪!

2021-3-9 9:52:00

经验教程

趣谈 DHCP 协议,有点意思。

2021-3-9 10:39:00

⚠️
免责声明:根据《计算机软件保护条例》第十七条规定“为了学习和研究软件内含的设计思想和原理,通过安装、显示、传输或者存储软件等方式使用软件的,可以不经软件著作权人许可,不向其支付报酬。”您需知晓本站所有内容资源均来源于网络,仅供用户交流学习与研究使用,版权归属原版权方所有,版权争议与本站无关,用户本人下载后不能用作商业或非法用途,需在24个小时之内从您的电脑中彻底删除上述内容,否则后果均由用户承担责任;如果您访问和下载此文件,表示您同意只将此文件用于参考、学习而非其他用途,否则一切后果请您自行承担,如果您喜欢该程序,请支持正版软件,购买注册,得到更好的正版服务。 本站为个人博客非盈利性站点,所有软件信息均来自网络,所有资源仅供学习参考研究目的,并不贩卖软件,不存在任何商业目的及用途,网站会员捐赠是您喜欢本站而产生的赞助支持行为,仅为维持服务器的开支与维护,全凭自愿无任何强求。本站部份代码及教程来源于互联网,仅供网友学习交流,若您喜欢本文可附上原文链接随意转载。
无意侵害您的权益,请发送邮件至 momeis6@qq.com 或点击右侧 私信:momeis 反馈,我们将尽快处理。
0 条回复 A文章作者 M管理员
    暂无讨论,说说你的看法吧
个人中心
今日签到
有新私信 私信列表
搜索