django_celery_beat

先祭上 https://github.com/celery/django-celery-beat

注意：

celery4.0支持django1.8以上版本，若是1.8如下版本請使用celry3.1

另外celery4.0版本不支持windows，須要啓動的時候指定下--pool=solo，下面會說明

本文從零開始，若是已有django環境和celery環境，請忽略如下步驟

VirtualEnv環境

請參照我以前的文章-virtualenv環境

這裏簡單列下

E:\virtualenv\nowamagic_venv>virtualenv proj_env1 --python=C:\Python36\python.exe
Running virtualenv with interpreter C:\Python36\python.exe
Using base prefix 'C:\\Python36'
New python executable in E:\virtualenv\nowamagic_venv\proj_env1\Scripts\python.exe
Installing setuptools, pip, wheel...done.

E:\virtualenv\nowamagic_venv>cd proj_env1

E:\virtualenv\nowamagic_venv\proj_env1>cd Scripts

E:\virtualenv\nowamagic_venv\proj_env1\Scripts>activate

(proj_env1) E:\virtualenv\nowamagic_venv\proj_env1\Scripts>pip list
DEPRECATION: The default format will switch to columns in the future. You can use --format=(legacy|columns) (or define a
ection) to disable this warning.
pip (9.0.1)
setuptools (38.5.1)
wheel (0.30.0)

有時安裝的pip版本過低，用pip安裝軟件時會報

pip._vendor.distlib.DistlibException: Unable to locate finder for 'pip._vendor.distlib'

這時須要卸載掉pip setuptools，從新安裝下

1. uninstall current pip:

   python -m pip uninstall pip setuptools

2. download get-pip.py from https://bootstrap.pypa.io/get-pip.py

3. execute get-pip script:

   python get-pip.py

 

安裝軟件

Django (2.0.2)
celery (4.1.0)
mysqlclient (1.3.12) 鏈接mysql客戶端
redis (2.10.6) 以redis做爲後端broker

安裝過程

(proj_env1) E:\virtualenv\nowamagic_venv\proj_env1\Scripts>pip install django
Collecting django
  Using cached Django-2.0.2-py3-none-any.whl
Collecting pytz (from django)
  Using cached pytz-2017.3-py2.py3-none-any.whl
Installing collected packages: pytz, django
Successfully installed django-2.0.2 pytz-2017.3

(proj_env1) E:\virtualenv\nowamagic_venv\proj_env1\Scripts>pip install celery
Collecting celery
  Using cached celery-4.1.0-py2.py3-none-any.whl
Requirement already satisfied: pytz>dev in e:\virtualenv\nowamagic_venv\proj_env1\lib\site-packages (from celery)
Collecting billiard<3.6.0,>=3.5.0.2 (from celery)
  Using cached billiard-3.5.0.3-py3-none-any.whl
Collecting kombu<5.0,>=4.0.2 (from celery)
  Using cached kombu-4.1.0-py2.py3-none-any.whl
Collecting amqp<3.0,>=2.1.4 (from kombu<5.0,>=4.0.2->celery)
  Using cached amqp-2.2.2-py2.py3-none-any.whl
Collecting vine>=1.1.3 (from amqp<3.0,>=2.1.4->kombu<5.0,>=4.0.2->celery)
  Using cached vine-1.1.4-py2.py3-none-any.whl
Installing collected packages: billiard, vine, amqp, kombu, celery
Successfully installed amqp-2.2.2 billiard-3.5.0.3 celery-4.1.0 kombu-4.1.0 vine-1.1.4

(proj_env1) E:\virtualenv\nowamagic_venv\proj_env1\Scripts>pip install mysqlclient
Collecting mysqlclient
  Using cached mysqlclient-1.3.12-cp36-cp36m-win_amd64.whl
Installing collected packages: mysqlclient
Successfully installed mysqlclient-1.3.12

(proj_env1) E:\virtualenv\nowamagic_venv\proj_env1\Scripts>pip install redis
Collecting redis
  Using cached redis-2.10.6-py2.py3-none-any.whl
Installing collected packages: redis
Successfully installed redis-2.10.6

 

啓動django項目

(proj_env1) e:\code\git\source\my>django-admin startproject django_celery_beat_test

如今結構以下

- django_celery_beat_test/
  - manage.py
  - django_celery_beat_test/
    - __init__.py
    - settings.py
    - urls.py
    - wsgi.py

定義 Celery 實例

file:  django_celery_beat_test/django_celery_beat_test/celery.py

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', 'django_celery_beat_test.settings')

app = Celery('django_celery_beat_test')

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


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

而後你須要在你的django_celery_beat_test/django_celery_beat_test/__ init__.py模塊中導入這個app。這能夠確保在Django啓動時加載app，以便@shared_task裝飾器（稍後說起）使用它：

file：  django_celery_beat_test/django_celery_beat_test/__ init__.py

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

其中

from __future__ import absolute_import

引入絕對導入是防止咱們定義的celery.py模塊與庫模塊celery衝突，具體做用可見這裏

app = Celery('django_celery_beat_test')

這是咱們celery的實例，能夠有不少實例，但django裏一個可能就夠了 

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

表明celery的配置文件在django的配置文件settings裏，而且定義了命名空間CELERY，也即在settings裏的celery的配置都得以CELERY開頭，好比broker_url須要定義爲CELERY_BROKER_URL

app.autodiscover_tasks()

使用上面的行，Celery會自動發現全部安裝的應用程序中的任務，遵循tasks.py約定：

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

這樣您就沒必要手動將各個模塊添加到CELERY_IMPORTS設置中

celery的配置

接着在settings.py中定義

CELERY_BROKER_URL = 'redis://:cds-china@172.20.3.3:6379/0'

#: Only add pickle to this list if your broker is secured
#: from unwanted access (see userguide/security.html)
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_RESULT_BACKEND = 'redis://:cds-china@172.20.3.3:6379/1'

建立app

python manage.py startapp demoapp

如今結構以下：

django_celery_beat_test/
├── demoapp
│   ├── __init__.py
│   ├── models.py
│   ├── tasks.py
│   ├── tests.py
│   ├── views.py
├── manage.py
├── django_celery_beat_test
│   ├── __init__.py
│   ├── celery.py
│   ├── settings.py
│   ├── urls.py
│   ├── wsgi.py
└── test.db

tasks.py內容以下：

# Create your tasks here
from __future__ import absolute_import, unicode_literals

from celery import shared_task


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

這裏使用shared_task，其與app.task區別見這裏

在函數中調用task

views.py見下：

from __future__ import absolute_import, unicode_literals

import json

from django.http import HttpResponse
# Create your views here.
from .tasks import add


def index(request):
    print('1 + 1 = ?')
    r = add.delay(1, 1)
    print('r.get() = %s ' % r.get())
    resp = {'errorcode': 100, 'detail': 'Get success'}
    return HttpResponse(json.dumps(resp), content_type="application/json")

urls.py

from django.contrib import admin
from django.urls import path
from django.conf.urls import url
from demoapp.views import index

urlpatterns = [

    url(r'^$', index, name='index'),
    # url(r'^proj/', include('proj.foo.urls')),
    path('admin/', admin.site.urls),
]

啓動celery worker

celery -A django_celery_beat_test worker --pool=solo -l info

(django_celery_beat_env) e:\code\git\source\my\django_celery_beat_test>celery -A
 django_celery_beat_test worker --pool=solo -l info

 -------------- celery@zq-PC v4.1.0 (latentcall)
---- **** -----
--- * ***  * -- Windows-7-6.1.7601-SP1 2018-02-08 13:19:43
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app:         django_celery_beat_test:0x3974198
- ** ---------- .> transport:   redis://:**@172.20.3.3:6379/0
- ** ---------- .> results:     redis://:**@172.20.3.3:6379/1
- *** --- * --- .> concurrency: 4 (solo)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery


[tasks]
  . demoapp.tasks.add
  . demoapp.tasks.mul
  . demoapp.tasks.xsum
  . django_celery_beat_test.celery.debug_task

注意這裏celery4以上的版本在window上運行時，須要加上--pool=solo，不然在執行任務時會報

ValueError: not enough values to unpack (expected 3, got 0)

啓動server

python runserver 8000

瀏覽器訪問http://127.0.0.1:8000/

看控制檯會打印結果

1 + 1 = ?
[07/Feb/2018 18:37:59] "GET / HTTP/1.1" 200 43
r.get() = 2

表示成功

 

django_celery_beat

接下來看django_celery_beat模塊

上面沒有說celery beat，celery beat就是一個定時模塊，而且包含crontab相似功能，後面是celery worker，能夠說很是強大。

默認調度程序是celery.beat.PersistentScheduler，它只是跟蹤本地shelve數據庫文件中的最後一次運行時間。

但還有一哥們寫的調度程序django_celery_beat，它以數據庫作爲載體，定時任務之類的記錄在庫裏，而且有web admin界面控制。

安裝

1.pip安裝

pip install django-celery-beat

2.添加到INSTALLED_APPS

INSTALLED_APPS = (
    ...,
    'django_celery_beat',
)

3. migrate安裝必要的庫

python manage.py migrate


(django_celery_beat_env) e:\code\git\source\my\django_celery_beat_test>python ma
nage.py migrate
System check identified some issues:

WARNINGS:
?: (mysql.W002) MySQL Strict Mode is not set for database connection 'default'
        HINT: MySQL's Strict Mode fixes many data integrity problems in MySQL, s
uch as data truncation upon insertion, by escalating warnings into errors. It is
 strongly recommended you activate it. See: https://docs.djangoproject.com/en/2.
0/ref/databases/#mysql-sql-mode
Operations to perform:
  Apply all migrations: admin, auth, contenttypes, django_celery_beat, sessions
Running migrations:
  Applying contenttypes.0001_initial... OK
  Applying auth.0001_initial... OK
  Applying admin.0001_initial... OK
  Applying admin.0002_logentry_remove_auto_add... OK
  Applying contenttypes.0002_remove_content_type_name... OK
  Applying auth.0002_alter_permission_name_max_length... OK
  Applying auth.0003_alter_user_email_max_length... OK
  Applying auth.0004_alter_user_username_opts... OK
  Applying auth.0005_alter_user_last_login_null... OK
  Applying auth.0006_require_contenttypes_0002... OK
  Applying auth.0007_alter_validators_add_error_messages... OK
  Applying auth.0008_alter_user_username_max_length... OK
  Applying auth.0009_alter_user_last_name_max_length... OK
  Applying django_celery_beat.0001_initial... OK
  Applying django_celery_beat.0002_auto_20161118_0346... OK
  Applying django_celery_beat.0003_auto_20161209_0049... OK
  Applying django_celery_beat.0004_auto_20170221_0000... OK
  Applying sessions.0001_initial... OK

4. 啓動celery beat的時候指定--scheduler

celery -A django_celery_beat_test beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler

django admin

建立superuser

(django_celery_beat_env) e:\code\git\source\my\django_celery_beat_test>python ma
nage.py createsuperuser
Username (leave blank to use 'zq'): admin
Email address: zq@126.com
Password:
Password (again):
Superuser created successfully.

登陸admin http://127.0.0.1:8000/admin/

以下：

可添加定時，任務等。