Celery 异步任务

时间：2023-12-23 20:35:07

Celery

介绍:

  • Celery 是芹菜

    Celery 是基于Python实现的模块, 用于执行异步定时周期任务的

  • 其结构的组成是由

    1. 用户任务 app
    2. 管道 broker 用于存储任务 官方推荐 redis rabbitMQ / backend 用于存储任务执行结果的
    3. 员工 worker

  • 工作流程: (多任务 异步任务 )(生产者消费者模型)

    app -> task -> 调度器(broker) -> worker -> 调度器(backend) -> task -> app

  • 定时任务

    task -> 0:05分执行--调度器(broker) -> 0:05分执行--worker:等待 -> 调度器(backend) -> task

  • 周期任务

    task -> 每天0:05分执行--调度器(broker) -> 每天0:05分执行--worker: 每天等待 -> 调度器(backend) -> task

    task -> 每60秒执行--调度器(broker) -> 每60秒执行--worker: 每60秒等待 -> 调度器(backend) -> task

问题: window 下会用jing'chen

window 下运行需要安装 eventlet 包

异步任务

  • s1.py
import time

from celery import Celery

diaoduqi = Celery("mytask",broker="redis://127.0.0.1:6379",backend="redis://127.0.0.1:6379")

@diaoduqi.task

def ab(a,b):

    time.sleep(15)

    return a+b
  • s2.py (发布任务)
from 异步任务.s1 import ab

l = []

for i in range(5):

    res = ab.delay(i+1,i*i)

    l.append(res.id)

print(l)
  • s3.py (取结果)
from celery.result import AsyncResult

from 异步任务.s1 import diaoduqi

task_id = ['abd700c5-990a-496d-9a2b-28461518e8a0', '5fff40e6-a2d8-48c8-923f-8c2fe988ca77', '9bfca50e-51f3-46a6-baf6-e4e786e0815d', '3d0aa849-bc51-4e74-a150-02a170bb8540', '8f806ae2-9f2f-4404-8c31-a7eeb86ced42']

for i in task_id :

    a = AsyncResult(i,app=diaoduqi)

    if a.successful():

        print(a.get())

    else:

        print("任务还在执行中")

  • 开启命令:

    celery worker -A s1 -l INFO -P eventlet -c 10
    
# -P eventlet  指定使用其他模块 线程执行
    
# -c 10 指定workon 的数量

项目中使用

  • 目录结构
    • CeleryTask
      • celery.py
      • TaskOne.py
      • TaskTwo.py
    • 发布和获取的 py文件

  • celery.py

    from celery import Celery

DDQ = Celery("DDQ", broker='redis://127.0.0.1:6379', backend='redis://127.0.0.1:6379'
    
             , include=['CeleryTask.TaskOne', 'CeleryTask.TaskTwo'])

  • TaskOne.py

    from CeleryTask.celery import DDQ

import time

@DDQ.task
    
def one1():
    
    time.sleep(1)
    
    return 'one1'

@DDQ.task
    
def one2():
    
    time.sleep(2)
    
    return 'one2'

  • TaskTwo.py

    from CeleryTask.celery import DDQ
    
import time

@DDQ.task
    
def two1():
    
    time.sleep(3)
    
    return 'two1'

@DDQ.task
    
def two2():
    
    time.sleep(4)
    
    return 'two2'

  • 发布 获取

    from CeleryTask.TaskOne import one1, one2
    
from CeleryTask.TaskTwo import two1, two2

one1.delay()
    
one2.delay()
    
two1.delay()
    
two2.delay()
    
# one1.delay()
    
from CeleryTask.celery import DDQ
    
from celery.result import AsyncResult

task_list = []
    
for i in range(1, 50):
    
    res = one1.delay()
    
    task_list.append(res.id)

print(task_list)
    
while task_list:

    ok_task = []
    
    for task_id in task_list:
    
        a = AsyncResult(task_id, app=DDQ)

        if a.successful():
    
            print(a.get())
    
            # print()
    
            ok_task.append(a.id)
    
        # else:
    
        #     print('还在执行中')

    [task_list.remove(x) for x in ok_task]

  • 启动命令:

    (Spider) F:\Cerery_demo>celery worker -A CeleryTask -l INFO -P eventlet -c 99
    
# -A CeleryTask 指定目录 会自动寻找 celery.py 文件执行

定时任务

  • 使用apply_async定时执行
    from CeleryTask.celery import DDQ
    
from celery.result import AsyncResult
    
from CeleryTask.TaskOne import one1, one, one2
    
from CeleryTask.TaskTwo import two1 as two, two2

# one1.delay()
    
# one2.delay()
    
# two1.delay()
    
# two2.delay()
    
# one1.delay()

# one.delay(10,10)
    
# two.delay(20,20)

# 定时任务我们不在使用delay这个方法了,delay是立即交给task 去执行
    
# 现在我们使用apply_async定时执行

# 首先我们要先给task一个执行任务的时间
    
import datetime, time

# 获取当前时间 此时间为东八区时间
    
ctime = time.time()
    
# 将当前的东八区时间改为 UTC时间 注意这里一定是UTC时间,没有其他说法
    
utc_time = datetime.datetime.utcfromtimestamp(ctime)
    
# 为当前时间增加 10 秒
    
add_time = datetime.timedelta(seconds=20)
    
action_time = utc_time + add_time

# action_time 就是当前时间未来10秒之后的时间
    
# 现在我们使用apply_async定时执行
    
res = one.apply_async(args=(10, 10), eta=action_time)
    
print(res.id)
    
# 这样原本延迟5秒执行的One函数现在就要在10秒钟以后执行了


    [2019-05-14 13:43:34,434: INFO/MainProcess] Received task: CeleryTask.TaskOne.one[18483e45-6f99-4c8f-a7c3-9586218f0425]  ETA:[2019-05-14 05:43:54.374500+00:00]
    
[2019-05-14 13:43:59,386: INFO/MainProcess] Task CeleryTask.TaskOne.one[18483e45-6f99-4c8f-a7c3-9586218f0425] succeeded in 5.0s: 100

# 13:43:34 发布 执行完城时间  13:43:59   任务中睡了5秒

周期任务

  • 在 celery 中配置

    from celery import Celery
    
from celery.schedules import crontab

DDQ = Celery("DDQ", broker='redis://127.0.0.1:6379', backend='redis://127.0.0.1:6379'
    
             , include=['CeleryTask.TaskOne', 'CeleryTask.TaskTwo'])
    
# 我要要对beat任务生产做一个配置,这个配置的意思就是每10秒执行一次Celery_task.task_one任务参数是(10,10)

DDQ.conf.beat_schedule = {
    
    "each10s_task": {
    
        "task": "CeleryTask.TaskOne.one",
    
        "schedule": 10,  # 每10秒钟执行一次
    
        "args": (10, 10)
    
    },
    
    "each1m_task": {
    
        "task": "CeleryTask.TaskOne.one1",
    
        "schedule": crontab(minute=1),  # 每一分钟执行一次
    
        # "args": (10, 10)
    
    },
    
    "each24hours_task": {
    
        "task": "CeleryTask.TaskOne.one",
    
        "schedule": crontab(hour=23),  # 每24小时执行一次
    
        "args": (10, 10)
    
    }

}

# 以上配置完成之后,还有一点非常重要
    
# 不能直接创建Worker了,因为我们要执行周期任务,所以首先要先有一个任务的生产方
    
# celery beat -A CeleryTask
    
# celery worker -A CeleryTask -l INFO -P eventlet

    • 生产者

      celery beat -A CeleryTask

    • 消费者

      celery worker -A CeleryTask -l INFO -P eventlet

相关文章