任务调度Celery

柔情只为你懂 2022-07-12 15:29 273阅读 0赞

安装：pip install -U Celery

中间件redis需要安装： pip install -U celery-with-redis

简单的任务：

\#tasks.py

# -*-coding:utf-8-*-
    """
    ayou
    """
    from celery import Celery
    #broker,是中间人(任务队列本身)，Celery 扮演生产者和消费者的角色，brokers 就是生产者和消费者存放/拿取任务的队列
    ##backend，是结果储存的地方，队列中的任务运行完后的结果或者状态需要被任务发送者知道，那么就需要一个地方储存这些结果
    #配置好celery的backend和broker
    app = Celery('tasks',  backend='redis://localhost:6379/0', broker='redis://localhost:6379/0')
    
    #普通函数装饰为 celery task
    #app.task实际上是将一个正常的函数修饰成了一个 celery task 对象
    @app.task
    def add(x, y):
        return x + y

启动Celery命令

celery -A tasks(文件名) worker --loglevel=info

\#意思就是运行 tasks 这个任务集合的 worker 进行工作（当然此时broker中还没有任务，worker此时相当于待命状态）

使用任务调度：

\#trigger.py

# -*-coding:utf-8-*-
    """
    ayou
    """
    from tasks import add
    import time
    
    #这里需要用 celery 提供的接口 delay 进行调用
    result = add.delay(4, 4)
    #result.ready()，判断任务是否完成
    while not result.ready():
        time.sleep(1)
    #result.get()，取出结果
    print ('task done: {0}'.format(result.get()))

根据任务状态执行不同操作

任务执行后，根据任务状态执行不同操作需要我们复写 task 的 on\_failure、on\_success 等方法：

修改tasks.py

# -*-coding:utf-8-*-
    """
    ayou
    """
    from celery import Celery,Task
    #broker,是中间人(任务队列本身)，Celery 扮演生产者和消费者的角色，brokers 就是生产者和消费者存放/拿取任务的队列
    ##backend，是结果储存的地方，队列中的任务运行完后的结果或者状态需要被任务发送者知道，那么就需要一个地方储存这些结果
    #配置好celery的backend和broker
    app = Celery('tasks',  backend='redis://localhost:6379/0', broker='redis://localhost:6379/0')
    
    #继承Task类
    class MyTask(Task):
        #重写成功后的处理
        def on_success(self, retval, task_id, args, kwargs):
            print('task done: {0}'.format(retval))
            # 返回父类中on_success的处理
            return super(MyTask, self).on_success(retval, task_id, args, kwargs)
    
        #重写失败后的处理
        def on_failure(self, exc, task_id, args, kwargs, einfo):
            print('task fail, reason: {0}'.format(exc))
            #返回父类中on_failure的处理
            return super(MyTask, self).on_failure(exc, task_id, args, kwargs, einfo)
    
    #普通函数装饰为 celery task
    #app.task实际上是将一个正常的函数修饰成了一个 celery task 对象
    @app.task(base=MyTask)
    def add(x, y):
        return x + y

trigger.py文件不变，启动Celery: celery -A tasks worker --loglevel=info

结果还是和之前一样，因为是成功的处理，只是多打印了task done: 8这段话

如果把trigger.py中的一个参数改成字符串，那么就会出现错误信息