Celery

Celery

1.什麼是Celery

Celery是一個簡單、靈活且可靠的，處理大量消息的分佈式系統

專門處理實時處理的異步任務隊列

同時也支持任務調度

2.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 on
        Python ❨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.

3.使用場景

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

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

4.Celery的安裝配置

pip install celery

消息中間件：RabbitMQ/Redis

app = Celery('任務名',backend='xxx',broker='xxx')

5.Celery執行異步任務

基本使用

建立項目celerytest

建立py文件：celery_app_task_.py

import celery
import time
broker = 'redis://127.0.0.1:6379/1' 
#消息中間件
#redis不加密，放到第1個庫中
# broker = 'reids://:123@127.0.0.1:6379/1'  redis加密，密碼是123

#結果存儲
backend = 'reids://127.0.0.1:6379/2'  # redis不加密，放到第二個庫中

# 實例化產生一個Celery對象，必定要指定名字
cel = celery.Celery('test',backend=backend,broker=broker)

#任務其實就是一個函數
#須要用一個裝飾器裝飾，表示該任務是被celery管理的，而且能夠用celery執行的
@cel.task
def add(x,y):
    time.sleep(2)
    return x+y

提交任務的py文件：add_task.py

import celery_task_s1
res = celery_task_s1.add.delay(3,4)
print(res)
# res就是任務的id號

右擊運行提交任務，

啓動worker，使用命令執行：celery worker -A celery_task_s1 -l info

注意：windows下：celery worker -A celery_task_s1 -l info -P eventlet

存儲任務結果的py文件：celery_result.py

from celery.result import AsyncResult
from celery_task_s1 import cel
# 根據id去查詢它的執行結果
async = AsyncResult(id="a5ea035f-0cc3-44ba-b334-f5d7c7ce681d", app=cel)
if async.successful():
    #取出它return的值
    result = async.get()
    print(result)
    # result.forget() # 將結果刪除
elif async.failed():
    print('執行失敗')
elif async.status == 'PENDING':
    print('任務等待中被執行')
elif async.status == 'RETRY':
    print('任務異常後正在重試')
elif async.status == 'STARTED':
    print('任務已經開始被執行')

執行add_task.py，添加任務，並獲取任務id

執行celery_result.py，檢查任務狀態並獲取結果

執行定時任務，3s鍾之後執行add任務

第一種獲取時間的方法

from datetime import datetime
v1 = datetime(2019,7,12,16,15,56)
print(v1)
#默認用UTC時間
v2 = datetime.utcfromtimestamp(v1.timestamp())
print(v2)
result = celery_task_s1.add.apply_async(args=[1,3],eta=v2)
print(result,id)

第二種獲取時間的方法

from datetime import timedelta
ctime = datetime.now()
#默認用utc時間
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
time_delay = timedelta(seconds=3)
task_time = utc_ctime + time_delay
result = celery_task_s1.add.apply_async(args=[2,8],eta=task_time)

多任務結構

# 項目結構
pro_cel
    ├── celery_task# celery相關文件夾
    │   ├── celery.py   # celery鏈接和配置相關文件,必須叫這個名字
    │   └── tasks1.py    #  全部任務函數
    │   └── tasks2.py    #  全部任務函數
    ├── check_result.py # 檢查結果
    └── add_task.py    # 觸發任務

celery.py文件

#必須叫celery，生成celery對象

from celery import Celery
from datetime import timedelta

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
cel = Celery('test',broker=broker,backend=backend,
             include=[
                 'celery_task.order_task',
                 'celery_task.user_task'
             ])

#時區
cel.conf.timezone = 'Asia/shanghai'
# 是否使用UTC
cel.conf.enable_utc = False

order_task.py

from celery_task.celery import cel

@cel.task
def order_add(x,y):
    import time
    time.sleep(2)
    return x+y

user_task.py

from celery_task.celery import cel
@cel.task
def user_add(x,y):
    import time
    time.sleep(2)
    return x+y

add_task.py

from celery_task.order_task import order_add
from celery_task.user_task import user_add


res=order_add.delay(5,6)
print(res.id)
res=user_add.delay(10,60)
print(res.id)

celery_result.py

from celery.result import AsyncResult
from celery_task.celery import cel

async = AsyncResult(id="c8815fd0-c126-4fed-b908-805974761381", app=cel)

if async.successful():
    #取出它return的值
    result = async.get()
    print(result)
    # result.forget() # 將結果刪除
    # async.revoke(terminate=True)  # 不管如今是何時，都要終止
    # async.revoke(terminate=False) # 若是任務尚未開始執行呢，那麼就能夠終止。  
elif async.failed():
    print('執行失敗')
elif async.status == 'PENDING':
    print('任務等待中被執行')
elif async.status == 'RETRY':
    print('任務異常後正在重試')
elif async.status == 'STARTED':
    print('任務已經開始被執行')

添加任務(執行send_task.py)，開啓work：celery worker -A celery_task -l info -P eventlet，檢查任務執行結果（執行check_result.py）

定時任務：多任務結構中celery.py修改以下

from datetime import timedelta
from celery import Celery
from celery.schedules import crontab
broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
cel = Celery('test',broker=broker,backend=backend,
             include=[
                 'celery_task.order_task',
                 'celery_task.user_task'
             ])

#時區
cel.conf.timezone = 'Asia/shanghai'
# 是否使用UTC
cel.conf.enable_utc = False

cel.conf.beat_schedule = {
    # 名字隨意命名
    'add_every_10_seconds':{
        'test':'celery_task.order_task.order_add',
        # 每隔2秒執行一次
        # 'sehedule':1.0,
        # 'schedule':crontab(minute="*/1"),
        'schedule':timedelta(seconds=2),
        # 傳遞參數
        'args':(5,6)
    },
    # 'add-every-12-seconds': {
    #     'task': 'celery_task.tasks1.test_celery',
    #     每一年4月11號，8點42分執行
    #     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),
    #     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),
    #     'args': (16, 16)
    # },
}

啓動一個beat：celery beat -A celery_task -l info

啓動work執行：celery worker -A celery_task -l info -P eventlet

6.django中使用celery

多任務結構的celery_task文件夾直接拷到根目錄下

views.py中

from django.shortcuts import render,HttpResponse

from celery_task.user_task import user_add

def index(request):
    result = user_add.delay(8,9)
    return HttpResponse(result.id)

def check_result(request):
    res = request.GET.get('id')
    from celery.result import AsynResult
    from celery_task.celery import cel
    async = AsynResult(id=res,app=cel)
    if async.sucessful():
        result = async.get()
        print(result)
        return HttpResponse('ok')

注意：上述作完後別忘了啓動worker，再配置路由就能夠了

強調：在celery的任務函數中不能直接調用django的環境(也就是不能直接操做數據庫)，須要手動添加

os.environ.setdefault('DJANGO_SETTINGS_MODEL','untitled15.settings')

import django

django.setup()

補充

pipreq：生成項目依賴的第三方包

項目依賴：pip3 install pipreqs

生成依賴文件：pipreqs ./

安裝依賴文件：pips install -r requirements.txt

方法和函數

他們是有區別的

方法：好比一個類的對象去調用類內部寫的函數就是方法

函數：類直接調用內部的函數就是函數

偏函數：先給函數傳一個值進去

from functions import partial
def test(x,y,z):
    return x+y+z

tsst = partial(test,1)
print(test(2,3))  # 先傳一個值進去，以後再傳只需傳兩個就能夠了

輪詢和長輪詢

輪詢：不停的發請求

長輪詢：hang一下子再發請求