Celery 分布式任务队列快速入门

proj/__init__.py
    /celery.py
    /tasks.py

from __future__ import absolute_import, unicode_literals
#从python的绝对路径导入而不是当前的脚本     #在python2和python3做兼容支持的
from celery import Celery

app = Celery(‘proj‘,#app的名字
             broker=‘redis://:123456@192.168.31.128‘,#连rabbitmq或redis
             backend=‘redis://:123456@192.168.31.128‘,
             include=[‘s3proj.tasks‘,‘s3proj.tasks2‘])

# Optional configuration, see the application user guide.
app.conf.update(#给app设置参数
    result_expires=3600,#保存时间为1小时
)

if __name__ == ‘__main__‘:
    app.start()

from __future__ import absolute_import, unicode_literals
#从python的绝对路径导入而不是当前的脚本     #在python2和python3做兼容支持的
from .celery import app

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

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

@app.task
def xsum(numbers):
    return sum(numbers)

####################################

from __future__ import absolute_import, unicode_literals
#从python的绝对路径导入而不是当前的脚本     #在python2和python3做兼容支持的
from .celery import app


@app.task
def cmd(cmd):
    print("running cmd",cmd)


@app.task
def file_transfer(filename):
    print("sending file",filename)

$ celery -A proj worker -l info

-------------- celery@Alexs-MacBook-Pro.local v4.0.2 (latentcall)
---- **** -----
--- * ***  * -- Darwin-15.6.0-x86_64-i386-64bit 2017-01-26 21:50:24
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app:         proj:0x103a020f0
- ** ---------- .> transport:   redis://localhost:6379//
- ** ---------- .> results:     redis://localhost/
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery

$ celery multi stopwait w1 -A proj -l info

app.conf.beat_schedule = {
    ‘add-every-30-seconds‘: {
        ‘task‘: ‘tasks.add‘,
        ‘schedule‘: 30.0,
        ‘args‘: (16, 16)
    },
}
app.conf.timezone = ‘UTC‘

$ celery -A periodic_task beat

$ celery -A s3proj.periodic_tasks beat -l debug

celery beat v4.0.2 (latentcall) is starting.
__    -    ... __   -        _
LocalTime -> 2017-02-08 18:39:31
Configuration ->
    . broker -> redis://localhost:6379//
    . loader -> celery.loaders.app.AppLoader
    . scheduler -> celery.beat.PersistentScheduler
    . db -> celerybeat-schedule
    . logfile -> [stderr]@%WARNING
    . maxinterval -> 5.00 minutes (300s)

$ celery -A periodic_task worker
  
 -------------- celery@Alexs-MacBook-Pro.local v4.0.2 (latentcall)
---- **** -----
--- * ***  * -- Darwin-15.6.0-x86_64-i386-64bit 2017-02-08 18:42:08
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x104d420b8
- ** ---------- .> transport:   redis://localhost:6379//
- ** ---------- .> results:     redis://localhost/
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery

$ celery -A periodic_task beat -s /home/celery/var/run/celerybeat-schedule

from celery.schedules import crontab
 
app.conf.beat_schedule = {
    # Executes every Monday morning at 7:30 a.m.
    ‘add-every-monday-morning‘: {
        ‘task‘: ‘s3proj.tasks.add‘,
        ‘schedule‘: crontab(hour=7, minute=30, day_of_week=1),
        ‘args‘: (16, 16),
    },
}

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

# set the default Django settings module for the ‘celery‘ program.
os.environ.setdefault(‘DJANGO_SETTINGS_MODULE‘, ‘PerfectCRM.settings‘)

app = Celery(‘celery_task‘)

# Using a string here means the worker don‘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‘)
#你可以把celery连接的RabbitMQ存到settings但你必须按他规定的格式，必须CELERY大写开头

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()
#能够找到所有app下的selery任务

@app.task(bind=True)
def debug_task(self):
    print(‘Request: {0!r}‘.format(self.request))

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‘]
#找这个所有项目下的所有APP

from __future__ import absolute_import

os.environ.setdefault(‘DJANGO_SETTINGS_MODULE‘, ‘PerfectCRM.settings‘)
　　　　　　　　　　　　　　　　　　　　　　　　　　　　#项目名.settings

app = Celery(‘celery_task‘)

app.config_from_object(‘django.conf:settings‘, namespace=‘CELERY‘)

#你可以把celery连接的RabbitMQ存到settings但你必须按他规定的格式，必须CELERY大写开头

app.autodiscover_tasks()
#不管你在那个app里创建clelry任务,他都能自动发现，就像djangoadmin一样能自动发现

- app1/
    - tasks.py
    - models.py
- app2/
    - tasks.py
    - models.py

from __future__ import absolute_import, unicode_literals
from celery import shared_task
#在这个app里的任务和其他app里的任务是可以互相共享d

@shared_task
def add(x, y):
    return x + y


@shared_task
def mul(x, y):
    return x * y


@shared_task
def xsum(numbers):
    return sum(numbers)

celery -A PerfectCRM worker -l debug

from crm.tasks import add,mul   #导入worker
from celery.result import AsyncResult
# Create your views here.



def celery_test(request):
    """
    掉一个视图，把任务交给celery，并返回任务id
    :param request:
    :return:
    """
    task=add.delay(4,42)
    # res=task.get()
    # return HttpResponse(res)#返回的是任务结果
    return HttpResponse(task.id)#返回的是任务id
    
def celery_res(request):
    """
    通过任务id取出任务结果
    :param request:
    :return:
    """
    task_id="686c19f5-89d1-4c57-8e91-669f9e2716e5"
    res=AsyncResult(id=task_id)
    return HttpResponse(res.get())