首页 > 其他分享 >celery学习md笔记:从0基础到系统性掌握用法 第(2)篇:celery的配置

celery学习md笔记:从0基础到系统性掌握用法 第(2)篇:celery的配置

时间:2023-11-01 19:04:42浏览次数:42  
标签:md task app 用法 celery 任务 proj my

Celery是一个功能完备即插即用的任务队列。它使得我们不需要考虑复杂的问题,使用非常简单。celery看起来似乎很庞大,本文我们先对其进行简单的了解,然后再去学习其他一些高级特性。

完整版笔记直接地址: 请移步这里


共 4 章,12 子模块,总计 5628 字


本章节我们需要快速了解celery一些概念,配置,使用.Celery使用简单,配置也非常简单。Celery有很多配置选项能够使得celery能够符合我们的需要,但是默认的几项配置已经足够应付大多数应用场景了。

配置信息可以直接在app中设置,或者通过专有的配置模块来配置。

1.直接通过app来配置


​ from celery import Celery ​ app = Celery('demo')

# 增加配置




app.conf.update(
    result_backend='redis://:[email protected]:6379/2',
    broker_url='redis://:[email protected]:6379/1',
)

2.专有配置文件

对于比较大的项目,我们建议配置信息作为一个单独的模块。我们可以通过调用app的函数来告诉Celery使用我们的配置模块。

配置模块的名字我们取名为celeryconfig, 这个名字不是固定的,我们可以任意取名,建议这么做。我们必须保证配置模块能够被导入。 配置模块的名字我们取名为celeryconfig, 这个名字不是固定的,我们可以任意取名,建议这么做。我们必须保证配置模块能够被导入。

下面我们在tasks.py模块 同级目录下创建配置模块celeryconfig.py:


​ result_backend = 'redis://:[email protected]:6379/2' ​ broker_url = 'redis://:[email protected]:6379/1'

tasks.py文件修改为:


​ from celery import Celery ​ import celeryconfig

# 我们这里案例使用redis作为broker




app = Celery('demo')





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




app.config_from_object('celeryconfig')

更多配置: http://docs.celeryproject.org/en/latest/userguide/configuration.html#configuration

介绍一下如何调用任务,队列路由.

我的项目目录:

TestCelery/ ├── proj │ ├── celeryconfig.py │ ├── celery.py │ ├── init.py │ └── tasks.py └── test.py

celery.py内容如下:


​ from celery import Celery

# 创建celery实例




app = Celery('demo')
app.config_from_object('proj.celeryconfig')





# 自动搜索任务




app.autodiscover_tasks(['proj'])

celeryconfig.p模块内容如下:


​ from kombu import Exchange, Queue ​ BROKER_URL = 'redis://:[email protected]:6379/1' ​ CELERY_RESULT_BACKEND = 'redis://:[email protected]:6379/2'

tasks.py模块内容如下:


​ from proj.celery import app as celery_app

# 创建任务函数




@celery_app.task
def my_task1():
    print("任务函数(my_task1)正在执行....")

@celery_app.task
def my_task2():
    print("任务函数(my_task2)正在执行....")

@celery_app.task
def my_task3():
    print("任务函数(my_task3)正在执行....")

启动worker:


​ celery -A proj worker -l info

键入ctrl+c可关闭worker.

调用任务,可使用delay()方法:


​ my_task.delay(2, 2)

也可以使用apply_async()方法,该方法可让我们设置一些任务执行的参数,例如,任务多久之后才执行,任务被发送到那个队列中等等.


​ my_task.apply_async((2, 2), queue='my_queue', countdown=10)

任务my_task将会被发送到my_queue队列中,并且在发送10秒之后执行。

如果我们直接执行任务函数,将会直接执行此函数在当前进程中,并不会向broker发送任何消息。

无论是delay()还是apply_async()方式都会返回AsyncResult对象,方便跟踪任务执行状态,但需要我们配置result_backend.

每一个被吊用的任务都会被分配一个ID,我们叫Task ID.

1. signature

我们到目前为止只是学习了如何使用delay()方法,当然这个方法也是非常常用的。但是有时我们并不想简单的将任务发送到队列中,我们想将一个任务函数(由参数和执行选项组成)作为一个参数传递给另外一个函数中,为了实现此目标,Celery使用一种叫做signatures的东西。

一个signature包装了一个参数和执行选项的单个任务调用。我们可将这个signature传递给函数。

我们先看下tasks.py模块中定义的任务函数:


​ from proj.celery import app as celery_app

# 创建任务函数




@celery_app.task
def my_task1():
    print("任务函数(my_task1)正在执行....")

@celery_app.task
def my_task2():
    print("任务函数(my_task2)正在执行....")

@celery_app.task
def my_task3():
    print("任务函数(my_task3)正在执行....")

我们将my_task1()任务包装称一个signature:


​ t1 = my_task1.signatures(countdown=10) ​ t1.delay()

2. Primitives

这些primitives本身就是signature对象,因此它们可以以多种方式组合成复杂的工作流程。primitives如下:

group: 一组任务并行执行,返回一组返回值,并可以按顺序检索返回值。

chain: 任务一个一个执行,一个执行完将执行return结果传递给下一个任务函数.

tasks.py模块如下:


​ from proj.celery import app as celery_app

# 创建任务函数




@celery_app.task
def my_task1(a, b):
    print("任务函数(my_task1)正在执行....")
    return a + b

@celery_app.task
def my_task2(a, b):
    print("任务函数(my_task2)正在执行....")
    return a + b

@celery_app.task
def my_task3(a, b):
    print("任务函数(my_task3)正在执行....")
    return a + b

group案例如下(test.py模块):


​ from proj.tasks import my_task1 ​ from proj.tasks import my_task2 ​ from proj.tasks import my_task3 ​ from celery import group

# 将多个signature放入同一组中




my_group = group((my_task1.s(10, 10), my_task2.s(20, 20), my_task3.s(30, 30)))
ret = my_group() # 执行组任务
print(ret.get())  # 输出每个任务结果

chain案例如下(test.py模块):


​ from proj.tasks import my_task1 ​ from proj.tasks import my_task2 ​ from proj.tasks import my_task3 ​ from celery import chain

# 将多个signature组成一个任务链






# my_task1的运行结果将会传递给my_task2






# my_task2的运行结果会传递给my_task3




my_chain = chain(my_task1.s(10, 10) | my_task2.s(20) | my_task3.s(30))
ret = my_chain()  # 执行任务链
print(ret.get())  # 输出最终结果

假如我们有两个worker,一个worker专门用来处理邮件发送任务和图像处理任务,一个worker专门用来处理文件上传任务。

我们创建两个队列,一个专门用于存储邮件任务队列和图像处理,一个用来存储文件上传任务队列。

Celery支持AMQP(Advanced Message Queue)所有的路由功能,我们也可以使用简单的路由设置将指定的任务发送到指定的队列中.

我们需要配置在celeryconfig.py模块中配置 CELERY_ROUTES 项, tasks.py模块修改如下:


​ from proj.celery import app as celery_app


​ @celery_app.task ​ def my_task1(a, b): ​ print("my_task1任务正在执行....") ​ return a + b


​ @celery_app.task ​ def my_task2(a, b): ​ print("my_task2任务正在执行....") ​ return a + b


​ @celery_app.task ​ def my_task3(a, b): ​ print("my_task3任务正在执行....") ​ return a + b


​ @celery_app.task ​ def my_task4(a, b): ​ print("my_task3任务正在执行....") ​ return a + b


​ @celery_app.task ​ def my_task5(): ​ print("my_task5任务正在执行....")


​ @celery_app.task ​ def my_task6(): ​ print("my_task6任务正在执行....")


​ @celery_app.task ​ def my_task7(): ​ print("my_task7任务正在执行....")

我们通过配置,将send_email和upload_file任务发送到queue1队列中,将image_process发送到queue2队列中。

我们修改celeryconfig.py:


​ broker_url='redis://:@127.0.0.1:6379/1' ​ result_backend='redis://:@127.0.0.1:6379/2'


​ task_routes=({ ​ 'proj.tasks.my_task5': {'queue': 'queue1'}, ​ 'proj.tasks.my_task6': {'queue': 'queue1'}, ​ 'proj.tasks.my_task7': {'queue': 'queue2'}, ​ }, ​ )

test.py:


​ from proj.tasks import *

# 发送任务到路由指定的队列中

my_task5.delay() my_task6.delay() my_task7.delay()


​   开启两个worker服务器,分别处理两个队列: ​ ```python ​ celery -A proj worker --loglevel=info -Q queue1 ​ celery -A proj worker --loglevel=info -Q queue2

我们同样也可以通过apply_aynsc()方法来设置任务发送到那个队列中:


​ my_task1.apply_async(queue='queue1')

我们也可设置一个worker服务器处理两个队列中的任务:


​ celery -A proj worker --loglevel=info -Q queue1,queue2

未完待续 下一期下一章

完整笔记请看文章开头

标签:md,task,app,用法,celery,任务,proj,my
From: https://blog.51cto.com/u_13578013/8131107

相关文章

  • 初学者需要注意的break 与continue的用法
    break1用法:break被执行时,会跳出所处的循环或这switch;2注意break需要在switch-case里面使用;如下例如果case的后面没有break,那么如上例case1的情况下它会一顺溜的走下来,而如果有break,那么到该case就会跳出switch,不会执行后面的case了。正确的写法如下:continue1用法:continue被执行时,跳......
  • CEEMDAN+PE自适应噪声完备集合经验模态分解+排列熵重构分量 程序语言为matlab
    ✅作者简介:热爱科研的Matlab仿真开发者,修心和技术同步精进,代码获取、论文复现及科研仿真合作可私信。......
  • Python eval的用法及注意事项
    eval是Python的一个内置函数,这个函数的作用是,返回传入字符串的表达式的结果。想象一下变量赋值时,将等号右边的表达式写成字符串的格式,将这个字符串作为eval的参数,eval的返回值就是这个表达式的结果。python中eval函数的用法十分的灵活,但也十分危险,安全性是其最大的缺点。本文从灵活......
  • 【MMD】MMD三渲二的终极解决方案msToonCoordinator丨核心概念丨赛璐璐丨技法丨资料
    本篇最开始放在B站的专栏,感觉B站的专栏越来越不方便,想增添补修都非常麻烦(B站专栏只允许修改三次),所以逐步搬迁到博客园。感谢MMD、MME的所有的制作者。使用msToonCoordinator渲染 前言一、这是经验分享,不是专业性质的文章。二、不要在评论区刷U3D、UE、Blender、C4D更强之类......
  • DecimalFormat的基本用法
    一、舍入模式RoundingMode1、舍入模式RoundingMode为截断Downpublicstaticvoidmain(String[]args){doublenum=11.256;DecimalFormatdf=newDecimalFormat("#.##");//格式化为两位小数//指定RoundingMode舍入模式df.setRoundi......
  • [win]CMD命令
    rmdir/sdel*.txtdel* mkdir 一个是删除文件一个是删除目录  什么是runtime:AsperWikipedia: runtimelibrary/run-timesystem.Incomputerprogramming,aruntimelibraryisaspecialprogramlibraryusedbyacompiler,toimplementfunctionsbuiltintoaprogr......
  • 2023.10.31 USACO 2020 选做.md
    P6009Non-DecreasingSubsequencesP由于值域很小,dp的转移不难想到写成矩阵的形式。考虑维护矩阵的前缀积和逆前缀积。然而单次的矩阵乘已经达到\(O(k^3)\)超时了,但是我们发现其实矩阵非\(0\)的位置是\(O(k)\)个的,所以复杂度降到了\(O(k^2)\).关于逆矩阵,我们无需高斯......
  • 无涯教程-Docker - CMD命令
    Docker有许多指令命令。这些是放置在DockerFile中的命令。CMD指令该命令用于在执行容器时在运行时执行命令。CMDcommandparam1command -这是启动容器时要运行的命令。param1    - 这是输入到命令的参数。该命令将相应执行。在我们的示例中,我们将输入一......
  • 【Django-DRF】多年积累md笔记 0基础到高手. 第(3)篇:使用Django开发REST 接口
    本文从分析现在流行的前后端分离Web应用模式说起,然后介绍如何设计RESTAPI,通过使用Django来实现一个RESTAPI为例,明确后端开发RESTAPI要做的最核心工作,然后介绍DjangoRESTframework能帮助我们简化开发RESTAPI的工作。完整版笔记直接地址:请移步这里共5章,24子模块,总计1.7......
  • Markdown使用心得(简单用法解析)
    Markdown使用心得(简单用法解析)Markdown的优势个人看来,MD的优势在于脱离对鼠标的依赖,在简单的熟悉后,从段落格式到字体特效的实现都可以完全脱离鼠标。避免了为了格式和艺术效果多次将右手在键盘和鼠标来回移动,打破码字的纯净体验,避免频繁的中断思路。简单的用法归纳接下来,我将......