當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

Celery架构

發布時間：2024/10/8 编程问答 23 如意码农

生活随笔收集整理的這篇文章主要介紹了 Celery架构小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

Celery

官方

# Celery 官網：http://www.celeryproject.org/

# Celery 官方文檔英文版：http://docs.celeryproject.org/en/latest/index.html

# Celery 官方文檔中文版：http://docs.jinkan.org/docs/celery/

cerely是什么？

　　cerely被用來稍后執行某些代碼，或者調度器調度這些代碼。

Celery架構

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

消息中間件

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

任務執行單元

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

任務結果存儲

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

總結：

"""

1、celery框架自帶socket，所以自身是一個獨立運行的服務

2、啟動celery服務，是來執行服務中的任務的，服務中帶一個執行任務的對象，會執行準備就緒的任務，將執行任務的結果保存起來

3、celery框架由三部分組成：存放要執行的任務broker，執行任務的對象worker，存放任務結果的backend

4、安裝的celery主體模塊，默認只提供worker，要結合其他技術提供broker和backend（兩個存儲的單位）

"""

工作的基本原理

1、準備配置了broker與backend的worker(任務的來源)，并啟動。

2、添加任務到broker，worker就會執行任務，將結果存儲到backend中。

3、想查看任務的執行結果，根據任務的id去bckend中查詢。

使用場景

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

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

Celery的安裝配置

pip install celery

消息中間件：RabbitMQ/Redis

app=celery.Celery('任務名', broker='xxx', backend='xxx', include=['xxx', 'xxx'])

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 = 'redis://127.0.0.1:6379/1'

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

app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])

from celery import Celery

# 通過Celery功能產生一個celery應用

broker = 'redis://127.0.0.1:6379/14'  # 任務倉庫

backend = 'redis://127.0.0.1:6379/15'  # 結果倉庫

include = ['celery_task.task1', 'celery_task.task2']  # 任務們，完成需求的函數所在的文件

app = Celery(broker=broker, backend=backend, include=include)

# 啟動worker：celery worker -A celery_task -l info -P eventlet

# 手動添加任務

celery內部的源碼init

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)

# 添加延遲任務

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

tasks.low.apply_async(args=(200, 50), eta=eta_second(10))

get_result.py

from celery_task.celery import app

from celery.result import AsyncResult

id = '21325a40-9d32-44b5-a701-9a31cc3c74b5'

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('任務已經開始被執行')

高級使用

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）添加任務：自動添加任務，所以要啟動一個添加任務的服務

# 命令：celery beat -A celery_task -l info

?

# 4）獲取結果

?

?

from celery import Celery

?

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

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

app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])

?

?

# 時區

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),

    }

}

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

get_result.py

from celery_task.celery import app

?

from celery.result import AsyncResult

?

id = '21325a40-9d32-44b5-a701-9a31cc3c74b5'

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('任務已經開始被執行')

django中使用

celery.py

# 重點：要將 項目名.settings 所占的文件夾添加到環境變量

# import sys

# sys.path.append(r'項目絕對路徑')

?

# 開啟django支持

import os

os.environ.setdefault('DJANGO_SETTINGS_MODULE', '項目名.settings')

import django

django.setup()

?

?

?

# 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）添加任務：自動添加任務，所以要啟動一個添加任務的服務

# 命令：celery beat -A celery_task -l info

?

# 4）獲取結果

?

?

from celery import Celery

?

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

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

app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])

?

?

# 時區

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.test_django_celery',

        'schedule': timedelta(seconds=3),

        'args': (),

    }

}

tasks.py

from .celery import app

?

from home.models import Banner

from settings.const import BANNER_COUNT  # 輪播圖最大顯示條數

from home.serializers import BannerModelSerializer

from django.core.cache import cache

@app.task

def update_banner_list():

    # 獲取最新內容

    banner_query = Banner.objects.filter(is_delete=False, is_show=True).order_by('-orders')[:BANNER_COUNT]

    # 序列化

    banner_data = BannerModelSerializer(banner_query, many=True).data

    for banner in banner_data:

        banner['image'] = 'http://127.0.0.1:8000' + banner['image']

    # 更新緩存

    cache.set('banner_list', banner_data)

    return True

celery異步任務提交框架

　　安裝：

pip install celery

使用消息中間件：

RabbitMQ/Redis

創建任務：

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

基本使用

import celery

import time

# broker='redis://127.0.0.1:6379/2' 不加密碼

backend='redis://:123456@127.0.0.1:6379/1'

broker='redis://:123456@127.0.0.1:6379/2'

cel=celery.Celery('test',backend=backend,broker=broker)

@cel.task

def add(x,y):

    return x+y

框架結構

pro_cel

    ├── celery_task# celery相關文件夾

    │   ├── celery.py   # celery連接和配置相關文件,必須叫這個名字

    │   └── tasks1.py    #  所有任務函數

    │    └── tasks2.py    #  所有任務函數

    ├── check_result.py # 檢查結果

    └── send_task.py    # 觸發任務

手動添加任務

from celery_app_task import add

result = add.delay(4,5)

print(result.id)

result查看結果

from celery.result import AsyncResult

from celery_app_task import cel

async = AsyncResult(id="e919d97d-2938-4d0f-9265-fd8237dc2aa3", app=cel)

if async.successful():

    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('任務已經開始被執行')

定時任務

from celery_app_task import add

from datetime import datetime

# 方式一

# v1 = datetime(2019, 2, 13, 18, 19, 56)

# print(v1)

# v2 = datetime.utcfromtimestamp(v1.timestamp())

# print(v2)

# result = add.apply_async(args=[1, 3], eta=v2)

# print(result.id)

# 方式二

ctime = datetime.now()

# 默認用utc時間

utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())

from datetime import timedelta

time_delay = timedelta(seconds=10)

task_time = utc_ctime + time_delay

# 使用apply_async并設定時間

result = add.apply_async(args=[4, 3], eta=task_time)

print(result.id)

循環定時提交任務

from datetime import timedelta

from celery import Celery

from celery.schedules import crontab

cel = Celery('tasks', broker='redis://127.0.0.1:6379/1', backend='redis://127.0.0.1:6379/2', include=[

    'celery_task.tasks1',

    'celery_task.tasks2',

])

cel.conf.timezone = 'Asia/Shanghai'

cel.conf.enable_utc = False

cel.conf.beat_schedule = {

    # 名字隨意命名

    'add-every-10-seconds': {

        # 執行tasks1下的test_celery函數

        'task': 'celery_task.tasks1.test_celery',

        # 每隔2秒執行一次

        # 'schedule': 1.0,

        # 'schedule': crontab(minute="*/1"),

        'schedule': timedelta(seconds=2),

        # 傳遞參數

        'args': ('test',)

    },

    # '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)

    # },

}

總結

以上是生活随笔為你收集整理的Celery架构的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： CVPR 2021 论文和开源项目合集(
下一篇：简述哲学Essay写作

编程问答

Celery架构

Celery

官方

cerely是什么？

Celery架構

消息中間件

任務執行單元

任務結果存儲

總結：

工作的基本原理

使用場景

Celery的安裝配置

Celery執行異步任務

基本使用

celery.py

celery內部的源碼__init__

tasks.py

add_task.py

get_result.py

高級使用

celery.py

tasks.py

get_result.py

django中使用

celery.py

tasks.py

celery異步任務提交框架

基本使用

手動添加任務

result查看結果

定時任務

循環定時提交任務

總結

celery內部的源碼init