Celery分布式任务队列入门

broker_url = 'amqp://guest:guest@localhost:5672//'

broker_url = 'redis://:password@localhost:6379'　　

pip3 install celery

from celery import Celery
 
#app是一个worker,负责执行任务，多个worker执行时，任务执行顺序为抢占式，并不会按照类似于rabbitmq分配任务
app = Celery('tasks',   
             broker='redis://:password@localhost',   #执行任务中间件  
             backend='redis://:password@localhost')    #保存结果中间件
 
@app.task
def add(x,y):   #这是worker可以执行的一个任务
    print("running...",x,y)
    return x+y

celery -A celery_test worker -l debug

>>> from celery_test import add
>>> add.delay(4, 4)

>>> result = add.delay(4, 4)　　

>>> result.ready()　　

result.get()  

proj/__init__.py
    /celery.py   #配置信息
    /tasks.py    #任务

from __future__ import absolute_import, unicode_literals  #声明celery从python包绝对路径里导入celery包
from celery import Celery
 
app = Celery('proj',
             broker='amqp://',
             backend='amqp://',
             include=['proj.tasks'])  #任务文件路径列表,可添加多个任务
 
# Optional configuration, see the application user guide.
app.conf.update(
    result_expires=3600,   #任务结果保存时间
)
 
if __name__ == '__main__':
    app.start()

from __future__ import absolute_import, unicode_literals
from .celery import app  #导入同级目录下celery文件中的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)

 celery -A proj worker -l debug　　

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', 'PERCRM.settings')    #根据具体项目配置
 
app = Celery('PERCRM')
 
# 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')
 
# Load task modules from all registered Django app configs.
app.autodiscover_tasks()   #自动发现各个app里面创建的celery任务，可以创建多个任务

@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']　　

CELERY_BROKER_URL = 'redis://:123456@192.168.2.107'
CELERY_RESULT_BACKEND = 'redis://:123456@192.168.2.107'　　

# Create your tasks here
from __future__ import absolute_import, unicode_literals
from celery import shared_task  #可以跟其他APP共享任务


@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)

from students.tasks import add,mul


def celery_test(request):
    task = add.delay(123,456)

    return HttpResponse(task.id)

from celery import Celery
from celery.schedules import crontab
 
app = Celery()
 
@app.on_after_configure.connect  #装饰器作用：只要脚本一启动便立刻自动执行被装饰的函数
def setup_periodic_tasks(sender, **kwargs):
    # Calls test('hello') every 10 seconds.
    sender.add_periodic_task(10.0, test.s('hello'), name='add every 10')  #每隔十秒钟，执行test函数，传入参数‘hello’,
 
    # Calls test('world') every 30 seconds
    sender.add_periodic_task(30.0, test.s('world'), expires=10)  #expires任务结果保存十秒钟
  
    # Executes every Monday morning at 7:30 a.m.
    sender.add_periodic_task(
        crontab(hour=7, minute=30, day_of_week=1),
        test.s('Happy Mondays!'),  #每周一早上十点半执行test函数
    )
 
@app.task
def test(arg):
    print(arg)

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

pip3 install django-celery-beat