Celery 3 版本 定時執行與 異步執行 | Django 案例

Celery介紹

Celery 是一個 基於python開發的分佈式異步消息任務隊列，經過它能夠輕鬆的實現任務的異步處理， 若是你的業務場景中須要用到異步任務，就能夠考慮使用celery。

軟件架構

Django案例

環境

* python3.6.4
* django 2.0 
* django-celery==3.2.1
* django-kombu==0.9.4
* celery-with-redis==3.0
* celery==3.1.25

目錄結構

autoops/
    autoops/settings
    tasks/tasks.py

settings

import djcelery

INSTALLED_APPS = [
    'djcelery',
    'kombu', 
]

djcelery.setup_loader()
BROKER_URL = 'redis://127.0.0.1:6379/0'  #消息存儲數據存儲在倉庫0

CELERY_RESULT_BACKEND = 'djcelery.backends.database:DatabaseBackend' # 指定 Backend
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

CELERY_TIMEZONE = 'Asia/Shanghai'

#CELERY_ALWAYS_EAGER = True   # 若是開啓，Celery便以eager模式運行, 則task便不須要加delay運行

CELERY_IMPORTS = ('tasks.tasks',)
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'  #這是使用了django-celery默認的數據庫調度模型,任務執行週期都被存在你指定的orm數據庫中

tasks.py

from celery import Celery, platforms

platforms.C_FORCE_ROOT = True

app = Celery('my_task')
app.config_from_object('django.conf:settings',)
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

@app.task()
def  ansbile():   ##若是想異步調用 ansible api,請在任務前面添加以下

    from multiprocessing import current_process
    # try:
    #     current_process()._config
    # except AttributeError:
    current_process()._config = {'semprefix': '/mp'}

@app.task()
def  cmd_job(host,cmd):   ## 執行命令
    i = asset.objects.get(network_ip=host)
    ret = ssh(ip=i.network_ip, port=i.port, username=i.username, password=i.password, cmd=cmd)
    return  ret['data']

def test():  ##  下面是異步調用 celery 的例子

    from tasks.tasks import cmd_job

    aa = cmd_job.apply_async(args=('43.241.238.109', 'pwd'))
    print("id",aa.task_id,"返回值",aa.get() ,aa.result, "狀態",aa.state)

    from  djcelery.models import TaskMeta
    b = TaskMeta.objects.get(task_id=aa).result
    print("返回值",b)

Django 後臺

如下是定時執行，直接後臺操做便可。很簡單。

數據庫結構

* | celery_taskmeta                   ##異步任務，會將結果寫入到這個表內
* | celery_tasksetmeta              
* | djcelery_crontabschedule    
* | djcelery_intervalschedule    
* | djcelery_periodictask           
* | djcelery_periodictasks        
* | djcelery_taskstate               ##django後臺執行的定時任務，會將結果寫到這個表裏
* | djcelery_workerstate

from  djcelery.models import TaskMeta，TaskState          ##這樣獲取表

在數據庫裏看 result 內容是亂碼，可是 經過orm獲取的時候，顯示是正常的。請知悉。

啓動命令

#實際執行任務的程序
/usr/bin/python   /opt/autoops/manage.py   celery worker  -c  4        --loglevel=info

#任務調度, 根據配置文件發佈定時任務
/usr/bin/python   /opt/autoops/manage.py celery beat --schedule=/tmp/celerybeat-schedule --pidfile=/tmp/django_celerybeat.pid --loglevel=INFO

# Django 檢查  workers  是否在線
/usr/bin/python   /opt/autoops/manage.py   celerycam --frequency=10.0