FastAPI后台任务报错：signal仅在主解释器主线程中可用

问题描述 #

使用FastAPI的BackgroundTasks处理耗时任务，任务完成后调用send_data_source_ack函数向远程服务发送状态通知，部署在Docker中时抛出错误：signal only works in main thread of the main interpreter。

相关代码及配置如下：

通知函数代码 #

def send_data_source_ack(community_id, dataSourceId,status='LOADED'):
    config = get_env_config()
    uri = '/api/rpc/chat-bot/data-source/ack'
    request = f"?status={status}&communityId="+str(community_id)+"&dataSourceId="+str(dataSourceId)
    url = config.APIBaselURL.url + uri + request
    response = requests.post(url, data={},verify=False)
    if response.status_code == 200:
        api_requests.debug(f'POST request successful for {community_id} with status {response.text}')
    else:
        api_requests.debug(f'POST request unsuccessful for {community_id} with status code {response.status_code}')

Dockerfile配置 #

FROM python:3.9.17-slim-bullseye

WORKDIR /src

COPY requirements.txt requirements.txt

RUN apt-get update
RUN apt-get -y install libpq-dev gcc 
RUN pip3 install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 9856

CMD ["uvicorn","chatbot.run:app","--host","0.0.0.0","--port","9856","--workers","4","--reload"]

FastAPI路由代码 #

from fastapi import BackgroundTasks, Request, Query, status
import logging

logger = logging.getLogger(__name__)

@app.post('/data-config/trigger',status_code=status.HTTP_200_OK)
async def handle_data_config_trigger(
    request:Request,
    tasks:BackgroundTasks,
    community_id: int =  Query(alias='community_id'),
    data_source_type: str = Query(alias='data_source_type')
    ):
    data_config = await request.json()
    data_config_source = DataConfigSource(data_config=data_config)
   
    # 添加后台任务
    tasks.add_task(run_training_in_background_task,
             community_id,data_source_type,
             data_config_source.data_config)

    return f"{data_source_type} started for community_id {community_id}"


async def run_training_in_background_task(community_id,data_source_type,data_config):
    try:
        logger.debug(f'Hit data config trigger for {community_id}')
        if data_source_type == "PDF":
            trigger_pdf_data_config(community_id, data_config)
        
        # 触发错误的函数调用
        send_data_source_ack(community_id, data_config.id,status='LOADED')

    except Exception as e:
        logger.debug(f'Exception thrown in processing the training for community id - {community_id}: {str(e)}')

错误原因 #

1. 多Worker模式冲突：Uvicorn启用--workers 4多进程模式，每个Worker是独立的Python进程，BackgroundTasks会在Worker进程的子线程中执行。而Python的信号机制仅允许在主解释器的主线程中操作，子进程的子线程无法执行信号相关操作。
2. 同步HTTP客户端的限制：requests是同步HTTP库，其底层可能依赖信号实现超时控制，在非主解释器主线程中调用时触发信号操作，导致报错。
3. Reload模式影响：--reload模式下Uvicorn会通过子进程运行应用，进一步加剧了线程/进程环境的冲突。

修复方案 #

方案一：禁用多Worker模式（快速修复） #

修改Dockerfile中的Uvicorn启动命令，移除--workers 4参数，改用单Worker运行：

# 开发环境保留--reload，生产环境建议移除
CMD ["uvicorn","chatbot.run:app","--host","0.0.0.0","--port","9856","--reload"]

# 生产环境配置（移除--reload）
# CMD ["uvicorn","chatbot.run:app","--host","0.0.0.0","--port","9856"]

优势：无需修改代码，快速解决问题。
局限：无法利用多进程提升并发能力，仅适合低流量场景或开发阶段。

方案二：改用异步HTTP客户端（推荐） #

将同步的requests替换为异步HTTP客户端aiohttp，避免信号操作冲突，同时适配FastAPI的异步生态：

1. 在requirements.txt中添加aiohttp依赖：

aiohttp>=3.9.0

2. 修改send_data_source_ack为异步函数：

import aiohttp
import logging

async def send_data_source_ack(community_id, dataSourceId, status='LOADED'):
    config = get_env_config()
    uri = '/api/rpc/chat-bot/data-source/ack'
    # 使用f-string简化URL拼接
    url = f"{config.APIBaselURL.url}{uri}?status={status}&communityId={community_id}&dataSourceId={dataSourceId}"
    
    async with aiohttp.ClientSession() as session:
        try:
            async with session.post(url, data={}, ssl=False) as response:
                if response.status == 200:
                    logging.debug(f'POST request successful for {community_id} with status {await response.text()}')
                else:
                    logging.debug(f'POST request unsuccessful for {community_id} with status code {response.status}')
        except Exception as e:
            logging.error(f'Failed to send ack for community {community_id}: {str(e)}')

3. 在后台任务中异步调用该函数：

async def run_training_in_background_task(community_id, data_source_type, data_config):
    try:
        logger.debug(f'Hit data config trigger for {community_id}')
        if data_source_type == "PDF":
            trigger_pdf_data_config(community_id, data_config)
        
        # 异步调用需加await
        await send_data_source_ack(community_id, data_config.id, status='LOADED')

    except Exception as e:
        logger.debug(f'Exception thrown in processing the training for community id - {community_id}: {str(e)}')

优势：适配FastAPI异步架构，支持多Worker模式，性能更优。
局限：需要修改现有代码，替换HTTP客户端。

方案三：改用专用任务队列（适合重度耗时任务） #

如果任务耗时较长，建议使用Celery + Redis/RabbitMQ等专用任务队列替代BackgroundTasks，避免FastAPI Worker被阻塞，同时提供更可靠的任务管理能力：

1. 安装Celery及依赖：

celery>=5.3.0
redis>=5.0.0

2. 配置Celery实例，将send_data_source_ack封装为Celery任务，在FastAPI路由中调用任务而非BackgroundTasks。
   优势：支持分布式任务执行，任务持久化，适合大规模生产环境。
   局限：需要额外部署任务队列服务，架构复杂度提升。

内容的提问来源于stack exchange，提问作者James K J