五分鍾看懂Celery定時任務

 

Django下使用Celery

使用場景：

1， Web應用。 當用戶觸發的一個操作需要很長時間才能執行完成，那么就可以把它當做一個任務去交給Celery去異步執行， 執行完成之后再返回給用戶，這短時間用戶不需要等待響應處理的過程， 反而提高了網站吞吐量和響應時間。

2， 定時任務，生產環境經常會跑一些定時任務，假如你有成千上萬台服務器，上千種任務， 定時任務的管理很困難，Celery可以幫助我們快速的在不同的機器設定不同種的任務。

3， 同步完成的附加工作都可以異步完成，比如：發送短信/郵件、消息推送、清理/設置緩存等。

特性：

1， 方便地查看定時任務的執行情況，比如是否執行成功，當前狀態，執行任務花費時長等。

2， 可以使用功能齊備的管理后台或者命令行添加，更新，刪除任務等。

3， 方便把任務和配置管理相關聯。

4， 可選多進程， Eventlet和Gevent三種模式並發執行。

5， 提供錯誤處理機制。

6， 提供多種任務原語，方便實現任務分組，拆分和調用鏈。

7， 支持多種消息代理和存儲后端。

Celery的使用

pip install django

pip install django-celery

以上兩個依賴安裝完成之后就開始編寫代碼吧。

創建Dajngo工程

django-admin.py startproject dc_test       # 創建工程

django-admin.py stratapp projectmanageapp      # 創建app

創建完工程之后， 打開dc_test/dc_test/settings.py  

INSTALLED_APPS = (
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'demo',
    'djcelery'           # 添加  djcelery應用
)        

from datetime import timedelta


CELERYBEAT_SCHEDULE = {
    'add-every-30-seconds': {
        'task': 'demo.tasks.hello_world',
        'schedule': timedelta(seconds=2),
    },
}

添加完成之后，在同級目錄下，創建Celery.py 文件

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from django.conf import settings

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

app = Celery('dc_test')

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

# Load task modules from all registered Django app configs.
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)


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

celery.py文件添加完成之后在同級目錄__init__.py中添加，

from __future__ import absolute_import, unicode_literals

from .celery import app as celery_app

__all__ = ['celery_app']

在app應用下創建tasks.py文件。

from __future__ import absolute_import, unicode_literals
from celery import shared_task

@shared_task
def hello_world():
    with open("D:\dc_test\output.txt", "a") as f:
        f.write("hello world")
        f.write("\n")

文件都創建好之后，就開始運行項目吧， 等項目運行成功之后會自定生成以下文件。

celerybeat-schedule.bak

celerybeat-schedule.dat

celerybeat-schedule.dir

執行tasks:

python manage.py celery worker -l info

 

執行bate：

python manage.py ceery beat

到這里之后，就可以了。趕快試試吧。

源碼地址： https://github.com/Mrwyc/Django-Celery-Demo