Python中celery的使用
Celery是一個簡單、靈活且可靠的，處理大量消息的分布式系統，專注于實時處理的異步任務隊列，同時也支持任務調度。
Celery的架構由三部分組成，消息中間件（message broker），任務執行單元（worker）和任務執行結果存儲（task result store）組成。
消息中間件：Celery本身不提供消息服務，但是可以方便的和第三方提供的消息中間件集成。包括，RabbitMQ, Redis等等。
任務執行單元：Worker是Celery提供的任務執行的單元，worker并發的運行在分布式的系統節點中。
任務結果存儲：Task result store用來存儲Worker執行的任務的結果，Celery支持以不同方式存儲任務的結果，包括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多用來執行異步任務，將耗時的操作交由Celery去異步執行，比如發送郵件、短信、消息推送、音視頻處理等。還可以執行定時任務，定時執行某件事情，比如Redis中的數據每天凌晨兩點保存至mysql數據庫，實現Redis的持久化。
celery的異步任務celery的使用
1.安裝celery$ pip install -U celery
1）安裝相關依賴
$ pip install 'celery[redis,auth,msgpack]'
序列化程序
celery[auth]
用于使用auth安全序列化程序。
celery[msgpack]
用于使用 msgpack 序列化程序。
celery[redis]
使用 Redis 作為消息傳輸或結果后端。
2.安裝redis這里我們使用redis作為celery的broker，作為任務隊列的存儲和結果的存儲。
對于 Redis 支持，您必須安裝其他依賴項。您可以使用celery[redis] bundle一次性安裝 Celery 和這些依賴項：
$ pip install -U 'celery[redis]'
1）配置
配置很簡單，只需配置你的 Redis 數據庫的位置：
app.conf.broker_url = ’redis://localhost:6379/0’
其中 URL 的格式為：
redis://:password@hostname:port/db_number
方案后面的所有字段都是可選的，并且將默認為localhost 端口 6379，使用數據庫 0。
3.使用ceelry1）首先我們可以創建一個celery的文件夾，然后創建一個tasks.py文件
celery/tasks.py
from celery import Celery# 第一個參數就是當前腳本的名稱，backend 任務執行結果的存儲地址broker 任務隊列的存儲地址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說明任務還沒有執行完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監控celery任務的執行情況
pip install flower
啟動flower，指定我們的應用，確保你的celery是啟動的。
cd celerycelery -A tasks flower --broker=redis://@localhost:6379/0
運行結果：
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異步任務的運行情況了
Django中使用celery官方地址：https://docs.celeryproject.org/en/latest/django/first-steps-with-django.html
1.創建celery文件根據官方文檔的說明，我們可以直接在Django項目同名的應用下創建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下的任務都能導入進來# 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.在別的應用下使用celery執行異步任務 [使用celery異步發送釘釘群消息通知]
1.首先我們需要在應用下創建一個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 # 初始化機器人小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 # 這里的消息發送到釘釘， 或者通過 Celery 異步發送到釘釘 send_dingtalk_message.delay(’候選人 %s 進入面試環節， 親愛的面試官請做好面試準備：%s。’ % (candidates, interviewers))4.啟動celery服務
啟動celery服務，到我們的項目根目錄啟動，然后執行
$ celery -A recruitment worker -l info
如果需要制定配置文件，如果在mac下可以執行：
$ DJANGO_SEttINGS_MODULE=settings.base celery --app=recruitment worker --loglevel=info
啟動flower監控異步任務
$ celery -A recruitment flower --broker=redis://localhost:6379/0
celery定時任務
到此這篇關于Python中celery的使用的文章就介紹到這了,更多相關celery的使用內容請搜索好吧啦網以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持好吧啦網！
相關文章：
網公網安備