导引

celery为异步任务框架，但在win平台会出现一些问题，但可以通过安装点补丁来解决，后期在linux上使用时不会出现

1 安装

pip install celery


2 celery的结构

3 celery基本使用

3.1 创建worker

from celery import Celery
broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery(__name__, broker=broker, backend=backend)
# 绑定方法
@app.task()
def add(x, y):
print(x + y)
return x + y


3.2 启动worker

win环境需要安装补丁
pip install eventlet
启动命令:celery --app=celery_test worker -P eventlet -l info


3.3 异步执行绑定方法

# 添加任务
from celery_test import add # 导入绑定方法
uid=add.delay(3,4) # 异步提交，返回任务id
# 获取结果
from celery_test import app
from celery.result import AsyncResult
id='f4e87056-4bb4-4133-92c7-0541f5687abd'
if __name__ == '__main__':
async=AsyncResult(id=id,app=app)
if async.successful():
result=async.get()
print(result)
elif async.failed():
print('error')
elif async.status=='PENDING':
print('waiting')
elif async.status=='RETRY':
print('trying')
elif async.status=='STARTED':
print('doing')


4 包结构划分异步任务

4.1 基本包结构

# celery.py
from celery import Celery
broker='redis://127.0.0.1:6379/1'
backend='redis://127.0.0.1:6379/2'
app=Celery(__name__,broker=broker,backend=backend,include=['celery_task.task1','celery_task.task2'])
# task1.py
from .celery import app
@app.task()
def add(x, y):
print(x, y)
return x + y
# task2.py
from .celery import app
@app.task()
def add2(x, y):
print(x, y)
return x * y


4.2 启动

celery --app=celery_task worker -P eventlet -l info


4.3 添加任务

from celery_task.task2 import add2
res=add2.delay(3,4) # 返回任务id


5 延迟添加任务

from datetime import datetime, timedelta
from celery_task.task2 import add2
eta = datetime.utcnow() + timedelta(secOnds=10) # 获取当前utc时间并加10秒
res = add2.apply_async(args=(10, 29), eta=eta) # 添加延迟任务，当时间为给定的eta时将任务提交
print(res)


6 定时任务

# celery.py
from celery import Celery
from datetime import timedelta
from celery.schedules import crontab
broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery(__name__, broker=broker, backend=backend, include=['celery_task.task1', 'celery_task.task2'])
app.conf.timezOne= 'Asia/Shanghai'
app.conf.enable_utc = False
# beat配置
app.conf.beat_schedule = {
'low_task': {
'task': 'celery_task.task1.add',
'schedule': timedelta(secOnds=3), # 每三秒提交一次
# 'schedule':crontab(hour=8,day_of_week=1), # 每周一早上八点提交
'args': (300, 100)
}
}


创建任务提交者beat

celery --app=celery_task beat -l info