celery 使用交流

celery简介

celery 是一个基于分布消息传递的异步任务队列，他可让任务的执行脱离主程序，也能够分配到其余得主机上运行。咱们能够用它实现异步任务和定时任务。
复制代码

使用场景

1.你想对100台机器执行一条批量命令，可能会花很长时间 ，但你不想让你的程序等着结果返回，而是给你返回 一个任务ID,
你过一段时间只须要拿着这个任务id就能够拿到任务执行结果， 在任务执行进行时，你能够继续作其它的事情。
2.你想作一个定时任务，好比天天检测一下大家全部客户的资料，若是发现今天 是客户的生日，就给他发个短信祝福
复制代码

celery模块

任务模块 Task

包含异步任务和定时任务，异步任务一般在业务逻辑中被触发并分发往任务队列，定时任务是由celery beat 进程周期性地姜任务发往任务队列
复制代码

消息中间件 Broker

celery使用消息进行通讯，broker也就是消息传递机制， 接手任务生产者发来的任务， 将任务存入队列中。

附：可以使用的broker，官方目前推荐的稳定性较高的主要是RabbitMQ和Redis：
复制代码

任务执行单元 Worker

worker是任务执行单元，他实时监控消息队列，获取队列中调度的任务并执行
复制代码

任务结果存储 Backend

Backend 用于存储任务的执行结果，以供查询。同消息中间件同样，存储也可以使用 RabbitMQ, Redis 和 MongoDB 等
复制代码

celery 基本工做流程简述

celery的简单使用

建立一个celery的异步任务能够简单的分为三步：（这里使用redis做为broker）

一、建立celery实例

二、启动worker

三、调用celery异步任务
复制代码

建立 celery 实例

首先安装celery

pip install celery
复制代码

建立 celery应用 tasks.py,在这里broker使用的是redis，关于redis的使用就不在这里说明了

# -*- coding: utf-8 -*-

import time

from celery import Celery


# celery实例，在这里个人broker和backend都是设置的redis
app = Celery('tasks', backend='redis://:@127.0.0.1:6379/2', broker='redis://:@127.0.0.1:6379/2')

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


@app.task
def sleep_task(timeout=20):
    for i in range(timeout):
        print i
        time.sleep(1)
    return timeout
复制代码

启动worker，这里的worker启动的就是以前写的tasks.py文件

celery -A tasks worker --loglevel=info
复制代码

进入Python环境

delay()是celery的任务调用方式，其中delay 是使用 apply_async 的快捷方式。apply_async 支持更多的参数，它的通常形式以下

ready()能够获取任务的完成状态 get()能够获取任务的结果 get(propagate=True),能够在任务异常时覆盖这一行为 result.traceback，能够获取异常

celery 工做流

首先看一下 signature() 与 chain()

首先引入两个包
from tasks import add
from celery import signature, chain
复制代码

利用signature建立一个任务，能够看到这个任务也支持delay(),也能够直接调用

Simple chain

chain能够将 signature任务连接在一块儿，造成一条任务链，将多个任务串联在一块儿相继调用，同时可使下一个任务使用上一个任务的返回值。注：s是signature的简写

Immutable signatures

不可变签名,能够看到每一个签名任务都与上一个任务没有关系，都是独立的任务，也能够用chain连接在一块儿

add.signature((2, 2), immutable=True)
# 能够简写为
add.si(2, 2)
res = chain(add.si(2, 2) | add.si(4, 4) | add.si(8, 8))()
复制代码

django 中使用celery

想要在django中使用celery，首先须要配置一个celery的应用，官方建议的是在settings文件的同目录下建立一个celery.py的实例

# 绝对导入，防止与celery冲突
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

# 设置Django的环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')
# celery app
app = Celery('proj')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
# should have a `CELERY_` prefix.
# 配置前缀
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django app configs.
# 自动扫描app下的tasks文件
app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))
复制代码

而后须要在init文件下引入这个模块,确保django启动时加载文件

from __future__ import absolute_import, unicode_literals

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ['celery_app']
复制代码

接下来是在settings的配置参数

# Celery settings
from __future__ import absolute_import, unicode_literals

# broker
CELERY_BROKER_URL = 'redis://:@127.0.0.1:6379/3'
# 内容类型，默认全部类型
CELERY_ACCEPT_CONTENT = ['application/json']
# 任务结果
CELERY_RESULT_BACKEND = 'django-db'
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'
# 更多配置能够查看
# http://docs.celeryproject.org/en/latest/userguide/configuration.html#std:setting-accept_content

# 定时任务配置
CELERY_BEAT_SCHEDULE = {

}
复制代码

若是 CELERY_RESULT_BACKEND 设置了django，注意须要INSTALLED_APPS配置下

配置下数据库，在数据库中就能够查看任务的执行状况了
python manage.py migrate django_celery_results
复制代码

INSTALLED_APPS = (
    ...,
    'django_celery_results',
)
复制代码

任务继承

继承Task，能够更好的控制任务，在这里重写了Task的几个方法

# -*- coding: utf-8 -*-
# ___author__ = 'gwx'

import celery


class BaseTask(celery.Task):
    """ celery 基类, 继承Task """

    def __call__(self, *args, **kwargs):
        """ :param args: :param kwargs: :return: """
        print('TASK STARTING: {0.name}[{0.request.id}]'.format(self))
        return super(BaseTask, self).__call__(*args, **kwargs)

    # 任务成功
    def on_success(self, retval, task_id, args, kwargs):
        print(self.request)
        print("success")
        print(retval)
        print(task_id)
        print(args)
        print(kwargs)

        return super(BaseTask, self).on_success(retval, task_id, args, kwargs)

    # 任务失败
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print("failure")
        print(exc)
        print(task_id)
        print(args)
        print(kwargs)
        print(einfo)
        # 失败重试
        self.retry(exc=exc)

    # finally
    def after_return(self, status, retval, task_id, args, kwargs, einfo):
        print(status, retval, task_id, args, kwargs, einfo)
        print('task %s is finished' % self.name)

复制代码

建立celery任务，这里建立了几种不一样类型的任务

# -*- coding: utf-8 -*-
# ___author__ = 'gwx'

import celery
from workflows.base import BaseTask


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


@celery.task(bind=True, base=BaseTask, name='add_success')
def add_success(self, x, y):
    """ success 演示 """
    s = x + y
    print("success: this result is %s" % s)
    return 'add success return'


@celery.task(bind=True, base=BaseTask, name='add_failure')
def add_failure(self, msg='this is a error task', max_retry=2):
    """ failure 演示 """
    print("task retries %s" % self.request.retries)
    if self.request.retries >= int(max_retry):
        return 'add failure return'
    raise Exception("add_failure message: %s" % msg)


@celery.task(bind=True, base=BaseTask, name='end_task')
def end_task(self):
    """ 最后的任务 """
    print('this is end task')
    return 'end task'


@celery.task(bind=True, base=BaseTask, name="work_flow_task")
def work_flow_task(self):
    """ 任务流, 使用chain将 si类型 任务串接起来，这样能够模板完成对主任务的拆分， 而自定义基类，能够更好的控制子任务 :param self: :return: """
    success_si = add_success.si(2, 4)
    failure_si = add_failure.si(msg='retry task', max_retry=0)
    end_si = end_task.si()
    celery.chain(success_si, failure_si, end_si)()
复制代码

启动worker，进入项目的根目录

# -l celery日志等级， -c worker数，
# 你们能够经过 celery -A proj.celery worker -- help查找更多适合项目的配置
celery -A proj.celery worker -l info -c 10
复制代码

调用任务，我这里是使用django的shell调用的 进入项目的根目录

python manage.py shell
# 引入了tasks
from workflows.tasks import *
# 调用add_success的celery任务
add_success.apply_async((2, 4), pri=0)
复制代码

worker执行结果

add_failure任务你们能够尝试本身执行下，不一样的是增长了一个重试的功能。

链式任务

多个子任务的输出结果

定时任务

celery支持定时任务, 在setting中配置

""" add 是此任务的lable name task 是要执行的定时任务，必须是celery修饰的 schedule 是定时设置，一下是3s一次 当引入时，也能够这样设置： 'schedule': timedelta(seconds=5) 同时也支持crontab() 'schedule': crontab(hour=7, minute=30, day_of_week=1) 更多可查看 http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html """
from datetime import timedelta

# 定时任务配置
CELERY_BEAT_SCHEDULE = {
    'add': {
        'task': 'demoapp.tasks.add',
        'schedule': 3,
        'args': (2, 3)
    }
}
复制代码

定时启动和worker相似

celery beat -A proj.celery -l info
复制代码

在worker中查看结果，能够看到没隔三秒钟都会执行一边add task

当任务更新时，须要从新启动worker或者定时worker，定时至关于worker中的一个进程