FastAPI多Worker模式下静态变量同步与服务正常退出问题

当uvicorn启用workers=5时，每个worker都是独立的Python进程，进程间内存完全隔离，单进程内的静态变量无法跨进程同步，这就是你关闭服务器时无法统一更新MessageBroker.run_message_broker、导致应用无法正常退出的根本原因。以下是几种可行的解决方案：

1. 使用进程间通信（IPC）机制 #

共享内存变量 #

利用Python标准库multiprocessing的Value创建跨进程共享的布尔变量，替代原静态变量：

from multiprocessing import Value
from fastapi import FastAPI
import threading
import time

app = FastAPI()
# 创建跨进程共享的布尔标志，初始值为True
run_flag = Value('b', True)

class MessageBroker:
    @staticmethod
    def consume_messages():
        while run_flag.value:
            # 模拟消息消费逻辑
            print("Consuming message...")
            time.sleep(1)

@app.on_event("startup")
async def startup():
    threading.Thread(target=MessageBroker.consume_messages, daemon=True).start()

@app.on_event("shutdown")
async def shutdown():
    # 跨进程更新标志
    run_flag.value = False

注意：需确保所有worker能访问同一个Value实例，可在启动前统一初始化或用multiprocessing.Manager管理。

消息队列传递停止信号 #

通过队列向所有worker的消费线程发送停止指令：

from multiprocessing import Queue
from fastapi import FastAPI
import threading
import time

app = FastAPI()
stop_queue = Queue()

class MessageBroker:
    @staticmethod
    def consume_messages():
        while True:
            if not stop_queue.empty() and stop_queue.get() == "STOP":
                break
            # 消息消费逻辑
            print("Consuming message...")
            time.sleep(1)

@app.on_event("startup")
async def startup():
    threading.Thread(target=MessageBroker.consume_messages, daemon=True).start()

@app.on_event("shutdown")
async def shutdown():
    # 给每个worker发送停止信号（对应workers=5）
    for _ in range(5):
        stop_queue.put("STOP")

2. 用外部存储维护全局状态 #

借助Redis等外部存储保存全局运行标志，所有worker统一读取该状态：

import redis
from fastapi import FastAPI
import threading
import time

app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, db=0)

@app.on_event("startup")
async def startup():
    redis_client.set("run_message_broker", "True")
    threading.Thread(target=MessageBroker.consume_messages, daemon=True).start()

class MessageBroker:
    @staticmethod
    def consume_messages():
        while redis_client.get("run_message_broker") == b"True":
            # 消息消费逻辑
            print("Consuming message...")
            time.sleep(1)

@app.on_event("shutdown")
async def shutdown():
    redis_client.set("run_message_broker", "False")

该方案适用于分布式场景，即使worker部署在不同机器也能同步状态。

3. 捕获系统信号处理退出 #

uvicorn关闭时会给每个worker发送SIGTERM信号，通过捕获该信号更新本地运行标志：

import signal
from fastapi import FastAPI
import threading
import time

app = FastAPI()
run_flag = True

def handle_shutdown(signum, frame):
    global run_flag
    run_flag = False

# 注册SIGTERM信号处理函数
signal.signal(signal.SIGTERM, handle_shutdown)

class MessageBroker:
    @staticmethod
    def consume_messages():
        while run_flag:
            # 消息消费逻辑
            print("Consuming message...")
            time.sleep(1)

@app.on_event("startup")
async def startup():
    threading.Thread(target=MessageBroker.consume_messages, daemon=True).start()

每个worker独立处理信号，无需跨进程同步，即可停止自身的消费线程。

4. 切换为单worker+多线程模式 #

若业务对多核CPU利用率要求不高，可去掉workers=5，使用单worker配合多线程处理请求和消息消费，此时静态变量可正常跨线程同步，避免进程隔离问题。

内容的提问来源于stack exchange，提问作者Kiran N S