Django的Celery與Flask的異步任務處理

Django 是一個基于 Python 的高級 Web 開發框架，它鼓勵快速開發和干凈、實用的設計。Celery 是一個分布式任務隊列，用于處理異步任務和消息傳遞。Flask 是一個輕量級的 Web 開發框架，用于構建 Web 應用程序。

將 Django 與 Celery 結合使用，可以在 Django 項目中輕松地處理異步任務。以下是如何在 Django 項目中使用 Celery 進行異步任務處理的步驟：

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

pip install celery redis

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

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

# 設置默認的 Django 設置模塊
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')

# 自動發現并加載任務模塊
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 應用中創建一個名為 tasks.py 的文件，用于定義異步任務：

from celery import shared_task

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

5. 在需要執行異步任務的地方，調用 my_async_task：

from .tasks import my_async_task

my_async_task.delay(param1, param2)

6. 運行 Celery worker：

celery -A your_project_name worker --loglevel=info

現在，當你在 Django 項目中調用 my_async_task.delay() 時，Celery 會將任務添加到消息隊列中，并在后臺異步執行。

Flask 與 Celery 的結合使用類似。首先，你需要在 Flask 應用中配置 Celery，然后定義異步任務并在需要時調用它們。這里是一個簡單的例子：

1. 安裝 Flask 和 Celery：

pip install flask celery redis

2. 創建一個名為 app.py 的 Flask 應用文件：

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):
    # 在這里執行異步任務
    pass

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

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

3. 運行 Flask 應用：

python app.py

現在，當用戶訪問應用的根 URL 時，異步任務將被添加到消息隊列中并在后臺執行。