Flask + Celery 应用

news2025/6/6 7:10:42

Flask + Celery 应用

对于Flask与Celery结合的例子，需要创建几个文件。首先安装必要的依赖：

pip install flask celery redis requests

项目结构

让我们创建以下文件结构：

flask_celery/
├── app.py          # Flask应用
├── celery_worker.py # Celery worker
├── tasks.py        # Celery任务定义
└── templates/      # 模板目录
    └── index.html  # 主页模板

1. 创建app.py

from flask import Flask, request, render_template, jsonify
from celery_worker import celery
import tasks
import time

app = Flask(__name__)

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/process', methods=['POST'])
def process():
    # 获取表单数据
    data = request.form.get('data', '')
    
    # 启动异步任务
    task = tasks.process_data.delay(data)
    
    return jsonify({
        'task_id': task.id,
        'status': '任务已提交',
        'message': f'正在处理数据: {data}'
    })

@app.route('/status/<task_id>')
def task_status(task_id):
    # 获取任务状态
    task = tasks.process_data.AsyncResult(task_id)
    
    if task.state == 'PENDING':
        response = {
            'state': task.state,
            'status': '任务等待中...'
        }
    elif task.state == 'FAILURE':
        response = {
            'state': task.state,
            'status': '任务失败',
            'error': str(task.info)
        }
    else:
        response = {
            'state': task.state,
            'status': '任务完成' if task.state == 'SUCCESS' else '任务处理中',
            'result': task.result if task.state == 'SUCCESS' else None
        }
    
    return jsonify(response)

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

2. 创建tasks.py

from celery_worker import celery
import time

@celery.task()
def process_data(data):
    # 模拟耗时操作
    time.sleep(5)
    
    # 处理数据（这里只是简单地转换为大写）
    result = data.upper()
    
    return {
        'original_data': data,
        'processed_data': result,
        'processing_time': '5秒'
    }

3. 创建celery_worker.py

from celery import Celery

# 创建Celery实例
celery = Celery(
    'tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/0'
)

# 配置Celery
celery.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='Asia/Shanghai',
    enable_utc=True,
)

4. 创建templates目录和index.html

首先创建templates目录：

mkdir templates

然后创建index.html文件：

<!DOCTYPE html>
<html>
<head>
    <title>Flask + Celery 示例</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 40px; line-height: 1.6; }
        h1 { color: #4285f4; }
        .container { max-width: 800px; margin: 0 auto; }
        .form-group { margin-bottom: 15px; }
        input[type="text"] { padding: 8px; width: 300px; }
        button { padding: 8px 15px; background-color: #4285f4; color: white; border: none; cursor: pointer; }
        button:hover { background-color: #3b78e7; }
        #result { margin-top: 20px; padding: 15px; background-color: #f5f5f5; border-radius: 5px; display: none; }
        #status { margin-top: 10px; font-style: italic; }
    </style>
</head>
<body>
    <div class="container">
        <h1>Flask + Celery 异步任务示例</h1>
        
        <div class="form-group">
            <label for="data">输入要处理的数据:</label><br>
            <input type="text" id="data" name="data" placeholder="输入一些文本...">
            <button onclick="submitTask()">提交任务</button>
        </div>
        
        <div id="status"></div>
        
        <div id="result">
            <h3>任务结果:</h3>
            <pre id="result-data"></pre>
        </div>
    </div>

    <script>
        function submitTask() {
            const data = document.getElementById('data').value;
            if (!data) {
                alert('请输入数据!');
                return;
            }
            
            const statusDiv = document.getElementById('status');
            statusDiv.textContent = '提交任务中...';
            
            // 提交任务
            fetch('/process', {
                method: 'POST',
                headers: {
                    'Content-Type': 'application/x-www-form-urlencoded',
                },
                body: `data=${encodeURIComponent(data)}`
            })
            .then(response => response.json())
            .then(data => {
                statusDiv.textContent = data.status;
                
                // 开始轮询任务状态
                pollTaskStatus(data.task_id);
            })
            .catch(error => {
                statusDiv.textContent = `错误: ${error}`;
            });
        }
        
        function pollTaskStatus(taskId) {
            const statusDiv = document.getElementById('status');
            const resultDiv = document.getElementById('result');
            const resultDataPre = document.getElementById('result-data');
            
            // 定期检查任务状态
            const interval = setInterval(() => {
                fetch(`/status/${taskId}`)
                .then(response => response.json())
                .then(data => {
                    statusDiv.textContent = `状态: ${data.status}`;
                    
                    if (data.state === 'SUCCESS') {
                        clearInterval(interval);
                        resultDiv.style.display = 'block';
                        resultDataPre.textContent = JSON.stringify(data.result, null, 2);
                    } else if (data.state === 'FAILURE') {
                        clearInterval(interval);
                        statusDiv.textContent = `错误: ${data.error}`;
                    }
                })
                .catch(error => {
                    clearInterval(interval);
                    statusDiv.textContent = `轮询错误: ${error}`;
                });
            }, 1000);
        }
    </script>
</body>
</html>

运行应用

首先，确保Redis服务器正在运行（Celery需要它作为消息代理）可以用docker启动：

docker run --name redis-server -p 6379:6379 -d redis

在这里插入图片描述

启动Celery worker：

celery -A tasks worker --loglevel=info

在这里插入图片描述

在另一个终端中启动Flask应用：

python app.py

在这里插入图片描述

测试文件

创建一个新文件test_app.py：

import requests
import time
import json

# 应用服务器地址
BASE_URL = 'http://localhost:5000'

def test_process_endpoint():
    """测试/process端点"""
    print("测试1: 提交任务处理")
    
    # 准备测试数据
    test_data = "hello world"
    
    # 发送POST请求到/process端点
    response = requests.post(
        f"{BASE_URL}/process",
        data={"data": test_data}
    )
    
    # 检查响应状态码
    if response.status_code == 200:
        print("✓ 状态码正确: 200")
    else:
        print(f"✗ 状态码错误: {response.status_code}")
    
    # 解析响应JSON
    result = response.json()
    
    # 检查响应内容
    if 'task_id' in result:
        print(f"✓ 成功获取任务ID: {result['task_id']}")
        return result['task_id']
    else:
        print("✗ 未能获取任务ID")
        return None

def test_status_endpoint(task_id):
    """测试/status/<task_id>端点"""
    print("\n测试2: 检查任务状态")
    
    if not task_id:
        print("✗ 无法测试状态端点: 缺少任务ID")
        return
    
    # 轮询任务状态，最多等待10秒
    max_attempts = 10
    for attempt in range(1, max_attempts + 1):
        print(f"\n轮询 {attempt}/{max_attempts}...")
        
        # 发送GET请求到/status/<task_id>端点
        response = requests.get(f"{BASE_URL}/status/{task_id}")
        
        # 检查响应状态码
        if response.status_code == 200:
            print("✓ 状态码正确: 200")
        else:
            print(f"✗ 状态码错误: {response.status_code}")
            continue
        
        # 解析响应JSON
        result = response.json()
        print(f"当前状态: {result.get('state', '未知')}")
        
        # 如果任务完成，显示结果并退出循环
        if result.get('state') == 'SUCCESS':
            print("\n✓ 任务成功完成!")
            print("结果:")
            print(json.dumps(result.get('result', {}), indent=2, ensure_ascii=False))
            return True
        
        # 如果任务失败，显示错误并退出循环
        if result.get('state') == 'FAILURE':
            print(f"\n✗ 任务失败: {result.get('error', '未知错误')}")
            return False
        
        # 等待1秒后再次检查
        time.sleep(1)
    
    print("\n✗ 超时: 任务未在预期时间内完成")
    return False

def test_index_endpoint():
    """测试首页端点"""
    print("\n测试3: 访问首页")
    
    # 发送GET请求到/端点
    response = requests.get(BASE_URL)
    
    # 检查响应状态码
    if response.status_code == 200:
        print("✓ 状态码正确: 200")
        print("✓ 成功访问首页")
    else:
        print(f"✗ 状态码错误: {response.status_code}")

def run_all_tests():
    """运行所有测试"""
    print("开始测试Flask+Celery应用...\n")
    
    # 测试1: 提交任务
    task_id = test_process_endpoint()
    
    # 测试2: 检查任务状态
    if task_id:
        test_status_endpoint(task_id)
    
    # 测试3: 访问首页
    test_index_endpoint()
    
    print("\n测试完成!")

if __name__ == "__main__":
    run_all_tests()

然后，在另一个终端中运行测试：