celery异步,延时任务, 周期任务

Posted robertx

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了celery异步,延时任务, 周期任务相关的知识,希望对你有一定的参考价值。

  celery中文译为芹菜,是一个分布式任务队列. 是异步的,所以能处理大量消息

  最新的celery不支持windows下使用了,所以在使用pycharm安装celery模块之后,需要再安装eventlet模块才能测试运行.

一.异步任务

启动客户端:

s1,s2要在项目目录下,如果在文件夹中执行,terminal输入命令的时候要-A 项目文件夹的名字

c=Celery("task",broker="redis://127.0.0.1:6379/2",backend="redis://127.0.0.1:6379/1", include="项目名.文件夹")

 

Terminal中输入

celery worker -A s1 -l info -P eventlet

 

  给定两个文件

  s1.py

from celery import Celery
import time
c=Celery("task",broker="redis://127.0.0.1:6379/2",backend="redis://127.0.0.1:6379/1")

@c.task
def myfun1(a,b):
    return f"myfun1{a}{b}"

@c.task
def myfun2():
    return "myfun2"

@c.task
def myfun3():
    return "myfun3"

celery worker -A s1 -l info -P eventlet

  s2.py

from s1 import myfun1,myfun2,myfun3,c
from celery.result import AsyncResult
#多个生产者
# for i in range(10):
#     s=myfun1.delay()
#     print(s)

s=myfun1.delay(10,20)
print(s.id)
r=AsyncResult(id=s.id,app=c)

#获取状态
# print(r.status)
# print(r.successful())

#获取值
# print(r.get())

#只获取报错信息
print(r.get(propagate=False))

#获取具体出错的位置
# print(r.traceback)

 

二.延时任务/定时任务

  apply_async

t=add.apply_async((1,2),countdown=5) #表示延迟5秒钟执行任务
print(t)
print(t.get())

  支持的参数

countdown : 等待一段时间再执行.
add.apply_async((2,3), countdown=5)
eta : 定义任务的开始时间.这里的时间是UTC时间,这里有坑 add.apply_async((
2,3), eta=now+tiedelta(second=10))
expires : 设置超时时间. add.apply_async((
2,3), expires=60)
retry : 定时如果任务失败后, 是否重试. add.apply_async((
2,3), retry=False)
retry_policy : 重试策略.   max_retries : 最大重试次数, 默认为
3 次.   interval_start : 重试等待的时间间隔秒数, 默认为 0 , 表示直接重试不等待.   interval_step : 每次重试让重试间隔增加的秒数, 可以是数字或浮点数, 默认为 0.2   interval_max : 重试间隔最大的秒数, 即 通过 interval_step 增大到多少秒之后, 就不在增加了, 可以是数字或者浮点数, 默认为 0.2 .

  s1.py

from celery import Celery
import time
c=Celery("task",broker="redis://127.0.0.1:6379/2",backend="redis://127.0.0.1:6379/1")

@c.task
def myfun1(a,b):
    return f"myfun1{a}{b}"

@c.task
def myfun2():
    return "myfun2"

@c.task
def myfun3():
    return "myfun3"

  s2.py

from s1 import myfun1,myfun2,myfun3,c
from celery.result import AsyncResult
from datetime import timedelta

#指定多长时间以后执行
# s=myfun1.apply_async((10,20),countdown=5)

#第二种方式,使用utc时间
s=myfun1.apply_async((10,20),eta="utc时间")
print(s.id)
# 延时
# 重试

三.周期任务

  启动: 在Terminal中

celery beat -A s2 -l info

 

  s1.py

from celery import Celery
import time
c=Celery("task",broker="redis://127.0.0.1:6379/2",backend="redis://127.0.0.1:6379/1")

@c.task
def myfun1(a,b):
    return f"myfun1{a}{b}"

@c.task
def myfun2():
    return "myfun2"

@c.task
def myfun3():
    return "myfun3"

  s2.py

from s1 import c
from celery.beat import crontab
c.conf.beat_schedule = {
    "name": {
        "task": "s1.myfun1",
        "schedule": 3,
        "args": (10, 20)
    },
    "crontab": {
        "task": "s1.myfun1",
        "schedule": crontab(minute=44),
        "args": (10, 20)
    }
}

 配置详解:

 from celery.schedules import crontab

 CELERYBEAT_SCHEDULE = {
     # Executes every Monday morning at 7:30 A.M
     add-every-monday-morning: {
         task: tasks.add,
         schedule: crontab(hour=7, minute=30, day_of_week=1),
         args: (16, 16),
    },
 }

 

以上是关于celery异步,延时任务, 周期任务的主要内容,如果未能解决你的问题,请参考以下文章

celery

Celery时区设置问题源码探究

Celery - 一个懂得 异步任务 , 定时任务 , 周期任务 的芹菜

Celery - 一个懂得 异步任务 , 定时任务 , 周期任务 的芹菜

Celery异步任务队列/周期任务+ RabbitMQ + Django

Celery异步任务框架