{"id":1414,"date":"2024-10-13T23:33:09","date_gmt":"2024-10-13T15:33:09","guid":{"rendered":"https:\/\/www.fanyamin.com\/wordpress\/?p=1414"},"modified":"2024-10-13T23:51:59","modified_gmt":"2024-10-13T15:51:59","slug":"%e9%80%9a%e8%bf%87-fastapi-%e5%92%8c-celery-%e6%9d%a5%e5%ae%89%e6%8e%92%e5%ae%9a%e6%97%b6%e4%bb%bb%e5%8a%a1","status":"publish","type":"post","link":"https:\/\/www.fanyamin.com\/wordpress\/?p=1414","title":{"rendered":"\u901a\u8fc7 fastapi \u548c celery \u6765\u5b89\u6392\u5b9a\u65f6\u4efb\u52a1"},"content":{"rendered":"

\u793a\u4f8b\u4e00<\/h2>\n

\u4f60\u60f3\u8981\u6784\u5efa\u4e00\u4e2a\u57fa\u4e8e FastAPI \u7684 Web \u5e94\u7528\uff0c\u5176\u4e2d\u96c6\u6210 Celery Worker \u548c Celery Beat\uff0c\u5e76\u4e14\u6839\u636e Web \u8bf7\u6c42\u6765\u52a8\u6001\u5b89\u6392\u5b9a\u65f6\u4efb\u52a1\u3002\u4ee5\u4e0b\u662f\u8be6\u7ec6\u7684\u6b65\u9aa4\u548c\u4ee3\u7801\u793a\u4f8b\uff0c\u5e2e\u52a9\u4f60\u5b8c\u6210\u8fd9\u4e00\u4efb\u52a1\u3002<\/p>\n

1. \u5b89\u88c5\u4f9d\u8d56<\/h3>\n

\u9996\u5148\uff0c\u4f60\u9700\u8981\u5b89\u88c5 FastAPI\u3001Celery\u3001Redis\uff08\u7528\u4e8e\u4efb\u52a1\u961f\u5217\uff09\u7b49\u4f9d\u8d56\uff1a<\/p>\n

pip install fastapi uvicorn celery redis<\/code><\/pre>\n
Redis \u5c06\u4f5c\u4e3a Celery \u7684\u6d88\u606f\u961f\u5217\u8fdb\u884c\u901a\u4fe1\uff0c\u56e0\u6b64\u4f60\u4e5f\u9700\u8981\u542f\u52a8 Redis \u670d\u52a1\u5668\u3002\u5982\u679c\u6ca1\u6709\u5b89\u88c5 Redis\uff0c\u53ef\u4ee5\u901a\u8fc7\u4ee5\u4e0b\u547d\u4ee4\u8fdb\u884c\u5b89\u88c5\uff1a<\/p>\n
# Ubuntu\/Debian\nsudo apt-get install redis-server\n# MacOS (brew)\nbrew install redis<\/code><\/pre>\n
\u542f\u52a8 Redis \u670d\u52a1\u5668\uff1a<\/p>\n
redis-server<\/code><\/pre>\n
2. \u9879\u76ee\u7ed3\u6784<\/h3>\n
fastapi-celery-app\/\n\u2502\n\u251c\u2500\u2500 app\/\n\u2502   \u251c\u2500\u2500 main.py        # FastAPI \u5e94\u7528\u7684\u5165\u53e3\n\u2502   \u251c\u2500\u2500 tasks.py       # Celery \u4efb\u52a1\u548c\u914d\u7f6e\n\u2502   \u2514\u2500\u2500 celery_worker.py # Celery worker \u548c beat \u7684\u542f\u52a8\u811a\u672c\n\u251c\u2500\u2500 celeryconfig.py    # Celery \u914d\u7f6e\u6587\u4ef6\n\u2514\u2500\u2500 requirements.txt<\/code><\/pre>\n
3. \u914d\u7f6e Celery\uff08celeryconfig.py<\/code>\uff09<\/h3>\n
# celeryconfig.py\n\nfrom datetime import timedelta\n\nbroker_url = "redis:\/\/localhost:6379\/0"\nresult_backend = "redis:\/\/localhost:6379\/0"\n\nbeat_schedule = {\n    "example-scheduled-task": {\n        "task": "app.tasks.scheduled_task",\n        "schedule": timedelta(minutes=1),  # \u6bcf\u5206\u949f\u6267\u884c\u4e00\u6b21\n    },\n}\n\ntimezone = "UTC"<\/code><\/pre>\n
4. \u5b9a\u4e49 Celery \u4efb\u52a1\uff08tasks.py<\/code>\uff09<\/h3>\n
\u5728 tasks.py<\/code> \u4e2d\u5b9a\u4e49 Celery \u7684\u4efb\u52a1\u548c\u52a8\u6001\u4efb\u52a1\u8c03\u5ea6\u7684\u529f\u80fd\uff1a<\/p>\n
# app\/tasks.py\n\nfrom celery import Celery\nfrom celery.schedules import crontab\nimport celeryconfig\n\napp = Celery("tasks")\napp.config_from_object(celeryconfig)\n\n@app.task\ndef example_task(data):\n    print(f"Received data: {data}")\n    return f"Task completed with data: {data}"\n\n@app.task\ndef scheduled_task():\n    print("Scheduled task is running!")\n\n# \u52a8\u6001\u6dfb\u52a0\u4efb\u52a1\ndef schedule_task_at(interval_in_minutes, data):\n    app.conf.beat_schedule[f"dynamic-task-{data}"] = {\n        "task": "app.tasks.example_task",\n        "schedule": crontab(minute=f"*\/{interval_in_minutes}"),\n        "args": [data],\n    }\n    app.conf.beat_schedule_filename = 'celerybeat-schedule'\n    app.conf.update()<\/code><\/pre>\n
\u8fd9\u91cc\uff0c\u6211\u4eec\u5b9a\u4e49\u4e86\u4e00\u4e2a\u7b80\u5355\u7684\u4efb\u52a1 example_task<\/code> \u548c\u4e00\u4e2a\u5b9a\u65f6\u4efb\u52a1 scheduled_task<\/code>\u3002\u540c\u65f6\uff0cschedule_task_at<\/code> \u51fd\u6570\u5141\u8bb8\u901a\u8fc7 Web \u8bf7\u6c42\u52a8\u6001\u6dfb\u52a0\u4efb\u52a1\u3002<\/p>\n
5. FastAPI \u5e94\u7528\uff08main.py<\/code>\uff09<\/h3>\n
\u5728 main.py<\/code> \u4e2d\uff0c\u5904\u7406\u7528\u6237\u8bf7\u6c42\u5e76\u8c03\u5ea6 Celery \u4efb\u52a1\uff1a<\/p>\n
# app\/main.py\n\nfrom fastapi import FastAPI\nfrom app.tasks import example_task, schedule_task_at\nfrom pydantic import BaseModel\n\napp = FastAPI()\n\nclass TaskRequest(BaseModel):\n    interval_in_minutes: int\n    data: str\n\n@app.post("\/schedule-task\/")\nasync def schedule_task(request: TaskRequest):\n    # \u8c03\u7528\u51fd\u6570\uff0c\u5b89\u6392\u52a8\u6001\u4efb\u52a1\n    schedule_task_at(request.interval_in_minutes, request.data)\n    return {"message": "Task scheduled successfully!"}\n\n@app.post("\/run-task\/")\nasync def run_task(request: TaskRequest):\n    task = example_task.delay(request.data)\n    return {"task_id": task.id}<\/code><\/pre>\n
6. \u542f\u52a8 Celery Worker \u548c Beat\uff08celery_worker.py<\/code>\uff09<\/h3>\n
\u521b\u5efa\u4e00\u4e2a\u811a\u672c\u6765\u540c\u65f6\u542f\u52a8 Celery Worker \u548c Celery Beat\uff1a<\/p>\n
# app\/celery_worker.py\n\nfrom app.tasks import app\n\n# \u542f\u52a8 Celery worker \u548c beat\nif __name__ == "__main__":\n    app.worker_main(["worker", "-B", "-l", "info"])<\/code><\/pre>\n
\u8fd9\u91cc\u6211\u4eec\u4f7f\u7528 -B<\/code> \u9009\u9879\u540c\u65f6\u542f\u52a8 Celery Worker \u548c Celery Beat\u3002<\/p>\n
7. \u542f\u52a8\u5e94\u7528<\/h3>\n
\u542f\u52a8 Celery Worker \u548c Beat<\/h4>\n
\u4f7f\u7528 celery_worker.py<\/code> \u6765\u542f\u52a8 Celery Worker \u548c Beat\uff1a<\/p>\n
python app\/celery_worker.py<\/code><\/pre>\n
\u542f\u52a8 FastAPI \u5e94\u7528<\/h4>\n
\u542f\u52a8 FastAPI \u5e94\u7528\uff1a<\/p>\n
uvicorn app.main:app --reload<\/code><\/pre>\n
8. \u6d4b\u8bd5 Web \u5e94\u7528<\/h3>\n
1. \u52a8\u6001\u5b89\u6392\u5b9a\u65f6\u4efb\u52a1<\/h4>\n
\u4f60\u53ef\u4ee5\u901a\u8fc7\u53d1\u9001 POST \u8bf7\u6c42\u6765\u5b89\u6392\u52a8\u6001\u7684\u5b9a\u65f6\u4efb\u52a1\u3002\u4f7f\u7528 curl<\/code> \u6216\u8005 Postman \u6765\u53d1\u9001\u8bf7\u6c42\u3002<\/p>\n
\u8bf7\u6c42\u793a\u4f8b\uff1a<\/p>\n
curl -X 'POST' \\\n  'http:\/\/127.0.0.1:8000\/schedule-task\/' \\\n  -H 'Content-Type: application\/json' \\\n  -d '{\n  "interval_in_minutes": 2,\n  "data": "Task data for every 2 minutes"\n}'<\/code><\/pre>\n
\u8fd9\u4e2a\u8bf7\u6c42\u4f1a\u5b89\u6392\u4e00\u4e2a\u4efb\u52a1\uff0c\u6bcf\u4e24\u5206\u949f\u6267\u884c\u4e00\u6b21\u3002<\/p>\n
2. \u624b\u52a8\u8fd0\u884c\u4efb\u52a1<\/h4>\n
\u4f60\u4e5f\u53ef\u4ee5\u624b\u52a8\u89e6\u53d1\u4e00\u4e2a\u4efb\u52a1\uff1a<\/p>\n
curl -X 'POST' \\\n  'http:\/\/127.0.0.1:8000\/run-task\/' \\\n  -H 'Content-Type: application\/json' \\\n  -d '{\n  "data": "Immediate task data"\n}'<\/code><\/pre>\n
\u8fd9\u4e2a\u8bf7\u6c42\u4f1a\u7acb\u5373\u8fd0\u884c\u4e00\u4e2a\u4efb\u52a1\u3002<\/p>\n
\u793a\u4f8b\u4e8c<\/h2>\n
\u6b65\u9aa4 1: \u8bbe\u7f6e\u9879\u76ee\u7ed3\u6784<\/h3>\n
\u9996\u5148\uff0c\u521b\u5efa\u4e00\u4e2a\u9879\u76ee\u6587\u4ef6\u5939\uff0c\u5e76\u5728\u5176\u4e2d\u8bbe\u7f6e\u4ee5\u4e0b\u6587\u4ef6\u7ed3\u6784\uff1a<\/p>\n
my_fastapi_app\/\n\u251c\u2500\u2500 app\/\n\u2502   \u251c\u2500\u2500 main.py\n\u2502   \u251c\u2500\u2500 tasks.py\n\u2502   \u2514\u2500\u2500 __init__.py\n\u251c\u2500\u2500 alembic.ini\n\u251c\u2500\u2500 Dockerfile\n\u251c\u2500\u2500 requirements.txt\n\u2514\u2500\u2500 .env<\/code><\/pre>\n
\u6b65\u9aa4 2: \u5b89\u88c5\u4f9d\u8d56<\/h3>\n
\u5728 requirements.txt<\/code> \u6587\u4ef6\u4e2d\uff0c\u5217\u51fa\u4f60\u7684\u9879\u76ee\u4f9d\u8d56\uff1a<\/p>\n
fastapi\nuvicorn\ncelery[redis]\nredis<\/code><\/pre>\n
\u7136\u540e\uff0c\u4f7f\u7528 pip \u5b89\u88c5\u8fd9\u4e9b\u4f9d\u8d56\uff1a<\/p>\n
pip install -r requirements.txt<\/code><\/pre>\n
\u6b65\u9aa4 3: \u914d\u7f6e Celery<\/h3>\n
\u5728 app\/__init__.py<\/code> \u6587\u4ef6\u4e2d\uff0c\u914d\u7f6e Celery\uff1a<\/p>\n
from celery import Celery\n\ndef make_celery(app_name=__name__, broker="redis:\/\/localhost:6379\/0", backend=None):\n    return Celery(app_name, broker=broker, backend=backend)\n\ncelery = make_celery()<\/code><\/pre>\n
\u6b65\u9aa4 4: \u5b9a\u4e49 Celery \u4efb\u52a1<\/h3>\n
\u5728 app\/tasks.py<\/code> \u6587\u4ef6\u4e2d\uff0c\u5b9a\u4e49\u4f60\u7684 Celery \u4efb\u52a1\uff1a<\/p>\n
from . import celery\nfrom time import sleep\n\n@celery.on_after_configure.connect\ndef setup_periodic_tasks(sender, **kwargs):\n    # Calls test('hello', 'world') every 10 seconds.\n    sender.add_periodic_task(10.0, test.s('hello', 'world'), name='my_every_10')\n\n@celery.task\ndef test(arg1, arg2):\n    print(f"Task received: {arg1}, {arg2}")\n    sleep(5)  # \u6a21\u62df\u8017\u65f6\u64cd\u4f5c\n    return f"{arg1} + {arg2} = {'hello world'}"<\/code><\/pre>\n
\u6b65\u9aa4 5: \u521b\u5efa FastAPI \u5e94\u7528<\/h3>\n
\u5728 app\/main.py<\/code> \u6587\u4ef6\u4e2d\uff0c\u521b\u5efa\u4f60\u7684 FastAPI \u5e94\u7528\uff0c\u5e76\u96c6\u6210 Celery\uff1a<\/p>\n
from fastapi import FastAPI, HTTPException\nfrom . import celery\nfrom .tasks import test\n\napp = FastAPI()\n\n@app.get("\/")\nasync def read_root():\n    return {"Hello": "World"}\n\n@app.post("\/run-task\/")\nasync def run_task(arg1: str, arg2: str):\n    task = test.delay(arg1, arg2)\n    if task:\n        return {"task_id": task.id, "status": "Pending"}\n    else:\n        raise HTTPException(status_code=500, detail="Failed to start the task")<\/code><\/pre>\n
\u6b65\u9aa4 6: \u542f\u52a8 Celery Worker \u548c Beat<\/h3>\n
\u5728 Dockerfile<\/code> \u4e2d\uff0c\u914d\u7f6e\u4f60\u7684\u5e94\u7528\u548c Celery\uff1a<\/p>\n
FROM python:3.9-slim\n\nWORKDIR \/app\n\nCOPY requirements.txt .\nRUN pip install --no-cache-dir -r requirements.txt\n\nCOPY . .\n\nCMD ["celery", "-A", "main", "worker", "--loglevel=info", "--beat"]<\/code><\/pre>\n
\u6b65\u9aa4 7: \u8fd0\u884c\u5e94\u7528<\/h3>\n
\u5728 .env<\/code> \u6587\u4ef6\u4e2d\uff0c\u8bbe\u7f6e\u73af\u5883\u53d8\u91cf\uff1a<\/p>\n
CELERY_BROKER_URL=redis:\/\/localhost:6379\/0\nCELERY_RESULT_BACKEND=redis:\/\/localhost:6379\/0<\/code><\/pre>\n
\u4f7f\u7528 Docker \u6784\u5efa\u5e76\u8fd0\u884c\u4f60\u7684\u5e94\u7528\uff1a<\/p>\n
docker build -t my_fastapi_app .\ndocker run -d -p 8000:8000 --name my_fastapi_app_container my_fastapi_app<\/code><\/pre>\n
\u73b0\u5728\uff0c\u4f60\u7684 FastAPI \u5e94\u7528\u548c Celery \u5e94\u8be5\u5728 Docker \u5bb9\u5668\u4e2d\u8fd0\u884c\u3002\u4f60\u53ef\u4ee5\u901a\u8fc7\u8bbf\u95ee http:\/\/localhost:8000<\/code> \u6765\u6d4b\u8bd5\u4f60\u7684 FastAPI \u5e94\u7528\u3002<\/p>\n
\u6ce8\u610f\u4e8b\u9879<\/h3>\n