異步任務(wù)介紹
在寫項(xiàng)目過(guò)程中經(jīng)常會(huì)遇到一些耗時(shí)的任務(wù), 比如:發(fā)送郵件席楚、發(fā)送短信等等~城菊。這些操作如果都同步執(zhí)行耗時(shí)長(zhǎng)對(duì)用戶體驗(yàn)不友好,在這種情況下就可以把任務(wù)放在后臺(tái)異步執(zhí)行
celery就是用于處理異步任務(wù)的框架,celery能完成的功能遠(yuǎn)不止異步任務(wù),還有一個(gè)很常用的功能定時(shí)任務(wù)
架構(gòu)圖
Celery包含如下組件：
Celery Beat：任務(wù)調(diào)度器撕彤，Beat進(jìn)程會(huì)讀取配置文件的內(nèi)容，周期性地將配置中到期需要執(zhí)行的任務(wù)發(fā)送給任務(wù)隊(duì)列场仲。Celery Worker：執(zhí)行任務(wù)的消費(fèi)者法竞，通常會(huì)在多臺(tái)服務(wù)器運(yùn)行多個(gè)消費(fèi)者來(lái)提高執(zhí)行效率。Broker：消息代理鹉动，或者叫作消息中間件轧坎，接受任務(wù)生產(chǎn)者發(fā)送過(guò)來(lái)的任務(wù)消息，存進(jìn)隊(duì)列再按序分發(fā)給任務(wù)消費(fèi)方（通常是消息隊(duì)列或者數(shù)據(jù)庫(kù)）泽示。Producer：調(diào)用了Celery提供的API缸血、函數(shù)或者裝飾器而產(chǎn)生任務(wù)并交給任務(wù)隊(duì)列處理的都是任務(wù)生產(chǎn)者。Result Backend：任務(wù)處理完后保存狀態(tài)信息和結(jié)果械筛，以供查詢捎泻。Celery默認(rèn)已支持Redis、RabbitMQ埋哟、MongoDB笆豁、Django ORM、SQLAlchemy等方式赤赊。
Celery 安裝
pip install django-celery celery-with-redis
項(xiàng)目結(jié)構(gòu)
[vagrant@reboot test_drf]$ tree opsweb/
opsweb/
├── apps
│ ├── account
│ │ ├── admin.py
│ │ ├── apps.py
│ │ ├── __init__.py
│ │ ├── migrations
│ │ ├── models.py
│ │ ├── __pycache__
│ │ ├── tasks.py
│ │ ├── tests.py
│ │ └── views.py
├── celerybeat.pid
├── logs
│ ├── celery
│ │ ├── beat.log
│ │ └── celery.log
├── manage.py
├── opsweb
│ ├── celery.py
│ ├── __init__.py
│ ├── __pycache__
│ ├── settings.py
│ ├── urls.py
│ └── wsgi.py
加載celery app(settings文件中)
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'corsheaders',
'rest_framework',
'rest_framework_swagger',
'account',
'djcelery' # celery app 很強(qiáng)大
]
添加Celery全局配置(settings文件中)
# Celery
import djcelery
djcelery.setup_loader() # 加載djcelery
CELERY_TIMEZONE = TIME_ZONE
CELERY_ENABLE_UTC = True
# 允許的格式
CELERY_ACCEPT_CONTENT = ['pickle', 'json', 'yaml']
BROKER_URL = 'redis://127.0.0.1:6379/0' # redis作為中間件
BROKER_TRANSPORT = 'redis'
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler' # Backend數(shù)據(jù)庫(kù)
# CELERYD_LOG_FILE = BASE_DIR + "/logs/celery/celery.log" # log路徑
# CELERYBEAT_LOG_FILE = BASE_DIR + "/logs/celery/beat.log" # beat log路徑
同步Celery表到數(shù)據(jù)庫(kù)
python manage.py migrate
創(chuàng)建celery.py文件(與settings同級(jí))
import os
import django
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', 'opsweb.settings')
django.setup()
app = Celery('opsweb')
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))
創(chuàng)建任務(wù)文件
在需要使用異步任務(wù)的app中創(chuàng)建tasks.py,寫入對(duì)應(yīng)的任務(wù)函數(shù),博主喜歡把tasks放在對(duì)應(yīng)的app下,其實(shí)放在其他目錄下也可以的,看個(gè)人習(xí)慣
from opsweb.celery import app
from celery.schedules import crontab
import traceback
from django.contrib.auth.models import User
import os
@app.task(name="create_user")
def useradd(username):
try:
print(username)
except:
print('fail')
traceback.print_exc()
觸發(fā)任務(wù)
在對(duì)應(yīng)的視圖中導(dǎo)入tasks中的任務(wù)函數(shù)調(diào)用即可
from account.tasks import useradd
# 調(diào)用異步任務(wù)函數(shù)
useradd.delay('username')
啟動(dòng)Celery
進(jìn)入opsweb工程下,啟動(dòng)Celery
[vagrant@reboot opsweb]$celery -A opsweb worker -B -l info
或：
[vagrant@reboot opsweb]$python manage.py celery worker -B -l info
備注:
本場(chǎng)景用戶訪問(wèn)觸發(fā)任務(wù)闯狱，流程如下：
用戶頁(yè)面上點(diǎn)擊事件->調(diào)用任務(wù)/定時(shí)計(jì)劃任務(wù)->任務(wù)進(jìn)入redis隊(duì)列
->如果celery啟動(dòng)則依次執(zhí)行任務(wù)->如果celery沒(méi)啟動(dòng)，則會(huì)存到redis
隊(duì)列里抛计，一旦啟動(dòng)就依次執(zhí)行
啟動(dòng)Django
[vagrant@reboot opsweb]$python manage.py runserver 0:8000
測(cè)試
頁(yè)面上觸發(fā)了異步任務(wù)就會(huì)在celery日志里看到任務(wù)信息,我這里只是寫了簡(jiǎn)單的任務(wù)例子
[2018-09-01 23:56:59,704: WARNING/Worker-2] hello
[2018-09-01 23:56:59,707: INFO/MainProcess] Received task: create_user[c9724e23-b9ba-44fc-b195-6b1153d2c161]
[2018-09-01 23:56:59,708: INFO/MainProcess] Task create_user[f3b3e644-b8aa-4679-8a42-0efc2574abf6] succeeded in 0.0038937819999773637s: None
計(jì)劃任務(wù) - djcelery
djcelery app提供了定時(shí)任務(wù)的功能哄孤，注冊(cè)并同步到數(shù)據(jù)庫(kù)之后，會(huì)生產(chǎn)五個(gè)表吹截，結(jié)構(gòu)如下：
MariaDB [test002]> show tables
-> ;
+---------------------------+
| Tables_in_test002 |
+---------------------------+
...
| djcelery_crontabschedule |
| djcelery_intervalschedule |
| djcelery_periodictask |
| djcelery_periodictasks |
| djcelery_taskstate |
| djcelery_workerstate |
+---------------------------+
在django后臺(tái)可以看到注冊(cè)的表
每個(gè)任務(wù)可以接受參數(shù)
定時(shí)任務(wù)函數(shù)
from opsweb.celery import app
@app.task(name="create_user" )
def create_user(*args,**kwargs): # 分別接收后臺(tái)傳入的列表和字典參數(shù)
print (args,kwargs)
啟動(dòng)Celery beat
[vagrant@reboot opsweb]$ python manage.py celery beat # 啟動(dòng)定時(shí)任務(wù)
Celery會(huì)通過(guò)celery beat進(jìn)程來(lái)完成. Celerybeat會(huì)保持運(yùn)行, 一旦到了某一定期任務(wù)需要執(zhí)行時(shí), Celery beat便將其加入到queue中
supervisor管理Celery任務(wù)
配置如下
- 主動(dòng)觸發(fā)任務(wù)
celery_worker.conf
[program:celery_worker]
# 進(jìn)入工作目錄
directory=/vagrant/test_drf/opsweb
# 執(zhí)行celery指令
command=python manage.py celery worker -B -l info
autorestart=true
loglevel=info
redirect_stderr=true
stdout_logfile=/var/log/supervisor/celery_worker.log
- 定時(shí)任務(wù)觸發(fā)
celery_beat.conf 同上, 區(qū)別如下:
[program:celery_beat]
command=python manage.py celery beat
stdout_logfile=/var/log/supervisor/celery_beat.log
一些不錯(cuò)的文章
http://www.cnblogs.com/znicy/p/5626040.html Django中使用celery瘦陈，非常經(jīng)典
https://www.cnblogs.com/huangxiaoxue/p/7266253.html 基于celery開(kāi)發(fā)的平臺(tái),非常棒
http://python.jobbole.com/81953/ 劉天斯大神 經(jīng)典案例
http://www.imooc.com/article/16164 慕課網(wǎng)老師博客
http://blog.csdn.net/michael_lbs/article/details/74923367 python+django+djcelery
http://blog.csdn.net/lizhihua0925/article/details/53842110 model api
http://www.reibang.com/p/7085dcc70d0e 挺好
http://blog.csdn.net/crb912/article/details/78344659 很詳細(xì)
