如何在Python中并行运行FastAPI与带健康检查的长时队列轮询任务

下面分两种场景给出无冗余依赖的实现方案：

方案1：单进程异步后台任务（最简洁，无额外依赖） #

直接利用FastAPI生命周期事件+asyncio原生能力实现，服务启动时自动运行轮询任务，不需要等待接口触发，天然共享内存无需额外通信机制，适合轻量负载场景。

实现代码 #

import asyncio
import time
from fastapi import FastAPI
from contextlib import asynccontextmanager

# 全局共享队列，用于接口和轮询任务通信
task_queue = []
run_poll = True

async def queue_poller(interval: int):
    """长时轮询任务"""
    while run_poll:
        # 批量处理队列中所有待处理消息
        while task_queue:
            current_task = task_queue.pop(0)
            # 替换为实际业务处理逻辑
            print(f"处理任务：{current_task}, 时间：{time.time()}")
        # 间隔n秒后下次轮询
        await asyncio.sleep(interval)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 服务启动时自动启动轮询任务
    poll_task = asyncio.create_task(queue_poller(interval=5))
    yield
    # 服务关闭时优雅停止轮询
    global run_poll
    run_poll = False
    await poll_task

app = FastAPI(lifespan=lifespan)

# 健康检查接口
@app.get("/health")
def health_check():
    return {"status": "healthy"}

# 任务下发接口
@app.post("/task")
def push_task(content: str):
    task_queue.append(content)
    return {"msg": "任务已提交"}

优缺点 #

• 优点：零第三方依赖，部署简单，直接通过uvicorn main:app启动即可，容器化只需要单个进程
• 缺点：若轮询逻辑为CPU密集型，会阻塞API请求，仅适合IO密集型、低负载的场景

方案2：多进程隔离部署（适合高负载/CPU密集场景） #

如果轮询逻辑较重，可将FastAPI服务和轮询任务拆分为两个独立进程，通过轻量中间件通信，完全不用引入Celery这类重型框架。这里推荐用Redis做中间件，同时可以直接复用为你的消息队列存储。

实现逻辑 #

1. 进程1：运行FastAPI服务，接收外部请求，将下发的任务写入Redis队列，同时读取轮询进程的心跳实现健康检查
2. 进程2：运行独立的轮询脚本，每隔n秒拉取Redis队列中的所有任务批量处理，定期上报心跳

代码示例 #

FastAPI服务代码（main.py）

import time
import redis
from fastapi import FastAPI

redis_client = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)
app = FastAPI()

@app.get("/health")
def health_check():
    # 校验轮询进程心跳，超过30秒未上报则判定为不健康
    heartbeat = redis_client.get("poller_heartbeat")
    if not heartbeat or (time.time() - float(heartbeat)) > 30:
        return {"status": "unhealthy", "reason": "轮询进程异常"}, 503
    return {"status": "healthy"}

@app.post("/task")
def push_task(content: str):
    redis_client.lpush("task_queue", content)
    return {"msg": "任务已提交"}

轮询进程代码（poller.py）

import time
import redis

redis_client = redis.Redis(host="127.0.0.1", port=6379, db=0)
POLL_INTERVAL = 5

if __name__ == "__main__":
    while True:
        # 上报心跳
        redis_client.set("poller_heartbeat", time.time())
        # 拉取所有待处理任务
        while True:
            task = redis_client.rpop("task_queue")
            if not task:
                break
            # 替换为实际业务处理逻辑
            print(f"处理任务：{task.decode()}")
        time.sleep(POLL_INTERVAL)

容器化部署 #

不需要拆分多个容器，直接在单个容器内用supervisord管理两个进程即可，supervisord配置示例：

[supervisord]
nodaemon=true
logfile=/dev/null
logfile_maxbytes=0

[program:fastapi]
command=uvicorn main:app --host 0.0.0.0 --port 8000
autostart=true
autorestart=true
stdout_logfile=/dev/stdout
stderr_logfile=/dev/stderr

[program:poller]
command=python poller.py
autostart=true
autorestart=true
stdout_logfile=/dev/stdout
stderr_logfile=/dev/stderr

优缺点 #

• 优点：两个进程完全隔离互不影响，Redis作为中间件稳定可靠，后续支持扩展分布式部署，整体复杂度远低于Celery
• 缺点：需要额外部署Redis服务，若项目本身已有Redis实例则无额外成本

如果不想引入Redis这类外部组件，也可以用Python标准库multiprocessing的Queue实现同主机跨进程通信，写一个启动脚本同时拉起两个进程即可，不过这种方式不支持分布式扩展，仅适合单实例运行场景。

内容的提问来源于stack exchange，提问作者srb