Celery在python中的单独使用

时间:2019-10-15
本文章向大家介绍Celery在python中的单独使用,主要包括Celery在python中的单独使用使用实例、应用技巧、基本知识点总结和需要注意事项,具有一定的参考价值,需要的朋友可以参考一下。

简单使用:

  1.目录结构

    -app_task.py
    -worker.py
    -result.py

  2.在需要进行异步执行的文件app_task.py中导入celery,并实例化出一个对象,传入消息中间和数据存储配置参数

broker = 'redis://127.0.0.1:6379/1' # 使用redis第一个库
backend = 'redis://127.0.0.1:6379/2' # 使用redis第二个库
cel = celery.Celery('test',broker=broker,backend=backend)

  3.在需要进行异步执行的函数上添加装饰器

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

  4.新建文件worker.py用来添加任务

from app_task import add
result = add.delay(4,2)
print(result)  # 此结果不是add执行结果,而是放入消息队列中的一个任务id

  5.新建文件result.py用来接收结果

from celery.result import AsyncResult
from app_task import cel
# 2e76b24d-364f-46b7-a0eb-f69f663dfb0d
async1 = AsyncResult(id="56db7832-d0f4-4b9b-ba92-4fb08d0f9592", app=cel)  # id处放置消息队列中任务的id

if async1.successful():  # 判断执行任务成功
    result = async1.get()
    print(result)
    # result.forget() # 将结果删除
elif async1.failed():
    print('执行失败')
elif async1.status == 'PENDING':
    print('任务等待中被执行')
elif async1.status == 'RETRY':
    print('任务异常后正在重试')
elif async1.status == 'STARTED':
    print('任务已经开始被执行')

  6.添加任务+启动工人+查看结果

    1.运行worker.py
    2.终端中输入命令:celery worker -A app_task -l info -P eventlet # 其中app_task为配置参数的文件名
    3.运行result.py查看结果

 

多任务使用:

  1.目录结构,将简单使用中的app_task改为一个celery_task文件夹,其中放至少两个文件:celery.py(名字固定)/task1.py(方具体任务的文件)/根据需求放置其他任务文件

    -celery_task
      -celery.py # 配置参数
      -task1.py # 任务文件,根据需求添加
      -task2.py
      ...
    -worker.py # 添加任务
    -result.py # 查看结果

  2.在任务文件task1.py中导入cel,并以装饰器形式加在函数上

from .celery import cel
@cel.task
def mul(x,y):
    return x*y

  3.celery.py中配置参数

from celery import Celery
cel = Celery(
    'celery_demo',
    backend = 'redis://127.0.0.1:6379/1',
    broker = 'redis://127.0.0.1:6379/2',
    include=['celery_task.task1','celery_task.task2']
)
cel.conf.timezone = 'Asia/Shanghai'
cel.conf.enable_utc = False

  4.worker.py和简单使用中的相同

  5.result.py与简单使用中的相同

  6.添加任务+启动工人+查看结果
    1.运行worker.py
    2.终端中输入命令:celery worker -A celery_task -l info -P eventlet    # 其中celery_task为文件夹名
    3.运行result.py查看结果

延时任务:

  1.目录结构

    -celery_task
      -celery.py
      -task1.py
      -task2.py
      ...
    -worker.py
    -result.py

  2.task1.py与多任务相同

  3.celery.py与多任务相同

  4.worker.py中将传入的时间改为指定时间

ctime = datetime.now() # 当前时间
utc_time = datetime.utcfromtimestamp(ctime.timestamp()) # 转成本地时间
time_delta = timedelta(seconds=30) # 设置延时30s
task_time = utc_time+time_delta # 设定时间点为30s后
ret1 = add.apply_async(args=[2,3],eta=task_time)

  5.result.py与简单使用中的相同

  6.与多任务相同

定时任务:

  1.目录结构,去掉worker.py,添加任务变为在终端输入命令
    -celery.py
      -celery.py
      -task1.py
      -task2.py
      ...
    -result.py

  2.task1.py与多任务相同

  3.celery.py中与定时任务相比,添加下面参数
    1.设定时间间隔,循环执行

cel.conf.beat_schedule = {
    'add-every-10-seconds':{  # 名字随意起
        'task': 'celery_task.task1.mul',  # 指定要执行的任务
        'schedule': timedelta(seconds=2),  # 指定时间间隔
        'args': (2,3)  # 传入参数
    }
}

    2.指定年月日具体时间,循环执行

from celery.schedules import crontab
cel.conf.beat_schedule = {
    'add-every-10-seconds':{
        'task': 'celery_task.task1.mul',
        'schedule': crontab(minute=28,hour=2,day_of_month=23,month_of_year=6),
        'args': (6,3)
    }
}

  4.没有worker.py

  5.result.py与简单使用中的相同

  6.添加任务+启动工人+查看结果
    1.启动一个终端,输入:celery beat -A celery_task -l info # celery_task为文件夹名
    2.再启动一个终端,输入:celery worker -A celery_task -l info -P eventlet
    3.运行result.py查看结果

 

原文地址:https://www.cnblogs.com/gyk1030/p/11677164.html