实现一个支持优先级调度的流式请求队列

在如今的业务平台中，AI 接口调用几乎成了标配。但随之而来的并发问题，往往让开发者头疼不已。想象一下这个场景：用户在前端同时发起了多个请求，一个是耗时较长的“代码生成”，另一个是简单的“句子翻译”。如果按照传统的先进先出（FIFO）队列处理，用户可能得盯着加载动画干等翻译完，才能看到代码结果。

这时候，我们应该如何实现一个支持优先级调度的流式请求队列，允许用户中断低优先级生成，以优先处理高优先级任务？

今天，我们就用通俗易懂的方式，把这个硬核的架构设计彻底讲透。

餐厅后厨点单系统：秒懂优先级调度

在敲代码之前，我们先用一个生活中的例子来建立直觉。你可以把这个调度队列想象成“餐厅后厨的点单系统”。

• 普通点单（低优先级）：客人点了一份“凉拌黄瓜”（相当于翻译句子），后厨开始切菜、拌料。
• VIP点单（高优先级）：老板突然插进来一个紧急订单“佛跳墙”（相当于生成代码）。
• 中断机制：厨师长一看是VIP订单，立刻喊停：“做黄瓜的停一下，把锅让出来！”于是，做黄瓜的厨师立刻停手，把半成品（状态）保存好，转身去做“佛跳墙”。
• 恢复机制：等“佛跳墙”做完端走后，厨师长再喊：“刚才那个黄瓜，接着做！”厨师从刚才停下的地方继续切菜、拌料。

在技术层面，核心答案就是：设计一个请求队列，每个请求带有优先级属性。队列管理器根据优先级调度请求，同一时间只执行一个请求。当高优先级请求到达时，可中断正在执行的低优先级请求。

核心实现要点

要实现上述逻辑，我们需要把握以下几个关键点：

代码实现：Python + asyncio

下面是具体的代码实现。为了直观，我们使用列表来模拟优先级队列（实际生产环境中建议使用 heapq 模块）。

import asyncio
from dataclasses import dataclass, field
from typing import Callable, Any, List

# 定义任务数据结构
@dataclass(order=True)
class Task:
priority: int # 优先级，数字越小优先级越高
timestamp: float = field(compare=False) # 用于同优先级的公平排序
coro_func: Callable = field(compare=False) # 异步任务函数
task_obj: asyncio.Task = field(default=None, compare=False) # asyncio任务对象

class PriorityScheduler:
def __init__(self