Celery簡介以及Django中使用celery

時間  2019-11-12
標籤 celery 簡介 以及 django 使用 欄目 Python 简体版
原文   原文鏈接

目錄python

Celery簡介

Celery是由三部分組成，消息中間件（message broker）、任務執行的單元（worker）和任務執行結果存儲（task result store）。redis

消息中間件

Celery自己不提供消息服務，可是能夠方便的和第三方提供的消息中間件集成。包括，RabbitMQ, Redis等等。django

任務執行單元

Worker是Celery提供的任務執行的單元，worker併發的運行在分佈式的系統節點中。windows

任務結果存儲

Task result store用來存儲Worker執行的任務的結果，Celery支持以不一樣方式存儲任務的結果，包括AMQP, redis等api

使用場景

異步任務：將耗時操做任務提交給Celery去異步執行，好比發送短信/郵件、消息推送、音視頻處理等等緩存

定時任務：定時執行某件事情，好比天天數據統計架構

Celery的安裝和配置

安裝：pip install celery併發

消息中間件：RabbitMQ/Redisapp

app = Celery('任務名'， broker=‘xxx’, bankend='vvv')框架

Celery執行異步任務

基本使用

包架構封裝

project
  |-- celery_task      # celery包
  |   |-- __init__.py  # 包文件
  |   |-- celery.py    # celery鏈接和配置相關文件，且名字必須交celery.py
  |   |-- tasks.py      # 全部任務函數
  |-- add_task.py      # 添加任務
  |-- get_result.py    # 獲取結果

celery.py

# 1）建立app + 任務

# 2）啓動celery(app)服務：
# 非windows
# 命令：celery worker -A celery_task -l info
# windows：
# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet

# 3）添加任務：手動添加，要自定義添加任務的腳本，右鍵執行腳本

# 4）獲取結果：手動獲取，要自定義獲取任務的腳本，右鍵執行腳本


from celery import Celery

"""
broker=存儲tasks的倉庫 
backend=存儲results的倉庫 
include=[任務文件們]
"""

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'

# redis有密碼時
# broker = 'redis://:123456@127.0.0.1:6379/2'
# backend = 'redis://:123456@127.0.0.1:6379/3'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])


# 啓動celery服務的命令：
# 前提：必定要進入celery_task所屬的文件夾
# celery worker -A celery_task -l info -P eventlet

tasks.py

from .celery import app
import time


@app.task
def add(n, m):
    print(n)
    print(m)
    time.sleep(10)
    print('n+m的結果：%s' % (n + m))
    return n + m

@app.task
def low(n, m):
    print(n)
    print(m)
    print('n-m的結果：%s' % (n - m))
    return n - m

add_task.py

from celery_task import tasks

# 添加當即執行任務
t1 = tasks.add.delay(10, 20)
t2 = tasks.low.delay(100, 50)
print(t1.id)

get_result.py

from celery_task.celery import app
from celery.result import AsyncResult

id = '4e249f2d-559a-4a3e-8b43-d498b3d6355e'  # 隨機的任務結果
if __name__ == '__main__':
    async = AsyncResult(id=id, app=app)
    if async.successful():
        result = async.get()
        print(result)
    elif async.failed():
        print('任務失敗')
    elif async.status == 'PENDING':
        print('任務等待中被執行')
    elif async.status == 'RETRY':
        print('任務異常後正在重試')
    elif async.status == 'STARTED':
        print('任務已經開始被執行')

延時任務

包架構封裝

project
  |-- celery_task
  |   |-- __init__.py 
  |   |-- celery.py 
  |   |-- tasks.py
  |-- add_task.py
  |-- get_result.py

celery.py

from celery import Celery

broker = 'redis://127.0.0.1:6379/11'
backend = 'redis://127.0.0.1:6379/12'
include = ['celery_task.tasks',]
app = Celery(broker=broker, backend=backend, include=include)

# 啓動celery服務的命令：
# 前提：必定要進入celery_task所屬的文件夾
# celery worker -A celery_task -l info -P eventlet

tasks.py

from .celery import app

@app.task
def add(n, m):
    res = n + m
    print(res)
    return res

@app.task
def low(n, m):
    res = n - m
    print(res)
    return res

add_task.py

from celery_task.tasks import add, low

# 添加延遲任務
from datetime import datetime, timedelta

def eta_second(second):
    ctime = datetime.now()
    utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
    time_delay = timedelta(seconds=second)
    return utc_ctime + time_delay

# 手動添加 延遲任務
if __name__ == '__main__':
    result = add.apply_async(args=(200, 50), eta=eta_second(10))
    print(result.id)
    result = low.apply_async(args=(200, 50), eta=eta_second(8))
    print(result.id)

get_result.py

from celery_task.celery import app
from celery.result import AsyncResult

id = '74b4f1b2-36ee-4eab-afa7-3da40b7e794e'

if __name__ == '__main__':
    async = AsyncResult(id=id, app=app)
    if async.successful():
        result = async.get()
        print(result)
    elif async.failed():
        print('任務失敗')
    elif async.status == 'PENDING':
        print('任務等待中被執行')
    elif async.status == 'RETRY':
        print('任務異常後正在重試')
    elif async.status == 'STARTED':
        print('任務已經開始被執行')

定時任務

包架構封裝

project
  |-- celery_task
  |   |-- __init__.py
  |   |-- celery.py
  |   |-- tasks.py

celery.py

from celery import Celery

broker = 'redis://127.0.0.1:6379/11'
backend = 'redis://127.0.0.1:6379/12'
include = ['celery_task.tasks',]
app = Celery(broker=broker, backend=backend, include=include)

# 啓動celery服務的命令：
# 前提：必定要進入celery_task所屬的文件夾
# celery worker -A celery_task -l info -P eventlet


# 自動添加任務
# 時區
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 任務的定時配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'low-task': {
        'task': 'celery_task.tasks.low',
        'schedule': timedelta(seconds=3),
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每週一早八點
        'args': (300, 150),
    },
    'my-add-task': {
        'task': 'celery_task.tasks.add',
        'schedule': timedelta(seconds=6),
        'args': (300, 150),
    }
}

# 啓動 添加任務 服務的命令
# celery beat -A celery_task -l info

tasks.py

from .celery import app


@app.task
def add(n, m):
    res = n + m
    print(res)
    return res

@app.task
def low(n, m):
    res = n - m
    print(res)
    return res

異步處理Django任務

包架構封裝

project
  |-- celery_task
  |   |-- __init__.py
  |   |-- celery.py
  |   |-- tasks.py

celery.py

# 啓動django依賴

import sys
sys.path.append(r'C:\Users\oldboy\Desktop\luffy\luffyapi')

import os, django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffyapi.settings.dev')
django.setup()


from celery import Celery

broker = 'redis://127.0.0.1:6379/11'
backend = 'redis://127.0.0.1:6379/12'
include = ['celery_task.tasks',]
app = Celery(broker=broker, backend=backend, include=include)

# 啓動celery服務的命令：
# 前提：必定要進入celery_task所屬的文件夾
# celery worker -A celery_task -l info -P eventlet


# 自動添加任務
# 時區
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 任務的定時配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'django-task': {
        'task': 'celery_task.tasks.django_task',
        'schedule': timedelta(seconds=3),
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每週一早八點
        'args': (300, 150),
    },
}

# 啓動 添加任務 服務的命令
# celery beat -A celery_task -l info

tasks.py

from .celery import app

@app.task
def django_task():
    pass

案例：

celery.py

# 啓動django依賴

# 將celery服務框架放在項目根目錄下
# import sys
# sys.path.append(r'C:\Users\oldboy\Desktop\luffy\luffyapi')

import os, django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffyapi.settings')
django.setup()


from celery import Celery

broker = 'redis://127.0.0.1:6379/11'
backend = 'redis://127.0.0.1:6379/12'
include = ['celery_task.tasks',]
app = Celery(broker=broker, backend=backend, include=include)

app.conf.timezone = 'Asia/Shanghai'

app.conf.enable_utc = False

# 任務的定時配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'django-task': {
        'task': 'celery_task.tasks.django_task',
        'schedule': timedelta(seconds=10),
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每週一早八點
        'args': (),
    },
}

tasks.py

from .celery import app
from django.core.cache import cache
from apps.home import models, serializers


@app.task
def django_task():
    banner_query = models.Banner.objects.filter(is_show=True, is_delete=False).order_by('-orders')
    banner_list_data = serializers.BannerModelSerializer(banner_query, many=True).data
    # 創建接口緩存
    cache.set('api_banner_list_data', banner_list_data)
    return '輪播圖緩存更新完畢'
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程