# Celery方式：立即返回，后台处理
@app.task
def generate_article(task_id):
    # 耗时操作
    return result

# 调用方式（立即返回）
task = generate_article.delay(task_id)
return {"task_id": task.id, "status": "processing"}

# 直接拉取方式：同步阻塞
def generate_article_sync(task_id):
    # 耗时操作（阻塞主线程）
    result = call_ollama_api()  # 可能需要几十秒
    return result

# 调用方式（必须等待）
result = generate_article_sync(task_id)  # Web请求会阻塞几十秒
return result  # 用户需要等待

# Celery可以轻松扩展到多台机器
# Worker1 (GPU服务器1)
celery -A tasks_ollama worker -Q ollama_queue -c 2 -n worker1@%h

# Worker2 (GPU服务器2)  
celery -A tasks_ollama worker -Q ollama_queue -c 2 -n worker2@%h

# Worker3 (普通服务器)
celery -A tasks_ollama worker -Q ollama_queue -c 1 -n worker3@%h

# Redis/RabbitMQ确保任务不丢失
# 即使Worker崩溃，任务仍在队列中
# Worker重启后会继续处理

# Celery内置重试机制
@app.task(bind=True, max_retries=3)
def process_task(self, task_id):
    try:
        result = call_external_api()
        return result
    except ExternalAPIError as exc:
        # 自动重试
        raise self.retry(exc=exc, countdown=60)  # 60秒后重试

# 使用Flower监控Celery
celery -A tasks_ollama flower --port=5555

# 可查看：
# - 实时任务状态
# - Worker健康状况
# - 队列长度
# - 任务历史

# 单个Python进程处理所有任务
while True:
    tasks = get_pending_tasks_from_mongodb()  # 频繁查询数据库
    
    for task in tasks:
        # 如果这里崩溃，所有任务都受影响
        result = process_task(task)  # 阻塞操作
        update_mongodb_task_status(task, result)
    
    time.sleep(1)  # 简单轮询

# 直接拉取方案的数据库压力：
# 1. 频繁查询：SELECT * FROM tasks WHERE status='pending'
# 2. 并发更新：UPDATE tasks SET status='processing'
# 3. 结果写入：UPDATE tasks SET result=...

# 高并发时容易导致数据库锁、连接池耗尽

# Web服务 + 任务处理 共享资源
# 当任务处理占用大量CPU/内存时：
# - Web响应变慢
# - API超时
# - 整体服务质量下降

# 要扩展处理能力：
# 1. 复制整个应用（包括Web服务）
# 2. 需要处理任务分配逻辑
# 3. 需要防止重复处理

# 而Celery只需增加Worker节点：
celery -A tasks worker --concurrency=10

# 需要生成1000篇文章
task_ids = []
for i in range(1000):
    task_id = create_ollama_task(f"生成文章{i}")
    process_ollama_task.delay(task_id)  # 立即返回，后台处理
    task_ids.append(task_id)

# 用户可以立即获得所有task_id
# 系统会按队列顺序处理

# 生成一篇长文章需要3分钟
# Celery：API立即响应，用户可轮询状态
# 直接拉取：HTTP连接保持3分钟（易超时）

# 可以设置任务优先级
process_urgent_task.apply_async(
    args=[task_id], 
    queue='high_priority',
    priority=0  # 最高优先级
)

process_normal_task.apply_async(
    args=[task_id],
    queue='default',
    priority=5  # 普通优先级
)

# 每天晚上自动生成日报
@app.task
def generate_daily_report():
    # 自动执行
    pass

# 配置定时
app.conf.beat_schedule = {
    'daily-report': {
        'task': 'tasks.generate_daily_report',
        'schedule': crontab(hour=23, minute=30),  # 每天23:30
    },
}

# 简单实现示例
def simple_task_processor():
    """简单任务处理器（适合小规模）"""
    while True:
        # 获取待处理任务
        task = db.tasks.find_one_and_update(
            {"status": "pending"},
            {"$set": {"status": "processing"}},
            sort=[("priority", -1), ("createdAt", 1)]
        )
        
        if not task:
            time.sleep(5)  # 没有任务时休眠
            continue
        
        try:
            # 处理任务
            result = process_task(task)
            
            # 更新结果
            db.tasks.update_one(
                {"_id": task["_id"]},
                {"$set": {"status": "completed", "result": result}}
            )
        except Exception as e:
            # 标记失败
            db.tasks.update_one(
                {"_id": task["_id"]},
                {"$set": {"status": "failed", "error": str(e)}}
            )

# hybrid_solution.py

import threading
from queue import Queue
from concurrent.futures import ThreadPoolExecutor

class HybridTaskProcessor:
    """混合任务处理器：简单队列 + 线程池"""
    
    def __init__(self, max_workers=4):
        self.task_queue = Queue()
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.running = False
        
    def start(self):
        """启动处理器"""
        self.running = True
        self.worker_thread = threading.Thread(target=self._worker_loop)
        self.worker_thread.daemon = True
        self.worker_thread.start()
        
    def submit_task(self, task_id, task_func):
        """提交任务（立即返回）"""
        future = self.executor.submit(task_func, task_id)
        return future
        
    def _worker_loop(self):
        """后台工作循环"""
        while self.running:
            # 检查MongoDB中的新任务
            tasks = task_manager.get_pending_tasks(limit=5)
            
            for task in tasks:
                task_id = str(task["_id"])
                task_manager.start_task(task_id)
                
                # 异步提交到线程池
                self.submit_task(task_id, process_ollama_task_local)
                
            time.sleep(1)  # 1秒间隔
    
    def shutdown(self):
        """关闭处理器"""
        self.running = False
        self.executor.shutdown()

def process_ollama_task_local(task_id):
    """本地处理Ollama任务"""
    task = task_manager.get_task(task_id)
    result = _process_ollama_article(task_id, task)
    task_manager.complete_task(task_id, result)
    return result