import time
from celery import Celery

# 1. 创建 Celery 实例
# 第一个参数 'tasks' 是当前模块的名称，这在自动生成任务名时很重要。
# broker 和 backend 都指向我们的 Redis 服务。
# /0 和 /1 代表 Redis 的不同数据库 (db)，避免数据混淆。
app = Celery('tasks',
             broker='redis://localhost:6379/0',
             backend='redis://localhost:6379/1')

# 2. 定义一个任务
# @app.task 装饰器将这个函数注册为 Celery 任务
@app.task
def add(x, y):
    print(f"开始执行任务: {x} + {y}")
    time.sleep(5)  # 模拟一个耗时的操作
    result = x + y
    print(f"任务执行完毕: {result}")
    return result

@app.task
def send_email(to, subject, body):
    print(f"向 {to} 发送邮件...")
    time.sleep(3) # 模拟I/O操作
    print(f"邮件发送成功: 主题 {subject}")
    return "Success"

# -A tasks: 指定 Celery 实例 (app) 在 'tasks' 模块中
# worker: 启动 worker 进程
# --loglevel=info: 设置日志级别为 INFO，方便我们观察
celery -A tasks worker --loglevel=info

>>> from tasks import add, send_email

# 1. 异步调用任务
# .delay() 是 .apply_async() 的快捷方式
# 注意：这行代码会 *立即* 返回，不会等待 5 秒
print("准备调用 add(4, 4)...")
task_result = add.delay(4, 4)
print("任务已发送，ID:", task_result.id)

# 2. 尝试调用另一个任务
print("准备发送邮件...")
email_task = send_email.delay("user@example.com", "你好", "这是一封测试邮件。")
print("邮件任务已发送，ID:", email_task.id)

# 3. 查看 Worker 终端
# 此时，你应该能在第一个终端（Worker 终端）看到任务开始执行的日志。

>>> # 检查任务是否完成
>>> task_result.ready()
False  # (可能还是 False，因为任务在 sleep(5))

# (等待几秒钟)

>>> task_result.ready()
True

>>> # 获取任务状态
>>> task_result.status
'SUCCESS'

>>> # 获取任务的返回值
>>> # .get() 会阻塞当前进程，直到拿到结果
>>> result_value = task_result.get()
>>> print(result_value)
8

>>> # 检查邮件任务
>>> email_task.get()
'Success'

>>> # 如果任务执行出错
>>> # 你可以通过 .status 查看
# 假设我们定义了一个出错的任务
# task_error = error_task.delay()
# task_error.status
# 'FAILURE'
# task_error.get() # 这将会抛出异常

my_fastapi_project/
├── main.py          # FastAPI 应用
├── celery_app.py    # Celery 实例和配置
├── tasks.py         # 任务定义
└── requirements.txt

fastapi==0.104.1
uvicorn[standard]==0.24.0
celery==5.3.6
redis==5.0.1

# celery_app.py
from celery import Celery

# 创建 Celery 实例
celery = Celery(__name__)

# 从配置中加载 Celery 配置
def init_celery(app, config):
    celery.conf.update(config)

    # (可选) 让任务可以使用 FastAPI 的 app context（如果任务需要访问 app 状态）
    # FastAPI 没有传统的 app_context，但我们可以用类似方式处理依赖
    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            # 这里可以注入 FastAPI 的依赖，如果需要
            # 对于简单任务，可省略
            return self.run(*args, **kwargs)

    celery.Task = ContextTask

# tasks.py
import time
from celery_app import celery

# 注意：我们现在从 celery_app 导入 celery 实例
@celery.task
def long_running_simulation(user_id):
    print(f"开始为用户 {user_id} 运行模拟...")
    time.sleep(10)  # 模拟一个非常耗时的操作
    print(f"用户 {user_id} 的模拟运行完毕。")
    return f"Report for {user_id}"

# main.py
from fastapi import FastAPI, Query
from celery_app import celery, init_celery
from tasks import long_running_simulation

app = FastAPI()

# 配置（这里直接定义，也可以用环境变量或 pydantic-settings）
config = {
    # Celery Broker
    "CELERY_BROKER_URL": "redis://localhost:6379/0",
    # Celery Backend
    "CELERY_RESULT_BACKEND": "redis://localhost:6379/1"
}

# 初始化 Celery
init_celery(app, config)

@app.get("/")
def index():
    return {"message": "你好, Celery 和 FastAPI!"}

@app.get("/start-task")
def start_task(user_id: str = Query("default_user", description="用户 ID")):
    # 这是一个 Web 请求，我们不想让它等待 10 秒
    # 我们立即将任务推送到队列
    task = long_running_simulation.delay(user_id)

    # 立即返回，并告诉前端任务 ID
    return {
        "message": "任务已启动",
        "task_id": task.id
    }, 202  # 202 Accepted 表示请求已被接受，正在处理

@app.get("/task-status/{task_id}")
def task_status(task_id: str):
    # 从 Backend 查询任务状态
    task_result = long_running_simulation.AsyncResult(task_id)

    if task_result.state == "PENDING":
        # 任务还在队列中，或刚开始
        response = {
            "state": task_result.state,
            "status": "Pending..."
        }
    elif task_result.state != "FAILURE":
        response = {
            "state": task_result.state,
            "status": "In progress...",
        }
        if task_result.state == "SUCCESS":
            response["result"] = task_result.get()  # 获取最终结果
        else:
            response["result"] = task_result.info  # 如果任务正在进行，.info 可能包含中间状态
    else:
        # 任务失败
        response = {
            "state": task_result.state,
            "status": str(task_result.info),  # 失败的异常信息
        }
    return response

# 在 tasks.py 中
from celery.schedules import crontab

# ... (app = Celery(...) 的定义) ...

@app.task
def periodic_task():
    print("这是一个每 30 秒执行一次的定时任务！")

# 配置定时任务
app.conf.beat_schedule = {
    # 任务名称
    'run-every-30-seconds': {
        'task': 'tasks.periodic_task', # 任务的完整路径名
        'schedule': 30.0,             # 每 30 秒
        'args': ()                    # 任务参数（这里没有）
    },
    # 也可以使用 crontab
    'run-every-midnight': {
        'task': 'tasks.add',
        'schedule': crontab(minute=0, hour=0), # 每天午夜
        'args': (100, 200),
    }
}

# (可选) 设置时区，确保 crontab 按你期望的时区执行
app.conf.timezone = 'Asia/Shanghai'

# -B 选项表示同时启动一个嵌入的 Beat 进程
celery -A tasks worker -B --loglevel=info

import requests

# bind=True: 将任务实例 (self) 作为第一个参数注入
# max_retries: 最大重试次数
# default_retry_delay: 默认重试间隔（秒）
@app.task(bind=True, max_retries=3, default_retry_delay=60)
def fetch_api_data(self, url):
    try:
        response = requests.get(url, timeout=5)
        response.raise_for_status() # 如果是 4xx/5xx 错误，抛出异常
        return response.json()
    except requests.exceptions.RequestException as exc:
        print(f"请求失败: {exc}。正在重试...")
        # self.retry() 会抛出一个 Retry 异常，Celery 会捕获它并稍后重试
        # exc=exc 会将原始异常信息传递下去
        raise self.retry(exc=exc)

# myproject/celery.py
from celery import Celery
app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')

# myproject/myapp/tasks.py
from myproject.celery import app

@app.task
def add(x, y):
    return x + y
    

from celery import shared_task

@shared_task
def add(x, y):
    print(f'Adding {x} + {y}')
    return x + y

fastapi_celery_project/
├── main.py             # 你的 FastAPI 应用
├── celery_app.py       # Celery 实例和配置
├── tasks/
│   ├── __init__.py
│   └── email_tasks.py  # 使用 @shared_task 定义任务
└── requirements.txt

fastapi
uvicorn[standard]
celery
redis
* celery[redis] 也可以，它会捆绑 redis 库。

from celery import Celery

# 1. 定义 Broker 和 Backend 的 URL
# 你应该使用环境变量来存储这些敏感信息，这里为了演示而硬编码
BROKER_URL = "redis://localhost:6379/0"
RESULT_BACKEND = "redis://localhost:6379/1"

# 2. 创建 Celery 实例
#    'tasks' 是这个 Celery 实例的默认名称
celery = Celery(
    'tasks',
    broker=BROKER_URL,
    backend=RESULT_BACKEND,
    
    # 3. 关键！指定 Celery 应该自动发现任务的模块
    #    这里我们指向 'tasks.email_tasks' 模块。
    #    这替代了 Django 的 `autodiscover_tasks()`。
    include=['tasks.email_tasks']
)

# 4. (可选) 更多 Celery 配置
celery.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='America/Los_Angeles', # 建议使用 UTC
    enable_utc=True,
    broker_connection_retry_on_startup=True,
)

import time
from celery import shared_task

# @shared_task 装饰器让这个函数成为一个 Celery 任务
# 它不依赖于任何特定的 Celery app 实例
@shared_task
def send_welcome_email(email: str):
    """
    一个模拟发送欢迎邮件的耗时任务
    """
    print(f"Task started: Sending welcome email to {email}...")
    
    # 模拟一个 5 秒钟的 I/O 操作 (例如 API 调用或 SMTP 通信)
    time.sleep(5) 
    
    print(f"Task finished: Email sent to {email}")
    return f"Successfully sent welcome email to {email}"

from fastapi import FastAPI
from tasks.email_tasks import send_welcome_email
from celery.result import AsyncResult
from celery_app import celery as celery_app_instance # 导入用于查询结果的 app 实例

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Welcome to FastAPI + Celery"}

@app.post("/register")
def register_user(email: str):
    """
    一个注册端点，它会异步触发一个欢迎邮件任务
    """
    # .delay() 是 .apply_async() 的快捷方式
    # 它将任务发送到 Celery 队列，并立即返回
    task = send_welcome_email.delay(email)
    
    # 立即返回响应，告诉客户端任务已加入队列
    return {
        "message": "User registration successful. Welcome email is being sent in the background.",
        "task_id": task.id  # 返回任务 ID，以便客户端稍后查询状态
    }

@app.get("/tasks/{task_id}")
def get_task_status(task_id: str):
    """
    根据任务 ID 查询任务的状态和结果
    """
    # 使用从 celery_app.py 导入的实例来创建 AsyncResult
    task_result = AsyncResult(task_id, app=celery_app_instance)
    
    if task_result.ready():
        # 任务已完成
        if task_result.successful():
            return {
                "task_id": task_id,
                "status": task_result.status,
                "result": task_result.get() # 获取任务的返回值
            }
        else:
            # 任务失败
            return {
                "task_id": task_id,
                "status": task_result.status,
                "result": str(task_result.info) # 获取异常信息
            }
    else:
        # 任务仍在进行中
        return {
            "task_id": task_id,
            "status": task_result.status,
            "result": None
        }

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

# -A celery_app.celery 指向 '文件名.Celery实例变量名'
# worker 是命令
# --loglevel=info 是日志级别
celery -A celery_app.celery worker --loglevel=info

uvicorn main:app --reload

from celery import shared_task
import requests

@shared_task(bind=True, max_retries=3)
def fetch_external_api(self, url):
    """
    一个带自动重试的任务
    self 是任务实例，不是类的 self
    """
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status() # 4xx 或 5xx 会抛出异常
        return response.json()
    except requests.RequestException as exc:
        print(f"API call failed for {url}. Retrying... (Attempt {self.request.retries + 1})")
        # self.retry() 会抛出一个异常，Celery 会捕获它并稍后重试
        # countdown=15 表示 15 秒后重试
        raise self.retry(exc=exc, countdown=15)

@shared_task(
    autoretry_for=(requests.RequestException, KeyError), 
    retry_kwargs={'max_retries': 5}, 
    retry_backoff=True  # 启用指数退避 (如 2s, 4s, 8s...)
)
def fetch_api_simple(url):
    # 不需要 try/except，也不需要 self
    # 如果 requests.get 抛出 RequestException，Celery 会自动重试
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    # 如果 response.json() 中缺少某个键，抛出 KeyError，也会重试
    return response.json()['some_key'] 

@shared_task(ignore_result=True)
def log_user_action(user_id, action):
    # 这个任务只是写入日志，不需要返回任何值
    print(f"[LOG] User {user_id} performed action: {action}")

@shared_task(rate_limit="5/m")
def call_rate_limited_api():
    # 这个任务在所有 worker 上加起来，每分钟最多执行 5 次
    ...

from celery import group

jobs = group(add.s(1, 1), add.s(2, 2), add.s(3, 3))
result = jobs.apply_async()
print(result.get())   [2, 4, 6]

from celery import chain

result = chain(add.s(2, 2), mul.s(4), div.s(2)).apply_async()
 等价于: div(mul(add(2,2), 4), 2) → div(mul(4,4),2) → 16/2 = 8

from celery import chord

callback = finalize.s()
header = [add.s(i, i) for i in range(4)]   [0+0, 1+1, 2+2, 3+3]
result = chord(header)(callback)   callback 接收 [0,2,4,6] 作为参数

# 语法: celery -A <app_name> control pool_grow <N>
# 假设你的应用在 tasks.py 中
# 增加 3 个工作进程
celery -A tasks control pool_grow 3

# 语法: celery -A <app_name> control pool_shrink <N>
# 减少 2 个工作进程
celery -A tasks control pool_shrink 2

# 1. 首先，查看所有正在运行的 Worker
celery -A tasks status

# 2. 假设你看到一个名为 worker1@my-server 的 Worker
# 只给它增加并发
celery -A tasks control pool_grow 2 -d worker1@my-server

# 语法: celery -A <app_name> worker --autoscale=<max_concurrency>,<min_concurrency>
#
# 启动一个 Worker，它最少保留 2 个工作进程，
# 但在负载高时，最多可以扩展到 10 个进程
celery -A tasks worker --autoscale=10,2 --loglevel=info

# 假设 Worker 以 --autoscale=10,2 启动
# 现在是高峰期，我们希望它最多能扩展到 20 个进程，最少保持 5 个
celery -A tasks control autoscale 20,5

# 启动一个固定 8 个工作进程的 Worker
celery -A tasks worker -c 8