目錄 Celery簡介celery的異步任務(wù)1.安裝celery2.安裝redis3.使用ceelryDjango中使用celery1.創(chuàng)建celery文件2.添加celery配置3.在別的應(yīng)用下使用celery執(zhí)行異步任務(wù) [使用celery異步發(fā)送釘釘群消息通知]4.啟動celery服務(wù) Celery簡介

Celery是一個簡單、靈活且可靠的，處理大量消息的分布式系統(tǒng)，專注于實時處理的異步任務(wù)隊列，同時也支持任務(wù)調(diào)度。

Celery的架構(gòu)由三部分組成，消息中間件（message broker），任務(wù)執(zhí)行單元（worker）和任務(wù)執(zhí)行結(jié)果存儲（task result store）組成。

消息中間件：Celery本身不提供消息服務(wù)，但是可以方便的和第三方提供的消息中間件集成。包括，RabbitMQ, Redis等等。

任務(wù)執(zhí)行單元：Worker是Celery提供的任務(wù)執(zhí)行的單元，worker并發(fā)的運(yùn)行在分布式的系統(tǒng)節(jié)點(diǎn)中。

任務(wù)結(jié)果存儲：Task result store用來存儲Worker執(zhí)行的任務(wù)的結(jié)果，Celery支持以不同方式存儲任務(wù)的結(jié)果，包括AMQP, redis等。

版本支持情況：

Celery version 4.0 runs onPython ❨2.7, 3.4, 3.5❩PyPy ❨5.4, 5.5❩ This is the last version to support Python 2.7, and from the next version (Celery 5.x) Python 3.5 or newer is required. If you’re running an older version of Python, you need to be running an older version of Celery:Python 2.6: Celery series 3.1 or earlier.Python 2.5: Celery series 3.0 or earlier.Python 2.4 was Celery series 2.2 or earlier. Celery is a project with minimal funding, so we don’t support Microsoft Windows. Please don’t open any issues related to that platform.

Celery多用來執(zhí)行異步任務(wù)，將耗時的操作交由Celery去異步執(zhí)行，比如發(fā)送郵件、短信、消息推送、音視頻處理等。還可以執(zhí)行定時任務(wù)，定時執(zhí)行某件事情，比如Redis中的數(shù)據(jù)每天凌晨兩點(diǎn)保存至mysql數(shù)據(jù)庫，實現(xiàn)Redis的持久化。

celery的異步任務(wù)

celery的使用

1.安裝celery

$ pip install -U celery

1）安裝相關(guān)依賴

$ pip install 'celery[redis,auth,msgpack]'

序列化程序

celery[auth]

用于使用auth安全序列化程序。

celery[msgpack]

用于使用 msgpack 序列化程序。

celery[redis]

使用 Redis 作為消息傳輸或結(jié)果后端。

2.安裝redis

這里我們使用redis作為celery的broker，作為任務(wù)隊列的存儲和結(jié)果的存儲。

對于 Redis 支持，您必須安裝其他依賴項。您可以使用celery[redis] bundle一次性安裝 Celery 和這些依賴項：

$ pip install -U 'celery[redis]'

1）配置

配置很簡單，只需配置你的 Redis 數(shù)據(jù)庫的位置：

app.conf.broker_url = ’redis://localhost:6379/0’

其中 URL 的格式為：

redis://:password@hostname:port/db_number

方案后面的所有字段都是可選的，并且將默認(rèn)為localhost 端口 6379，使用數(shù)據(jù)庫 0。

3.使用ceelry

1）首先我們可以創(chuàng)建一個celery的文件夾，然后創(chuàng)建一個tasks.py文件

celery/tasks.py

from celery import Celery# 第一個參數(shù)就是當(dāng)前腳本的名稱，backend 任務(wù)執(zhí)行結(jié)果的存儲地址broker 任務(wù)隊列的存儲地址app = Celery(’tasks’, backend=’redis://127.0.0.1’, broker=’redis://127.0.0.1’)@app.taskdef add(x, y): return x + y

celery/run_tasks.py

from tasks import addresult = add.delay(1, 2)print(’Is task ready: %s’ % result.ready()) # False說明任務(wù)還沒有執(zhí)行完run_result = result.get(timeout=1)print(’task result: %s’ % run_result)print(’Is task ready: %s’ % result.ready())

4.啟動celery

$ cd celry$ celery -A tasks worker --loglevel=info

使用flower監(jiān)控celery任務(wù)的執(zhí)行情況

pip install flower

啟動flower，指定我們的應(yīng)用，確保你的celery是啟動的。

cd celerycelery -A tasks flower --broker=redis://@localhost:6379/0

運(yùn)行結(jié)果：

celery [celery args] flower [flower args].[I 210825 10:54:00 command:152] Visit me at http://localhost:5555[I 210825 10:54:00 command:159] Broker: redis://127.0.0.1:6379//[I 210825 10:54:00 command:160] Registered tasks:

我們就可以通過5555端口看到celery異步任務(wù)的運(yùn)行情況了


Django中使用celery

官方地址：https://docs.celeryproject.org/en/latest/django/first-steps-with-django.html

1.創(chuàng)建celery文件

根據(jù)官方文檔的說明，我們可以直接在Django項目同名的應(yīng)用下創(chuàng)建celery.py文件

recruitment/recruitment/celery.py

import osfrom celery import Celery# set the default Django settings module for the ’celery’ program.os.environ.setdefault(’DJANGO_SEttINGS_MODULE’, ’recruitment.base’) # 這里我把配置文件放到了根目錄下的settings/base.py 中app = Celery(’recruitment’)# 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 apps.app.autodiscover_tasks()def debug_task(self): print(f’Request: {self.request!r}’)

然后我們需要在這個celery.py文件所在的目錄的__init__文件中添加：

from __future__ import absolute_import, unicode_literals# This will make sure the app is always imported when/保證所有app下的任務(wù)都能導(dǎo)入進(jìn)來# Django starts so that shared_task will use this app.from .celery import app as celery_app__all__ = (’celery_app’,)2.添加celery配置

settings/base.py

CELERY_BROKER_URL = ’redis://localhost:6379/0’CELERY_RESULT_BACKEND = ’redis://localhost:6379/1’CELERY_ACCEPT_CONTENT = [’application/json’]CELERY_RESULT_SERIALIZER = ’json’CELERY_TASK_SERIALIZER = ’json’CELERY_TIMEZONE = ’Asia/Shanghai’CELERYD_MAX_TASKS_PER_CHILD = 10CELERYD_LOG_FILE = os.path.join(BASE_DIR, 'logs', 'celery_work.log')CELERYBEAT_LOG_FILE = os.path.join(BASE_DIR, 'logs', 'celery_beat.log')3.在別的應(yīng)用下使用celery執(zhí)行異步任務(wù) [使用celery異步發(fā)送釘釘群消息通知]

1.首先我們需要在應(yīng)用下創(chuàng)建一個tasks.py文件interview/tasks.py

from __future__ import absolute_import, unicode_literalsfrom celery import shared_taskfrom .dingtalk import send@shared_taskdef send_dingtalk_message(message): send(message)

interview/dingtalk.py

from dingtalkchatbot.chatbot import DingtalkChatbotfrom django.conf import settingsdef send(message, at_mobiles=[]): # 引用 settings里面配置的釘釘群消息通知的WebHook地址: webhook = settings.DINGTALK_WEB_HOOK # 初始化機(jī)器人小Y， xiaoY = DingtalkChatbot(webhook) # 方式二：勾選“加簽”選項時使用（v1.5以上新功能） # xiaoY = DingtalkChatbot(webhook, secret=secret) # Text消息@所有人 xiaoY.send_text(msg=(’消息通知: %s’ % message), at_mobiles=at_mobiles)

interview.views.py

from interview.tasks import send_dingtalk_messagedef notify_interview(modeladmin, request, queryset): candidates = ’’ interviewers = ’’ for obj in queryset:candidates = obj.userame + ’’ + candidatesinterviewers = obj.first_interviewer_user + ’’ + interviewers # 這里的消息發(fā)送到釘釘， 或者通過 Celery 異步發(fā)送到釘釘 send_dingtalk_message.delay(’候選人 %s 進(jìn)入面試環(huán)節(jié)， 親愛的面試官請做好面試準(zhǔn)備：%s。’ % (candidates, interviewers))4.啟動celery服務(wù)

啟動celery服務(wù)，到我們的項目根目錄啟動，然后執(zhí)行

$ celery -A recruitment worker -l info

如果需要制定配置文件，如果在mac下可以執(zhí)行：

$ DJANGO_SEttINGS_MODULE=settings.base celery --app=recruitment worker --loglevel=info

啟動flower監(jiān)控異步任務(wù)

$ celery -A recruitment flower --broker=redis://localhost:6379/0

celery定時任務(wù)

到此這篇關(guān)于Python中celery的使用的文章就介紹到這了,更多相關(guān)celery的使用內(nèi)容請搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！