所有演示均基于Django2.0

celery是一個(gè)基于python開發(fā)的簡(jiǎn)單、靈活且可靠的分布式任務(wù)隊(duì)列框架，支持使用任務(wù)隊(duì)列的方式在分布式的機(jī)器/進(jìn)程/線程上執(zhí)行任務(wù)調(diào)度。采用典型的生產(chǎn)者-消費(fèi)者模型，主要由三部分組成：

• 消息隊(duì)列broker：broker實(shí)際上就是一個(gè)MQ隊(duì)列服務(wù)，可以使用redis、rabbitmq等作為broker
• 處理任務(wù)的消費(fèi)者workers：broker通知worker隊(duì)列中有任務(wù)，worker去隊(duì)列中取出任務(wù)執(zhí)行，每一個(gè)worker就是一個(gè)進(jìn)程
• 存儲(chǔ)結(jié)果的backend：執(zhí)行結(jié)果存儲(chǔ)在backend，默認(rèn)也會(huì)存儲(chǔ)在broker使用的MQ隊(duì)列服務(wù)中，也可以單獨(dú)配置用何種服務(wù)做backend

異步任務(wù)

我的異步使用場(chǎng)景為項(xiàng)目上線：前端web上有個(gè)上線按鈕，點(diǎn)擊按鈕后發(fā)請(qǐng)求給后端，后端執(zhí)行上線過程要5分鐘，后端在接收到請(qǐng)求后把任務(wù)放入隊(duì)列異步執(zhí)行，同時(shí)馬上返回給前端一個(gè)任務(wù)執(zhí)行中的結(jié)果。若果沒有異步執(zhí)行會(huì)怎么樣呢？同步的情況就是執(zhí)行過程中前端一直在等后端返回結(jié)果，頁面轉(zhuǎn)呀轉(zhuǎn)的就轉(zhuǎn)超時(shí)了。

異步任務(wù)配置

1.安裝rabbitmq，這里我們使用rabbitmq作為broker，安裝完成后默認(rèn)啟動(dòng)了，也不需要其他任何配置

# apt-get install rabbitmq-server

2.安裝celery

# pip3 install celery

3.celery用在django項(xiàng)目中，django項(xiàng)目目錄結(jié)構(gòu)(簡(jiǎn)化)如下

website/
|-- deploy
|  |-- admin.py
|  |-- apps.py
|  |-- __init__.py
|  |-- models.py
|  |-- tasks.py
|  |-- tests.py
|  |-- urls.py
|  `-- views.py
|-- manage.py
|-- README
`-- website
  |-- celery.py
  |-- __init__.py
  |-- settings.py
  |-- urls.py
  `-- wsgi.py

4.創(chuàng)建 website/celery.py 主文件

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

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

app = Celery('website')

# 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()

# 允許root 用戶運(yùn)行celery
platforms.C_FORCE_ROOT = True

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

5.在 website/__init__.py 文件中增加如下內(nèi)容，確保django啟動(dòng)的時(shí)候這個(gè)app能夠被加載到

from __future__ import absolute_import

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

6.各應(yīng)用創(chuàng)建tasks.py文件，這里為 deploy/tasks.py

from __future__ import absolute_import
from celery import shared_task

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

注意tasks.py必須建在各app的根目錄下，且只能叫tasks.py，不能隨意命名

7.views.py中引用使用這個(gè)tasks異步處理

from deploy.tasks import add

def post(request):
  result = add.delay(2, 3)

result.ready()
result.get(timeout=1)
result.traceback

8.啟動(dòng)celery

# celery -A website worker -l info

9.這樣在調(diào)用post這個(gè)方法時(shí)，里邊的add就可以異步處理了

定時(shí)任務(wù)

定時(shí)任務(wù)的使用場(chǎng)景就很普遍了，比如我需要定時(shí)發(fā)送報(bào)告給老板~

定時(shí)任務(wù)配置

1. website/celery.py 文件添加如下配置以支持定時(shí)任務(wù)crontab

from celery.schedules import crontab

app.conf.update(
  CELERYBEAT_SCHEDULE = {
    'sum-task': {
      'task': 'deploy.tasks.add',
      'schedule': timedelta(seconds=20),
      'args': (5, 6)
    }
    'send-report': {
      'task': 'deploy.tasks.report',
      'schedule': crontab(hour=4, minute=30, day_of_week=1),
    }
  }
)

定義了兩個(gè)task：

• 名字為'sum-task'的task，每20秒執(zhí)行一次add函數(shù)，并傳了兩個(gè)參數(shù)5和6
• 名字為'send-report'的task，每周一早上4：30執(zhí)行report函數(shù)

timedelta是datetime中的一個(gè)對(duì)象，需要 from datetime import timedelta 引入，有如下幾個(gè)參數(shù)

• days
• seconds
• microseconds
• milliseconds
• minutes
• hours

crontab的參數(shù)有：

month_of_year
day_of_month
day_of_week
hour
minute

2. deploy/tasks.py 文件添加report方法：

@shared_task
def report():
  return 5

3.啟動(dòng)celery beat，celery啟動(dòng)了一個(gè)beat進(jìn)程一直在不斷的判斷是否有任務(wù)需要執(zhí)行

# celery -A website beat -l info

Tips

1.如果你同時(shí)使用了異步任務(wù)和計(jì)劃任務(wù)，有一種更簡(jiǎn)單的啟動(dòng)方式 celery -A website worker -b -l info ，可同時(shí)啟動(dòng)worker和beat

2.如果使用的不是rabbitmq做隊(duì)列那么需要在主配置文件中 website/celery.py 配置broker和backend，如下：

# redis做MQ配置
app = Celery('website', backend='redis', broker='redis://localhost')
# rabbitmq做MQ配置
app = Celery('website', backend='amqp', broker='amqp://admin:admin@localhost')

3.celery不能用root用戶啟動(dòng)的話需要在主配置文件中添加 platforms.C_FORCE_ROOT = True

4.celery在長時(shí)間運(yùn)行后可能出現(xiàn)內(nèi)存泄漏，需要添加配置 CELERYD_MAX_TASKS_PER_CHILD = 10 ，表示每個(gè)worker執(zhí)行了多少個(gè)任務(wù)就死掉

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評(píng)論