分布式任务队列Celery入门与进阶

一、简介

　　celery是由python开发、简单、灵活、可靠的分布式任务队列，其本质是生产者消费者模型，生产者发送任务到消息队列，消费者负责处理任务。celery侧重于实时操作，但对调度支持也很好，其每天可以处理数以百万计的任务。特点：

• 简单：熟悉celery的工作流程后，配置使用简单
• 高可用：当任务执行失败或执行过程中发生连接中断，celery会自动尝试重新执行任务
• 快速：一个单进程的celery每分钟可处理上百万个任务
• 灵活：几乎celery的各个组件都可以被扩展及自定制

应用场景举例：

　　1.web应用：当用户在网站进行某个操作需要很长时间完成时，我们可以将这种操作交给celery执行，直接返回给用户，等到celery执行完成以后通知用户，大大提好网站的并发以及用户的体验感。

　　2.任务场景：比如在运维场景下需要批量在几百台机器执行某些命令或者任务，此时celery可以轻松搞定。

　　3.定时任务：向定时导数据报表、定时发送通知类似场景，虽然linux的计划任务可以帮我实现，但是非常不利于管理，而celery可以提供管理接口和丰富的api。

二、架构&工作原理

　　celery由以下三部分构成：消息中间件(broker)、任务执行单元worker、结果存储(backend)，如下图：

　　

工作原理：

1. 任务模块task包含异步任务和定时任务。其中，异步任务通常在业务逻辑中被触发并发往消息队列，而定时任务由celery beat进程周期性地将任务发往消息队列；
2. 任务执行单元worker实时监视消息队列获取队列中的任务执行；
3. woker执行完任务后将结果保存在backend中;

消息中间件broker

　　消息中间件broker官方提供了很多备选方案，支持rabbitmq、redis、amazon sqs、mongodb、memcached 等，官方推荐rabbitmq。

任务执行单元worker

　　worker是任务执行单元，负责从消息队列中取出任务执行，它可以启动一个或者多个，也可以启动在不同的机器节点，这就是其实现分布式的核心。

结果存储backend

　　backend结果存储官方也提供了诸多的存储方式支持：rabbitmq、 redis、memcached,sqlalchemy, django orm、apache cassandra、elasticsearch。

三、安装使用　

　　这里我使用的redis作为消息中间件，redis安装可以参考https://www.cnblogs.com/wdliu/p/9360286.html。

celery安装：　

pip3 install celery

简单使用

　　目录结构：

project/
├── __init__.py  
├── config.py
└── tasks.py

各目录文件说明：

__init__.py:初始化celery以及加载配置文件

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd
from celery import celery
app = celery('project')                                # 创建 celery 实例
app.config_from_object('project.config')               # 加载配置模块

config.py:  celery相关配置文件，更多配置参考：http://docs.celeryproject.org/en/latest/userguide/configuration.html

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd

broker_url = 'redis://10.1.210.69:6379/0' # broker配置，使用redis作为消息中间件

celery_result_backend = 'redis://10.1.210.69:6379/0' # backend配置，这里使用redis

celery_result_serializer = 'json' # 结果序列化方案

celery_task_result_expires = 60 * 60 * 24 # 任务过期时间

celery_timezone='asia/shanghai'   # 时区配置

celery_imports = (     # 指定导入的任务模块,可以指定多个
    'project.tasks',
)

tasks.py ：任务定义文件

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd

from project import app
@app.task
def show_name(name):
    return name

启动worker：

celery worker -a project -l debug

各个参数含义：

　　worker: 代表第启动的角色是work当然还有beat等其他角色；

　　-a ：项目路径，这里我的目录是project

　　-l：启动的日志级别，更多参数使用celery --help查看

查看日志输出，会发现我们定义的任务，以及相关配置：

 

　　虽然启动了worker，但是我们还需要通过delay或apply_async来将任务添加到worker中，这里我们通过交互式方法添加任务，并返回asyncresult对象，通过asyncresult对象获取结果：

asyncresult除了get方法用于常用获取结果方法外还提以下常用方法或属性：

• state: 返回任务状态；
• task_id: 返回任务id；
• result: 返回任务结果，同get()方法；
• ready(): 判断任务是否以及有结果，有结果为true，否则false；
• info(): 获取任务信息，默认为结果；
• wait(t): 等待t秒后获取结果，若任务执行完毕，则不等待直接获取结果，若任务在执行中，则wait期间一直阻塞，直到超时报错；
• successfu(): 判断任务是否成功，成功为true，否则为false；

四、进阶使用

　　对于普通的任务来说可能满足不了我们的任务需求，所以还需要了解一些进阶用法，celery提供了诸多调度方式，例如任务编排、根据任务状态执行不同的操作、重试机制等，以下会对常用高阶用法进行讲述。

定时任务&计划任务

　　celery的提供的定时任务主要靠schedules来完成，通过beat组件周期性将任务发送给woker执行。在示例中，新建文件period_task.py，并添加任务到配置文件中：

period_task.py：

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd
from project import app
from celery.schedules import crontab

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    sender.add_periodic_task(10.0, add.s(1,3), name='1+3=') # 每10秒执行add
    sender.add_periodic_task(
        crontab(hour=16, minute=56, day_of_week=1),      #每周一下午四点五十六执行sayhai
        sayhi.s('wd'),name='say_hi'
    )



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


@app.task
def sayhi(name):
    return 'hello %s' % name

config.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd

broker_url = 'redis://10.1.210.69:6379/0' # broker配置，使用redis作为消息中间件

celery_result_backend = 'redis://10.1.210.69:6379/0' # backend配置，这里使用redis

celery_result_serializer = 'json' # 结果序列化方案

celery_task_result_expires = 60 * 60 * 24 # 任务过期时间

celery_timezone='asia/shanghai'   # 时区配置

celery_imports = (     # 指定导入的任务模块,可以指定多个
    'project.tasks',
    'project.period_task', #定时任务
)

启动worker和beat：

celery worker -a project -l debug #启动work
celery beat -a  project.period_task -l  debug #启动beat，注意此时对应的文件路径

我们可以观察worker日志：

还可以通过配置文件方式指定定时和计划任务,此时的配置文件如下：

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd

from project import app
from celery.schedules import crontab

broker_url = 'redis://10.1.210.69:6379/0' # broker配置，使用redis作为消息中间件

celery_result_backend = 'redis://10.1.210.69:6379/0' # backend配置，这里使用redis

celery_result_serializer = 'json' # 结果序列化方案

celery_task_result_expires = 60 * 60 * 24 # 任务过期时间

celery_timezone='asia/shanghai'   # 时区配置

celery_imports = (     # 指定导入的任务模块,可以指定多个
    'project.tasks',
    'project.period_task',
)

app.conf.beat_schedule = {
    'period_add_task': {    # 计划任务
        'task': 'project.period_task.add',  #任务路径
        'schedule': crontab(hour=18, minute=16, day_of_week=1),
        'args': (3, 4),
    },
'add-every-30-seconds': {          # 每10秒执行
        'task': 'project.period_task.sayhi',  #任务路径
        'schedule': 10.0,
        'args': ('wd',)
    },
}

此时的period_task.py只需要注册到woker中就行了，如下：

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd
from project import app

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


@app.task
def sayhi(name):
    return 'hello %s' % name

同样启动worker和beat结果和第一种方式一样。更多详细的内容请参考：http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html#crontab-schedules

任务绑定

　　celery可通过任务绑定到实例获取到任务的上下文，这样我们可以在任务运行时候获取到任务的状态，记录相关日志等。

修改任务中的period_task.py,如下：

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd
from project import app
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)
@app.task(bind=true)  # 绑定任务
def add(self,x,y):
    logger.info(self.request.__dict__)  #打印日志
    try:
        a=[]
        a[10]==1
    except exception as e:
        raise self.retry(exc=e, countdown=5, max_retries=3) # 出错每5秒尝试一次，总共尝试3次
    return x+y

在以上代码中，通过bind参数将任务绑定，self指任务的上下文，通过self获取任务状态，同时在任务出错时进行任务重试，我们观察日志：

内置钩子函数

　　celery在执行任务时候，提供了钩子方法用于在任务执行完成时候进行对应的操作，在task源码中提供了很多状态钩子函数如：on_success(成功后执行)、on_failure(失败时候执行)、on_retry(任务重试时候执行)、after_return(任务返回时候执行),在进行使用是我们只需要重写这些方法，完成相应的操作即可。

在以下示例中，我们继续修改period_task.py，分别定义三个任务来演示任务失败、重试、任务成功后执行的操作：

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd
from project import app
from celery.utils.log import get_task_logger
from celery import task

logger = get_task_logger(__name__)

class demotask(task):

    def on_success(self, retval, task_id, args, kwargs):   # 任务成功执行
        logger.info('task id:{} , arg:{} , successful !'.format(task_id,args))



    def on_failure(self, exc, task_id, args, kwargs, einfo):  #任务失败执行
        logger.info('task id:{} , arg:{} , failed ! erros : {}' .format(task_id,args,exc))


    def on_retry(self, exc, task_id, args, kwargs, einfo):    #任务重试执行
        logger.info('task id:{} , arg:{} , retry !  einfo: {}'.format(task_id, args, exc))

@app.task(base=demotask,bind=true)
def add(self,x,y):
    try:
        a=[]
        a[10]==1
    except exception as e:
        raise self.retry(exc=e, countdown=5, max_retries=1) # 出错每5秒尝试一次，总共尝试1次
    return x+y

@app.task(base=demotask)
def sayhi(name):
    a=[]
    a[10]==1
    return 'hi {}'.format(name)

@app.task(base=demotask)
def sum(a,b):
    return 'a+b={} '.format(a+b)

此时的配置文件config.py：

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd

from project import app
from celery.schedules import crontab

broker_url = 'redis://10.1.210.69:6379/0' # broker配置，使用redis作为消息中间件

celery_result_backend = 'redis://10.1.210.69:6379/0' # backend配置，这里使用redis

celery_result_serializer = 'json' # 结果序列化方案

celery_task_result_expires = 60 * 60 * 24 # 任务过期时间

celery_timezone='asia/shanghai'   # 时区配置

celery_imports = (     # 指定导入的任务模块,可以指定多个
    'project.tasks',
    'project.period_task',
)

app.conf.beat_schedule = {
'add': {          # 每10秒执行
        'task': 'project.period_task.add',  #任务路径
        'schedule': 10.0,
        'args': (10,12),
    },
'sayhi': {          # 每10秒执行
        'task': 'project.period_task.sayhi',  #任务路径
        'schedule': 10.0,
        'args': ('wd',),
    },
'sum': {          # 每10秒执行
        'task': 'project.period_task.sum',  #任务路径
        'schedule': 10.0,
        'args': (1,3),
    },
}

然后重启worker和beat，查看日志：

 

任务编排

　　在很多情况下，一个任务需要由多个子任务或者一个任务需要很多步骤才能完成，celery同样也能实现这样的任务，完成这类型的任务通过以下模块完成：

• group: 并行调度任务

• chain: 链式任务调度

• chord: 类似group，但分header和body2个部分，header可以是一个group任务，执行完成后调用body的任务

• map: 映射调度，通过输入多个入参来多次调度同一个任务

• starmap: 类似map，入参类似＊args

• chunks: 将任务按照一定数量进行分组

 

修改tasks.py：

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# author:wd
from project import app

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


@app.task
def mul(x,y):
    return x*y


@app.task
def sum(data_list):
    res=0
    for i in data_list:
        res+=i
    return res

 

group: 组任务，组内每个任务并行执行

和project同级目录新建consumer.py如下：

from celery import group
from project.tasks import add,mul,sum
res = group(add.s(1,2),add.s(1,2))()  # 任务 [1+2,1+2] 
while true:
    if res.ready():
        print('res:{}'.format(res.get()))
        break

结果：

 

chain：链式任务

链式任务中，默认上一个任务的返回结果作为参数传递给子任务

from celery import chain
from project.tasks import add,mul,sum
res = chain(add.s(1,2),add.s(3),mul.s(3))()  # 任务((1+2)+3)*3
while true:
    if res.ready():
        print('res:{}'.format(res.get()))
        break
#结果
#res：18

还可以使用|表示链式任务，上面任务也可以表示为：

res = (add.s(1,2) | add.s(3) | (mul.s(3)))()
res.get()

 

chord：任务分割，分为header和body两部分，hearder任务执行完在执行body,其中hearder返回结果作为参数传递给body

from celery import chord
from project.tasks import add,mul,sum
res = chord(header=[add.s(1,2),mul.s(3,4)],body=sum.s())()  # 任务(1+2)+(3*4)
while true:
    if res.ready():
        print('res:{}'.format(res.get()))
        break

#结果：
#res:15

 

chunks：任务分组，按照任务的个数分组

from project.tasks import add,mul,sum
res = add.chunks(zip(range(5),range(5)),4)()  # 4 代表每组的任务的个数
while true:
    if res.ready():
        print('res:{}'.format(res.get()))
        break

结果：

 

delay &apply_async

　　对于delay和apply_async都可以用来进行任务的调度，本质上是delay对apply_async进行了再一次封装（或者可以说是快捷方式），两者都返回asyncresult对象，以下是两个方法源码。

    def delay(self, *args, **kwargs):
        """star argument version of :meth:`apply_async`.

        does not support the extra options enabled by :meth:`apply_async`.

        arguments:
            *args (any): positional arguments passed on to the task.
            **kwargs (any): keyword arguments passed on to the task.
        returns:
            celery.result.asyncresult: future promise.
        """
        return self.apply_async(args, kwargs)

delay源码

    def apply_async(self, args=none, kwargs=none, task_id=none, producer=none,
                    link=none, link_error=none, shadow=none, **options):
        """apply tasks asynchronously by sending a message.

        arguments:
            args (tuple): the positional arguments to pass on to the task.

            kwargs (dict): the keyword arguments to pass on to the task.

            countdown (float): number of seconds into the future that the
                task should execute.  defaults to immediate execution.

            eta (~datetime.datetime): absolute time and date of when the task
                should be executed.  may not be specified if `countdown`
                is also supplied.

            expires (float, ~datetime.datetime): datetime or
                seconds in the future for the task should expire.
                the task won't be executed after the expiration time.

            shadow (str): override task name used in logs/monitoring.
                default is retrieved from :meth:`shadow_name`.

            connection (kombu.connection): re-use existing broker connection
                instead of acquiring one from the connection pool.

            retry (bool): if enabled sending of the task message will be
                retried in the event of connection loss or failure.
                default is taken from the :setting:`task_publish_retry`
                setting.  note that you need to handle the
                producer/connection manually for this to work.

            retry_policy (mapping): override the retry policy used.
                see the :setting:`task_publish_retry_policy` setting.

            queue (str, kombu.queue): the queue to route the task to.
                this must be a key present in :setting:`task_queues`, or
                :setting:`task_create_missing_queues` must be
                enabled.  see :ref:`guide-routing` for more
                information.

            exchange (str, kombu.exchange): named custom exchange to send the
                task to.  usually not used in combination with the ``queue``
                argument.

            routing_key (str): custom routing key used to route the task to a
                worker server.  if in combination with a ``queue`` argument
                only used to specify custom routing keys to topic exchanges.

            priority (int): the task priority, a number between 0 and 9.
                defaults to the :attr:`priority` attribute.

            serializer (str): serialization method to use.
                can be `pickle`, `json`, `yaml`, `msgpack` or any custom
                serialization method that's been registered
                with :mod:`kombu.serialization.registry`.
                defaults to the :attr:`serializer` attribute.

            compression (str): optional compression method
                to use.  can be one of ``zlib``, ``bzip2``,
                or any custom compression methods registered with
                :func:`kombu.compression.register`.
                defaults to the :setting:`task_compression` setting.

            link (signature): a single, or a list of tasks signatures
                to apply if the task returns successfully.

            link_error (signature): a single, or a list of task signatures
                to apply if an error occurs while executing the task.

            producer (kombu.producer): custom producer to use when publishing
                the task.

            add_to_parent (bool): if set to true (default) and the task
                is applied while executing another task, then the result
                will be appended to the parent tasks ``request.children``
                attribute.  trailing can also be disabled by default using the
                :attr:`trail` attribute

            publisher (kombu.producer): deprecated alias to ``producer``.

            headers (dict): message headers to be included in the message.

        returns:
            celery.result.asyncresult: promise of future evaluation.

        raises:
            typeerror: if not enough arguments are passed, or too many
                arguments are passed.  note that signature checks may
                be disabled by specifying ``@task(typing=false)``.
            kombu.exceptions.operationalerror: if a connection to the
               transport cannot be made, or if the connection is lost.

        note:
            also supports all keyword arguments supported by
            :meth:`kombu.producer.publish`.
        """
        if self.typing:
            try:
                check_arguments = self.__header__
            except attributeerror:  # pragma: no cover
                pass
            else:
                check_arguments(*(args or ()), **(kwargs or {}))

        app = self._get_app()
        if app.conf.task_always_eager:
            with denied_join_result():
                return self.apply(args, kwargs, task_id=task_id or uuid(),
                                  link=link, link_error=link_error, **options)

        if self.__v2_compat__:
            shadow = shadow or self.shadow_name(self(), args, kwargs, options)
        else:
            shadow = shadow or self.shadow_name(args, kwargs, options)

        preopts = self._get_exec_options()
        options = dict(preopts, **options) if options else preopts

        options.setdefault('ignore_result', self.ignore_result)

        return app.send_task(
            self.name, args, kwargs, task_id=task_id, producer=producer,
            link=link, link_error=link_error, result_cls=self.asyncresult,
            shadow=shadow, task_type=self,
            **options
        )

apply_async源码

对于其使用，apply_async支持常用参数：

• eta：指定任务执行时间，类型为datetime时间类型；
• countdown：倒计时,单位秒，浮点类型；
• expires：任务过期时间，如果任务在超过过期时间还未执行则回收任务，浮点类型获取datetime类型；
• retry：任务执行失败时候是否尝试，布尔类型。；
• serializer：序列化方案，支持pickle、json、yaml、msgpack；
• priority：任务优先级，有0～9优先级可设置，int类型；
• retry_policy：任务重试机制，其中包含几个重试参数，类型是dict如下：

max_retries:最大重试次数

interval_start:重试等待时间

interval_step:每次重试叠加时长，假设第一重试等待1s，第二次等待1＋n秒

interval_max:最大等待时间

####示例
 add.apply_async((1, 3), retry=true, retry_policy={
        'max_retries': 1,
        'interval_start': 0,
        'interval_step': 0.8,
        'interval_max': 5,
    })

view code

更多参数参考：http://docs.celeryproject.org/en/latest/reference/celery.app.task.html#celery.app.task.task.apply_async

 

　　

 五、管理与监控

　　celery管理和监控功能是通过flower组件实现的，flower组件不仅仅提供监控功能，还提供http api可实现对woker和task的管理。

安装使用

pip3 install flower

启动

 flower -a project --port=5555   
# -a ：项目目录
#--port 指定端口

访问http:ip:5555

api使用，例如获取woker信息：

curl http://127.0.0.1:5555/api/workers

结果：

更多api参考：https://flower.readthedocs.io/en/latest/api.html