FastAPI BackgroundTasks执行顺序问题：如何将LIFO改为FIFO？

问题说明 #

使用FastAPI的BackgroundTasks时，按request1 - request2 - request3的顺序调用任务创建接口，期望任务按相同顺序执行，但实际执行顺序为request1 - request3 - request2。后续任务呈现**后进先出（LIFO）**特性：新请求的任务会优先执行，比如新增request4、request5后，执行顺序变为request5 - request4 - request3 - request2。

不同环境下表现不一致：

• Windows 10（Python 3.8.2 + FastAPI 0.98）：任务按**先进先出（FIFO）**正常执行
• CentOS（Python 3.9.12 + FastAPI 0.96）：出现LIFO执行顺序问题

复现代码 #

App.py #

from uuid import UUID
from pydantic import BaseModel
from typing import Dict
from anyio.lowlevel import RunVar
from anyio import CapacityLimiter
from http import HTTPStatus
import time
from fastapi import FastAPI, BackgroundTasks


class Job(BaseModel):
    uid: UUID
    status: str = 'in_progress'
    
# API 初始化
app = FastAPI()
jobs: Dict[UUID, Job] = {}   

def mock_function(id_job) -> str:
    jobs[id_job].status = 'in_progress'
    time.sleep(5)

def process_request(job_id):
    mock_function(job_id)   
    jobs[job_id].status = 'complete'
    
@app.on_event("startup")
def startup():
    # 限制同时执行的后台任务数为1
    RunVar("_default_thread_limiter").set(CapacityLimiter(1))

       
@app.get("/status")
async def status_handler() :
    return jobs

 
@app.post('/request/{uid}', status_code=HTTPStatus.ACCEPTED)
async def request_API(uid:UUID, background_tasks: BackgroundTasks):
    new_task = Job(uid=uid)
    new_task.status = 'in_queue'
    jobs[new_task.uid] = new_task
    background_tasks.add_task(process_request, new_task.uid)

    return new_task  

测试脚本 #

import requests
import time
query_tasks = {}
jobs_id = [
    'c880cc1b-dc27-4175-b616-29a69322d156', 
    '4f0ea1f1-a5a3-4a7c-a114-17e57a1c55db',
    '15ba0d0a-3c94-4906-a331-054a3847171c'
]
for i in range(3):
    r = requests.post(f'http://127.0.0.1:8000/request/{jobs_id[i]}')
    id_ = r.json()['uid']
    query_tasks[id_] = i+1

for t in range(20):
    r = requests.get('http://127.0.0.1:8000/status')
    tasks = r.json()
    for k,v in tasks.items():
        print(query_tasks[k])
        print(v['status'])
    time.sleep(1)
    print('-----------------------------------------------')

测试输出（简化版） #

1
complete
2
in_queue
3
in_progress
-----------------------------------------------
1
complete
2
in_progress
3
complete

解决思路 #

1. 升级FastAPI及依赖版本 #

Windows环境下高版本FastAPI（0.98）表现正常，说明低版本（0.96）可能存在Starlette任务调度的bug。升级FastAPI到0.98+版本，同步更新Starlette，大概率能解决环境不一致的问题。

2. 自定义FIFO任务队列替代BackgroundTasks #

BackgroundTasks的底层调度未明确保证FIFO顺序，可通过asyncio.Queue实现自定义队列，自己控制任务执行顺序：

from uuid import UUID
from pydantic import BaseModel
from typing import Dict
import asyncio
import time
from fastapi import FastAPI


class Job(BaseModel):
    uid: UUID
    status: str = 'in_progress'
    
app = FastAPI()
jobs: Dict[UUID, Job] = {}
task_queue = asyncio.Queue()

def mock_function(id_job):
    jobs[id_job].status = 'in_progress'
    time.sleep(5)

def process_request(job_id):
    mock_function(job_id)   
    jobs[job_id].status = 'complete'

async def queue_worker():
    while True:
        job_id = await task_queue.get()
        try:
            process_request(job_id)
        finally:
            task_queue.task_done()

@app.on_event("startup")
async def startup():
    # 启动队列消费线程
    asyncio.create_task(queue_worker())

@app.get("/status")
async def status_handler() :
    return jobs

@app.post('/request/{uid}', status_code=202)
async def request_API(uid:UUID):
    new_task = Job(uid=uid)
    new_task.status = 'in_queue'
    jobs[new_task.uid] = new_task
    await task_queue.put(new_task.uid)
    return new_task  

此方案通过单消费者队列天然保证FIFO顺序，不受环境或版本影响。

3. 自定义BackgroundTasks子类（进阶） #

查看Starlette的BackgroundTasks源码，其任务存储在列表中，若执行时采用逆序调用（比如用pop()）会导致LIFO。可自定义子类修改执行逻辑：

from starlette.background import BackgroundTasks as StarletteBackgroundTasks

class FIFOBackgroundTasks(StarletteBackgroundTasks):
    async def __call__(self, scope, receive, send):
        # 按添加顺序执行任务，而非逆序
        for task in self.tasks:
            await task()

然后在接口中使用自定义的FIFOBackgroundTasks替代默认的BackgroundTasks。

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