?

基本步驟

安裝

install redis
yum install -y http://rpms.famillecollet.com/enterprise/remi-release-7.rpm
yum --enablerepo=remi install redis
service redis start
chkconfig redis on
pip3 install celery
pip3 install celery[redis]

?

啟動celery?worker?

manage.py文件同級目錄

celery? -A dj_test? worker -l info?

項目代碼，項目名稱?dj_test

代碼地址?https://github.com/infaaf/dj_test_ops.git? ? tag ：?v1_withcelery

dj_test/dj_test/celery.py?

#!/usr/bin/env python # -*- coding=utf-8 -*- #__author__ = 'infaaf' # refer https://github.com/celery/celery/tree/master/examples/django/ 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', 'dj_test.settings') # change proj name here app = Celery('dj_test') # app = Celery('dj_test', backend='redis://:pwd@127.0.0.1:6379/0', broker='redis://:pwd@127.0.0.1:6379/0')# 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)) View Code

dj_test/dj_test/__init__.py

from __future__ import absolute_import, unicode_literalsimport pymysql pymysql.install_as_MySQLdb()# 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',) View Code

dj_test/dj_test/settings.py

額外增加如下 ### celery CELERY_BROKER_URL = 'redis://:pwd@127.0.0.1:6379/0' CELERY_ACCEPT_CONTENT = ['json'] CELERY_RESULT_BACKEND = 'redis://:pwd@127.0.0.1:6379/0' CELERY_TASK_SERIALIZER = 'json' View Code

dj_test/dj_test/app01/tasks.py

#!/usr/bin/env python # -*- coding=utf-8 -*- # __author__ = 'infaaf'from __future__ import absolute_import, unicode_literals from celery import shared_task@shared_task def add(x, y):return x + y View Code

dj_test/dj_test/app01/views.py? ? ?#?取返回值需要在setting定義 celery_result_backend

from django.shortcuts import render,HttpResponse from .utils.simple_allinone_ansible_runner import Runner as AnsibleRunner from .tasks import add# Create your views here. import logging logger=logging.getLogger(__name__)def index(request):r=add.delay(3,4)print(r.ready())import timetime.sleep(1)print(r.ready())print(r.get(timeout=3))return HttpResponse('123') View Code

日志信息

django日志

System check identified no issues (0 silenced). August 26, 2018 - 15:12:18 Django version 2.1, using settings 'dj_test.settings' Starting development server at http://192.168.188.200:8100/ Quit the server with CONTROL-C. False 調用瞬間并未完成 20180826 15:12:21 [INFO] basehttp/basehttp.py[line:124] "GET / HTTP/1.1" 200 3 True 7 View Code

celery?日志

[2018-08-26 15:12:20,523: INFO/MainProcess] Received task: app01.tasks.add[be1ae35e-d104-4015-a0fa-c4316221611d] [2018-08-26 15:12:20,526: INFO/ForkPoolWorker-4] Task app01.tasks.add[be1ae35e-d104-4015-a0fa-c4316221611d] succeeded in 0.0008414150070166215s: 7 View Code

redis?存儲

127.0.0.1:6379> keys *1) "_kombu.binding.celery.pidbox"2) "celery-task-meta-8f64c6a8-7b65-42df-b556-819991417aeb"3) "a"4) "_kombu.binding.celeryev"5) "celery-task-meta-5f6d5a5e-8782-4535-bc24-48d109939634"6) "celery-task-meta-55e1a767-38a0-4222-b173-f9ff4f4f4e05"7) "celery-task-meta-6dcc4017-6287-4f53-9ffc-51cb11073bba"8) "celery-task-meta-cd34ade8-5afc-4c6b-814e-3eb65c72d25a"9) "celery-task-meta-d8e8c21d-5df2-4141-b8ee-250fd3302301" 10) "celery-task-meta-be1ae35e-d104-4015-a0fa-c4316221611d" 11) "_kombu.binding.celery" 12) "unacked_mutex" 127.0.0.1:6379> 127.0.0.1:6379> get celery-task-meta-be1ae35e-d104-4015-a0fa-c4316221611d "{\"status\": \"SUCCESS\", \"result\": 7, \"traceback\": null, \"children\": [], \"task_id\": \"be1ae35e-d104-4015-a0fa-c4316221611d\"}" 127.0.0.1:6379> View Code

?

celery?ansible

celery multi start w1 -A dj_test -l info --logfile=celerylog.log --pidfile=celerypid.pid??

pkill -9 -f 'celery worker'

tasks文件案例

@shared_task() def ansible_test(host_data,tasks):current_process()._config = {'semprefix': '/mp'}inventory = BaseInventory(host_data)runner = AdHocRunner(inventory)ret = runner.run(tasks, "all")print(ret.results_summary)print(ret.results_raw)return 'a' View Code

?

額外選項

結果記錄到django?db

pip3 install django-celery-resultsINSTALLED_APPS = (...,'django_celery_results', ) CELERY_RESULT_BACKEND = 'django-db' #CELERY_RESULT_BACKEND = 'django-cache' $ python manage.py migrate django_celery_results View Code

?

轉載于:https://www.cnblogs.com/infaaf/p/9535167.html

總結

以上是生活随笔為你收集整理的django celery的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。