首页 > 其他分享 >celery学习

celery学习

时间:2023-02-24 23:33:06浏览次数:44  
标签:学习 Celery task 默认 celery 任务 CELERY

 

原文:https://www.bbsmax.com/A/gVdnwBZXzW/

Celery对象

核心的对象就是Celery了,初始化方法:

  1. class Celery(object):
  2. def __init__(self, main=None, loader=None, backend=None,
  3. amqp=None, events=None, log=None, control=None,
  4. set_as_current=True, accept_magic_kwargs=False,
  5. tasks=None, broker=None, include=None, changes=None,
  6. config_source=None, fixups=None, task_cls=None,
  7. autofinalize=True, **kwargs):
  8. # 常用的需要配置的参数
  9. main:如果作为__main__运行,则为主模块的名称。用作自动生成的任务名称的前缀
  10. loader:当前加载器实例。
  11. backend:任务结果url;
  12. amqp:AMQP对象或类名,一般不管;
  13. log:日志对象或类名;
  14. set_as_current:将本实例设为全局当前应用
  15. tasks:任务注册表。
  16. broker:使用的默认代理的URL,任务队列;
  17. include:每个worker应该导入的模块列表,以实例创建的模块的目录作为起始路径;

这些参数都是celery实例化的配置,我们也可以不写,然后使用config_from_object方法加载配置;

创建异步任务的方法task

任何被task修饰的方法都会被创建一个Task对象,变成一个可序列化并发送到远程服务器的任务;它有多种修饰方式:

  • 使用默认的参数
  1. @celery.task
  2. def function_name():
  3. pass
  • 指定相关参数
  1. @celery.task(bind=True, name='name')
  2. def function_name():
  3. pass
  4. # task方法参数
  5. name:可以显式指定任务的名字;默认是模块的命名空间中本函数的名字。
  6. serializer:指定本任务的序列化的方法;
  7. bind:一个bool值,设置是否绑定一个task的实例,如果绑定,task实例会作为参数传递到任务方法中,可以访问task实例的所有的属性,即前面反序列化中那些属性
  8. base:定义任务的基类,可以以此来定义回调函数,默认是Task类,我们也可以定义自己的Task类
  9. default_retry_delay:设置该任务重试的延迟时间,当任务执行失败后,会自动重试,单位是秒,默认3分钟;
  10. autoretry_for:设置在特定异常时重试任务,默认False即不重试;
  11. retry_backoff:默认False,设置重试时的延迟时间间隔策略;
  12. retry_backoff_max:设置最大延迟重试时间,默认10分钟,如果失败则不再重试;
  13. retry_jitter:默认True,即引入抖动,避免重试任务集中执行;
  1. # 当bind=True时,add函数第一个参数是self,指的是task实例
  2. @task(bind=True) # 第一个参数是self,使用self.request访问相关的属性
  3. def add(self, x, y):
  4. try:
  5. logger.info(self.request.id)
  6. except:
  7. self.retry() # 当任务失败则进行重试
  • 自定义Task基类
  1. import celery
  2. class MyTask(celery.Task):
  3. # 任务失败时执行
  4. def on_failure(self, exc, task_id, args, kwargs, einfo):
  5. print('{0!r} failed: {1!r}'.format(task_id, exc))
  6. # 任务成功时执行
  7. def on_success(self, retval, task_id, args, kwargs):
  8. pass
  9. # 任务重试时执行
  10. def on_retry(self, exc, task_id, args, kwargs, einfo):
  11. pass
  12. @task(base=MyTask)
  13. def add(x, y):
  14. raise KeyError()
  15. #方法相关的参数
  16. exc:失败时的错误的类型;
  17. task_id:任务的id;
  18. args:任务函数的参数;
  19. kwargs:键值对参数;
  20. einfo:失败或重试时的异常详细信息;
  21. retval:任务成功执行的返回值;

Task的一般属性

  1. Task.name:任务名称;
  2. Task.request:当前任务的信息;
  3. Task.max_retries:设置重试的最大次数
  4. Task.throws:预期错误类的可选元组,不应被视为实际错误,而是结果失败;
  5. Task.rate_limit:设置此任务类型的速率限制
  6. Task.time_limit:此任务的硬限时(以秒为单位)。
  7. Task.ignore_result:不存储任务状态。默认False;
  8. Task.store_errors_even_if_ignored:如果True,即使任务配置为忽略结果,也会存储错误。
  9. Task.serializer:标识要使用的默认序列化方法的字符串。
  10. Task.compression:标识要使用的默认压缩方案的字符串。默认为task_compression设置。
  11. Task.backend:指定该任务的结果存储后端用于此任务。
  12. Task.acks_late:如果设置True为此任务的消息将在任务执行后确认 ,而不是在执行任务之前(默认行为),即默认任务执行之前就会发送确认;
  13. Task.track_started:如果True任务在工作人员执行任务时将其状态报告为“已启动”。默认是False;

调用异步任务

调用异步任务有三个方法,如下:

  1. task.delay():这是apply_async方法的别名,但接受的参数较为简单;
  2. task.apply_async(args=[arg1, arg2], kwargs={key:value, key:value}):可以接受复杂的参数
  3. send_task():可以发送未被注册的异步任务,即没有被celery.task装饰的任务;

1. app.send_task

  1. # tasks.py
  2. from celery import Celery
  3. app = Celery()
  4. def add(x,y):
  5. return x+y
  6. app.send_task('tasks.add',args=[3,4]) # 参数基本和apply_async函数一样
  7. # 但是send_task在发送的时候是不会检查tasks.add函数是否存在的,即使为空也会发送成功,所以celery执行是可能找不到该函数报错;

2. Task.delay

delay方法是apply_async方法的简化版,不支持执行选项,只能传递任务的参数。

  1. @app.task
  2. def add(x, y, z=0):
  3. return x + y
  4. add.delay(30,40,z=5) # 包括位置参数和关键字参数

3. Task.apply_async

apply_async支持执行选项,它会覆盖全局的默认参数和定义该任务时指定的执行选项,本质上还是调用了send_task方法;

  1. add.apply_async(args=[30,40], kwargs={'z':5})
  2. # 其他参数
  3. task_id:为任务分配唯一id,默认是uuid;
  4. countdown : 设置该任务等待一段时间再执行,单位为s;
  5. eta : 定义任务的开始时间;eta=time.time()+10;
  6. expires : 设置任务时间,任务在过期时间后还没有执行则被丢弃;
  7. retry : 如果任务失败后, 是否重试;使用true或false,默认为true
  8. shadow:重新指定任务的名字str,覆盖其在日志中使用的任务名称;
  9. retry_policy : {},重试策略.如下:
  10. max_retries : 最大重试次数, 默认为 3 次.
  11. interval_start : 重试等待的时间间隔秒数, 默认为 0 , 表示直接重试不等待.
  12. interval_step : 每次重试让重试间隔增加的秒数, 可以是数字或浮点数, 默认为 0.2
  13. interval_max : 重试间隔最大的秒数, 即 通过 interval_step 增大到多少秒之后, 就不在增加了, 可以是数字或者浮点数, 默认为 0.2 .
  14. routing_key:自定义路由键;
  15. queue:指定发送到哪个队列;
  16. exchange:指定发送到哪个交换机;
  17. priority:任务队列的优先级,0到255之间,对于rabbitmq来说0是最高优先级;
  18. serializer:任务序列化方法;通常不设置;
  19. compression:压缩方案,通常有zlib, bzip2
  20. headers:为任务添加额外的消息;
  21. link:任务成功执行后的回调方法;是一个signature对象;可以用作关联任务;
  22. link_error: 任务失败后的回调方法,是一个signature对象;
  23. # 如下
  24. add.apply_async((2, 2), retry=True, retry_policy={
  25. 'max_retries': 3,
  26. 'interval_start': 0,
  27. 'interval_step': 0.2,
  28. 'interval_max': 0.2,
  29. })
  • 自定义发布者,交换机,路由键, 队列, 优先级,序列方案和压缩方法:
  1. task.apply_async((2,2),
  2. compression='zlib',
  3. serialize='json',
  4. queue='priority.high',
  5. routing_key='web.add',
  6. priority=0,
  7. exchange='web_exchange')

获取任务结果和状态

由于celery发送的都是去其他进程执行的任务,如果需要在客户端监控任务的状态,有如下方法:

  1. r = task.apply_async()
  2. r.ready() # 查看任务状态,返回布尔值, 任务执行完成, 返回 True, 否则返回 False.
  3. r.wait() # 会阻塞等待任务完成, 返回任务执行结果,很少使用;
  4. r.get(timeout=1) # 获取任务执行结果,可以设置等待时间,如果超时但任务未完成返回None;
  5. r.result # 任务执行结果,未完成返回None;
  6. r.state # PENDING, START, SUCCESS,任务当前的状态
  7. r.status # PENDING, START, SUCCESS,任务当前的状态
  8. r.successful # 任务成功返回true
  9. r.traceback # 如果任务抛出了一个异常,可以获取原始的回溯信息

但是一般业务中很少用到,因为获取任务执行的结果需要阻塞,celery使用场景一般是不关心结果的。

使用celery

  1. # seting.py
  2. # 设置配置
  3. BROKER_URL = 'amqp://username:password@localhost:5672/yourvhost'
  4. CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
  5. CELERY_TASK_SERIALIZER = 'msgpack'
  6. CELERY_RESULT_SERIALIZER = 'msgpack'
  7. CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24
  8. CELERY_ACCEPT_CONTENT = ["msgpack"]
  9. CELERY_DEFAULT_QUEUE = "default"
  10. CELERY_QUEUES = {
  11. "default": { # 这是上面指定的默认队列
  12. "exchange": "default",
  13. "exchange_type": "direct",
  14. "routing_key": "default"
  15. }
  16. }
  17. # app.py --- 初始化celery对象
  18. from celery import Celery
  19. import seting
  20. from task import test_one, test_two
  21. celery = Celery(__name__, include=["task"]) # 设置需要导入的模块
  22. # 引入配置文件
  23. celery.config_from_object(seting)
  24. if __name__ == '__main__':
  25. test_one.apply_async((2,2),
  26. routing_key='default',
  27. priority=0,
  28. exchange='default')
  29. # task.py --- 定义需要执行的任务
  30. from app import celery
  31. @celery.task
  32. def test_one(x, y):
  33. return x + y
  34. @celery.task(name="one_name")
  35. def test_two(x, y):
  36. return x * y

小结

分析了celery任务一些方法参数和相关源码,接下来我们去研究celery更复杂的用法。

@_@||下面是官网

http://docs.jinkan.org/docs/celery/getting-started/index.html

你正在阅读 Celery 3.1 的文档。开发版本文档见: 此处.

Celery 初步

Celery 是一个“自带电池”的的任务队列。它易于使用,所以你可以无视其所解决问题的复杂程度而轻松入门。它遵照最佳实践设计,所以你的产品可以扩展,或与其他语言集成,并且它自带了在生产环境中运行这样一个系统所需的工具和支持。

在此教程中,你会了解使用 Celery 的最基础部分。包括:

  • 选择和安装消息传输方式(中间人)。
  • 安装 Celery 并创建第一个任务
  • 运行职程并调用任务。
  • 追踪任务在不同状态间的迁移,并检视返回值。

Celery 起初可能令人却步——但不要担心——此教程会快速带你入门。此教程刻意简化,所以不会让你在高级特性上困扰。在你完成了此教程的学习后, 阅读文档其余部分是个明智的选择,例如展示 Celery 能力的 Next Steps 教程。

选择中间人

Celery 需要一个发送和接收消息的解决方案,其通常以独立服务形式出现, 称为 消息中间人 。

可行的选择包括:

RabbitMQ

RabbitMQ 功能完备、稳定、耐用,并且安装简便,是生产环境的绝佳选择。 配合 Celery 使用 RabbitMQ 的详情见:

使用 RabbitMQ

如果你使用 Ubuntu 或 Debian,可以执行这条命令来安装 RabbitMQ:

$ sudo apt-get install rabbitmq-server

命令执行完成后,中间人就已经运行在后台,准备好传输消息: Starting rabbitmq-server: SUCCESS 。

如果你不使用 Ubuntu 或 Debian 也无须担心,你可以访问这个网站来寻找同样简单的其他平台上(包括 Microsoft Windows)的安装指南:

http://www.rabbitmq.com/download.html

Redis

Redis 也是功能完备的,但更易受突然中断或断电带来数据丢失的影响。使用 Redis 的详细信息见:

使用 Redis

使用数据库

不推荐把数据库用于消息队列,但对于很小的项目可能是合适的。你的选择包括:

例如如果你已经使用了 Django 的数据库后端,用它作为你的消息中间人在开发时会很方便,即使在生产环境中你会采用更稳健的系统。

其他中间人

除了上面列出的之外,还有其他的实验性传输实现可供选择,包括 Amazon SQS 、 Using MongoDB 和 IronMQ 。

完整列表见 中间人概览 。

安装 Celery

Celery 提交到了 Python Package Index(PyPI)上,所以你可以用标准的 Python 工具,诸如 pip 或 easy_install 来安装:

$ pip install celery

应用

首先你需要一个 Celery 实例,称为 Celery 应用或直接简称应用。既然这个实例用于你想在 Celery 中做一切事——比如创建任务、管理职程——的入口点,它必须可以被其他模块导入。

在此教程中,你的一切都容纳在单一模块里,对于更大的项目,你会想创建 独立模块 。

让我们创建 tasks.py :

from celery import Celery

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

@app.task
def add(x, y):
    return x + y

Celery 的第一个参数是当前模块的名称,这个参数是必须的,这样的话名称可以自动生成。第二个参数是中间人关键字参数,指定你所使用的消息中间人的 URL,此处使用了 RabbitMQ,也是默认的选项。更多可选的中间人见上面的 选择中间人 一节。例如,对于 RabbitMQ 你可以写 amqp://localhost ,而对于 Redis 你可以写 redis://localhost .

你定义了一个单一任务,称为 add ,返回两个数字的和。

运行 Celery 职程服务器

你现在可以用 worker 参数执行我们的程序:

$ celery -A tasks worker --loglevel=info

注解

如果职程没有启动,请查阅 故障处理 一节。

在盛传环境中你会想要让职程作为守护程序在后台运行。你需要用你所在平台提供的工具来实现,或是像 supervisord 这样的东西(更多信息见 Running the worker as a daemon)。

想要查看完整的命令行参数列表,如此:

$  celery worker --help

也有几个其他的命令,帮助也是可用的:

$ celery help

调用任务

你可以用 delay() 方法来调用任务。

这是 apply_async() 方法的快捷方式,该方法允许你更好地控制任务执行(见 Calling Tasks ):

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

这个任务已经由之前启动的职程执行,并且你可以查看职程的控制台输出来验证。

调用任务会返回一个 AsyncResult 实例,可用于检查任务的状态,等待任务完成或获取返回值(如果任务失败,则为异常和回溯)。 但这个功能默认是不开启的,你需要设置一个 Celery 的结果后端,下一节将会详细介绍。

保存结果

如果你想要保持追踪任务的状态,Celery 需要在某个地方存储或发送这些状态。可以从内建的几个结果后端选择:SQLAlchemy/Django ORM、 Memcached 、 Redis 、 AMQP( RabbitMQ )或 MongoDB , 或者你可以自制。

下例中你将会使用 amqp 结果后端来发送状态消息。后端通过 Celery 的 backend 参数来指定。如果你选择使用配置模块,则通过 CELERY_RESULT_BACKEND 选项来设置:

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

或者如果你想要把 Redis 用作结果后端,但仍然用 RabbitMQ 作为消息中间人 (常见的搭配):

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

更多关于结果后端的内容见 Result Backends 。

配置好结果后端后,让我们再次调用任务。这次你会得到调用任务后返回的 AsyncResult 实例:

>>> result = add.delay(4, 4)

ready() 方法查看任务是否完成处理:

>>> result.ready()
False

你可以等待任务完成,但这很少使用,因为它把异步调用变成了同步调用:

>>> result.get(timeout=1)
8

倘若任务抛出了一个异常, get() 会重新抛出异常, 但你可以指定 propagate 参数来覆盖这一行为:

>>> result.get(propagate=False)

如果任务抛出了一个异常,你也可以获取原始的回溯信息:

>>> result.traceback
…

完整的结果对象参考见 celery.result 。

配置

Celery,如同家用电器一般,并不需要太多的操作。它有一个输入和一个输出, 你必须把输入连接到中间人上,如果想则把输出连接到结果后端上。但如果你仔细观察后盖,有一个盖子露出许多滑块、转盘和按钮:这就是配置。

默认配置对大多数使用案例已经足够好了,但有许多事情需要微调来让 Celery 如你所愿地工作。阅读可用选项是熟悉可以配置什么的明智之举。你可以在 Configuration and defaults 参考中查阅这些选项。

配置可以直接在应用上设置,也可以使用一个独立的配置模块。

例如你可以通过修改 CELERY_TASK_SERIALIZER 选项来配置序列化任务载荷的默认的序列化方式:

app.conf.CELERY_TASK_SERIALIZER = 'json'

如果你一次性设置多个选项,你可以使用 update :

app.conf.update(
    CELERY_TASK_SERIALIZER='json',
    CELERY_ACCEPT_CONTENT=['json'],  # Ignore other content
    CELERY_RESULT_SERIALIZER='json',
    CELERY_TIMEZONE='Europe/Oslo',
    CELERY_ENABLE_UTC=True,
)

对于大型项目,采用独立配置模块更为有效,事实上你会为硬编码周期任务间隔和任务路由选项感到沮丧,因为中心化保存配置更合适。尤其是对于库而言,这使得用户控制任务行为成为可能,你也可以想象系统管理员在遇到系统故障时对配置做出简单修改。

你可以调用 config_from_object() 来让 Celery 实例加载配置模块:

app.config_from_object('celeryconfig')

配置模块通常称为 celeryconfig ,你也可以使用任意的模块名。

名为 celeryconfig.py 的模块必须可以从当前目录或 Python 路径加载,它可以是这样:

celeryconfig.py:

BROKER_URL = 'amqp://'
CELERY_RESULT_BACKEND = 'amqp://'

CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT=['json']
CELERY_TIMEZONE = 'Europe/Oslo'
CELERY_ENABLE_UTC = True

要验证你的配置文件可以正确工作,且不包含语法错误,你可以尝试导入它:

$ python -m celeryconfig

配置选项的完整参考见 Configuration and defaults 。

要证明配置文件的强大,比如这个例子展示了如何把“脏活”路由到专用的队列:

celeryconfig.py:

CELERY_ROUTES = {
    'tasks.add': 'low-priority',
}

或者,你可以限制任务的速率,这样每分钟只允许处理 10 个该类型的任务:

celeryconfig.py:

CELERY_ANNOTATIONS = {
    'tasks.add': {'rate_limit': '10/m'}
}

如果你使用 RabbitMQ 或 Redis 作为中间人,那么你也可以在运行时直接在职程上设置速率限制:

$ celery control rate_limit tasks.add 10/m
[email protected]: OK
    new rate limit set successfully

任务路由的详情见 Routing Tasks 。关于注解的更多见 CELERY_ANNOTATIONS 选项。关于远程控制命令和如何监视职程行为的更多见 Monitoring and Management Guide 。

何去何从

如果你想要进一步了解,你应该继续阅读 进阶 教程, 之后你可以学习 用户指南 。

故障处理

Frequently Asked Questions 章节也有一份故障处理提示。

职程无法启动:权限错误

  • 如果你使用 Debian、Ubuntu 或其他 Debian 系的发行版:

    Debian 最近把 /dev/shm/ 特殊文件重命名为 /run/shm 。

    简单的处置方式就是创建一个符号链接:

    # ln -s /run/shm /dev/shm
    
  • 其他:

    如果你提供了 --pidfile 、 --logfile 或 --statedb 参数中的任意一个,那么你必须确保它们指向了启动职程的那个用户可写可读的文件/目录。

结果后端没有奏效或任务总处于 PENDING (待处理)状态

所有任务默认都是 PENDING 的,所以状态会更好地命名为“未知”。 Celery 在任务发出时不更新任何状态,并且任何没有历史状态的任务被假定为待处理(毕竟你能获知任务 ID)。

  1. 确保任务没有启用 ignore_result 。

    启用这个选项会强制所有职程跳过状态更行。

  2. 确保没有启用 CELERY_IGNORE_RESULT 选项。

  3. 确保你没有仍在运行旧职程。

    偶然启动多个职程序是很容易的,所以确保之前的职程在你启动新的职程时已经恰当地关闭了。

    为配置所期望的结果后端的旧职程可能会一直运行并劫持任务。

    –pidfile 参数可以设置为一个绝对路径来避免该状况。

  4. 确保客户端配置了正确的结果后端。

    如果因为某些原因,客户端被配置使用了与职程不同的结果后端,那么你讲收不到结果,所以请确保检视后端是否正确:

    >>> result = task.delay(…)
    >>> print(result.backend)

标签:学习,Celery,task,默认,celery,任务,CELERY
From: https://www.cnblogs.com/machangwei-8/p/17153533.html

相关文章

  • 【新手入门】JDBC+Servlet+jsp【新闻系统的增删改查】:持续学习,持续优化,争取做一个真正
    新闻的增删改查  学习了一些博客知识【结尾会贴上,大家可以参考】在借鉴的基础上又进行了代码的优化,从而实现了CRUD,下面分享我的代码过程:包结构,是为了方便管理,所以我......
  • 「AcWing学习记录」KMP
    AcWing831.KMP字符串原题链接1.暴力算法怎么做chars[N],p[M];for(inti=1;i+m-1<=n;i++){boolflag=true;for(intj=1;j<=m;j++)......
  • 「matlab学习笔记」数据分析与多项式计算
    中国大学MOOC科学计算与MATLAB语言(点击此处跳转)MATLAB官方文档(点击此处跳转)5.1数据统计分析常用统计函数函数解释max()求向量或矩阵的最大元素min()求......
  • 297个机器学习彩图知识点(13)
    导读本系列将持续更新20个机器学习的知识点,欢迎关注。1.阈值类别特征的方差2.切分文本3.领域链接4.TSS5.误差6.训练错误率7.真正类率8.丢弃概率......
  • 【C语言经典算法100道实战题】学习资料大全
    ​【C语言经典算法100道实战题】适合具备C语言基础语法的同学学习,提高编写程序的逻辑思维能力和算法设计能力专门精心设计。100个经典的算法供大家练习及配套对应的录播视......
  • ArrayList学习笔记
    目录1、继承关系1.1、Serializable标记性接口1.2、Cloneable标记性接口1.3、RandomAccess标记性接口2、属性3、构造方法3.1、无参构造方法-ArrayList()3.2、有参构造方......
  • python学习——【第十三弹】
    前言上一篇文章​​python学习——【第十二弹】​​中学习了python中的深拷贝和浅拷贝,这篇文章接着学习python中的模块。什么是模块在python中,一个文件(以“.py”为后缀名的......
  • 机器学习(Tensowflow) 环境搭建(Anacoda的lab 使用)
    说明:本篇文章用于tensorflow学习,学习地址在bilibi,我会将自己做的笔记写成博客,最后会写成目录的,放上链接,方便大家阅读.1.创建的名字随便取一个就行2.确保有Tens......
  • monodepth2学习
    monodepth2学习相关网站​​https://www.cnblogs.com/blackworld-sp/category/2205853.html​​​​https://blog.csdn.net/avideointerfaces/article/details/105925104​......
  • 学习进度
    今天自学了4小时计算机网络,对计算机网络的所有课后习题都做了一遍,对此,我深有所得,通过对习题的学习,我更加清楚地掌握了计算机网络的知识,以及大的框架,这门学科很费时间,它的题......