Python FastAPI跨线程变量共享问题：Webhook与任务线程互通

核心原因 #

FastAPI默认使用多进程ASGI服务器（如Uvicorn），模块级普通字典job_events和job_statuses在每个进程中是独立副本，导致Webhook所在进程无法读取到任务启动进程中添加的事件数据。

修正代码 #

1. 修改任务启动模块（otherservice.py） #

使用multiprocessing.Manager创建跨进程共享的数据结构：

from multiprocessing import Manager
import requests
import threading

# 创建跨进程共享的字典与事件管理器
manager = Manager()
job_events = manager.dict()
job_statuses = manager.dict()

class OtherService:
    def launch(self, **data):
        # 调用外部服务启动任务
        response = requests.post("YOUR_EXTERNAL_SERVICE_LAUNCH_URL", json=data)
        job_id = response.json()["id"]
        print(f"Launched job with id {job_id}")
        
        # 创建跨进程共享的事件对象
        event = manager.Event()
        job_events[job_id] = event

        print(f"Waiting for job {job_id} to complete...")  
        # 添加超时逻辑，避免无限阻塞
        if not event.wait(timeout=300):
            raise TimeoutError(f"Job {job_id} timed out after 5 minutes")

        # 获取任务状态并返回
        status = job_statuses[job_id]
        print(f"Job {job_id} completed with status {status}")
        # 清理资源，避免内存泄漏
        del job_events[job_id]
        del job_statuses[job_id]
        return status

2. Webhook端点代码（main.py） #

保持导入方式不变，逻辑无需大幅修改，仅补充必要处理：

from fastapi import FastAPI
from pydantic import BaseModel
from otherservice import job_events, job_statuses

app = FastAPI()

class WebhookData(BaseModel):
    job_id: str
    status: str

@app.post("/webhook/")
async def webhook_handler(webhook_data: WebhookData):
    job_id = webhook_data.job_id
    status = webhook_data.status
    
    job_statuses[job_id] = status

    if job_id in job_events:
        event = job_events[job_id]
        event.set()
        print(f"Event triggered for job {job_id} (status: {status})")
    
    return {"result": "success"}

额外建议 #

• 生产环境优化：高并发场景下建议用Redis替代multiprocessing.Manager，避免管理进程成为性能瓶颈，同时支持分布式部署。
• 异常处理：增加Webhook签名验证，防止非法请求；处理重复通知时，先检查任务状态是否已完成，避免重复触发事件。
• 资源清理：务必在任务完成后删除共享字典中的数据，防止内存泄漏。

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