Django的Celery與Flask的異步任務(wù)處理

Django 是一個基于 Python 的高級 Web 開發(fā)框架，它鼓勵快速開發(fā)和干凈、實用的設(shè)計。Celery 是一個分布式任務(wù)隊列，用于處理異步任務(wù)和消息傳遞。Flask 是一個輕量級的 Web 開發(fā)框架，用于構(gòu)建 Web 應(yīng)用程序。

將 Django 與 Celery 結(jié)合使用，可以在 Django 項目中輕松地處理異步任務(wù)。以下是如何在 Django 項目中使用 Celery 進(jìn)行異步任務(wù)處理的步驟：

1. 安裝 Celery 和 Redis（或其他消息代理）：

pip install celery redis

2. 在 Django 項目的根目錄下創(chuàng)建一個名為 celery.py 的文件，用于配置 Celery：

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

# 設(shè)置默認(rèn)的 Django 設(shè)置模塊
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project_name.settings')

app = Celery('your_project_name')

# 使用字符串來配置 Celery，這樣 Celery 就不會在啟動時查找模塊
app.config_from_object('django.conf:settings', namespace='CELERY')

# 自動發(fā)現(xiàn)并加載任務(wù)模塊
app.autodiscover_tasks()

3. 在 Django 項目的 settings.py 文件中，添加 Celery 配置：

# ...

# Celery 配置
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'UTC'

4. 在 Django 應(yīng)用中創(chuàng)建一個名為 tasks.py 的文件，用于定義異步任務(wù)：

from celery import shared_task

@shared_task
def my_async_task(param1, param2):
    # 在這里執(zhí)行異步任務(wù)
    pass

5. 在需要執(zhí)行異步任務(wù)的地方，調(diào)用 my_async_task：

from .tasks import my_async_task

my_async_task.delay(param1, param2)

6. 運(yùn)行 Celery worker：

celery -A your_project_name worker --loglevel=info

現(xiàn)在，當(dāng)你在 Django 項目中調(diào)用 my_async_task.delay() 時，Celery 會將任務(wù)添加到消息隊列中，并在后臺異步執(zhí)行。

Flask 與 Celery 的結(jié)合使用類似。首先，你需要在 Flask 應(yīng)用中配置 Celery，然后定義異步任務(wù)并在需要時調(diào)用它們。這里是一個簡單的例子：

1. 安裝 Flask 和 Celery：

pip install flask celery redis

2. 創(chuàng)建一個名為 app.py 的 Flask 應(yīng)用文件：

from flask import Flask
from celery import Celery

app = Flask(__name__)

# 配置 Celery
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

@celery.task()
def my_async_task(param1, param2):
    # 在這里執(zhí)行異步任務(wù)
    pass

@app.route('/')
def index():
    my_async_task.delay(param1, param2)
    return 'Async task started.'

if __name__ == '__main__':
    app.run()

3. 運(yùn)行 Flask 應(yīng)用：

python app.py

現(xiàn)在，當(dāng)用戶訪問應(yīng)用的根 URL 時，異步任務(wù)將被添加到消息隊列中并在后臺執(zhí)行。