如何实现带返回结果的并行任务队列？（多线程有状态Worker场景）

我有一台接收数据处理请求的Web服务器，处理工作必须由有状态Worker（例如独占数据库连接）执行，可并行运行任意数量的Worker。需要实现任务分发到共享任务队列并返回结果的方案，要求满足：

• 使用async-await语法
• 无需轮询
• 每个Worker运行在独立线程

我了解过ThreadPoolExecutor，但它的API过于复杂，需要用到ExitStack及__enter__/__exit__等；也曾考虑编写类似的抽象，但需要重复实现“Backend”的API。想知道是否有更优替代方案？

附伪代码：

workers = []
workers.append(start_worker(db1))
workers.append(start_worker(db2))
workers.append(start_worker(db3))
workers.append(start_worker(db4))

async def request(data):
  return await any_worker_do_work(data)


async def any_worker_do_work(data):
  # ?

def worker_do_work(data):
  # 同步处理逻辑
  return self.db.do_work(data)

# 这些请求应分配到不同Worker执行
print("Result: " + str(await request("hello")))
print("Result: " + str(await request("hello")))
print("Result: " + str(await request("hello")))
print("Result: " + str(await request("hello")))
print("Result: " + str(await request("hello")))

这个方案用asyncio.Queue做任务分发，给每个Worker绑定专属状态（如DB连接）并分配独立线程，通过asyncio.Future实现无轮询的异步结果返回，完全符合你的需求，且API简洁。

核心思路 #

1. 每个StatefulWorker实例持有专属的状态资源（如数据库连接），并运行在独立线程中
2. 用asyncio.Queue传递任务数据和Future对象，Worker处理完成后直接通过Future返回结果
3. 实现Worker池进行任务调度（轮询或负载均衡），让请求自动分配到空闲或指定Worker

完整代码实现 #

import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import Any

class StatefulWorker:
    def __init__(self, db_conn: Any):
        self.db_conn = db_conn
        self.task_queue = asyncio.Queue()
        # 每个Worker独占一个线程，确保状态资源的线程安全
        self.thread_executor = ThreadPoolExecutor(max_workers=1)
        self.loop = asyncio.get_event_loop()

    async def start(self):
        # 在独立线程启动同步任务循环
        await self.loop.run_in_executor(self.thread_executor, self._task_process_loop)

    def _task_process_loop(self):
        # 同步循环：持续从队列取任务并处理
        while True:
            # 在线程中通过事件循环获取队列任务
            task_data, result_future = self.loop.run_until_complete(self.task_queue.get())
            try:
                # 执行同步处理逻辑，使用专属DB连接
                result = self._do_work(task_data)
                # 线程安全地设置Future结果
                self.loop.call_soon_threadsafe(result_future.set_result, result)
            except Exception as e:
                # 传递异常到异步上下文
                self.loop.call_soon_threadsafe(result_future.set_exception, e)
            finally:
                self.task_queue.task_done()

    def _do_work(self, data: Any) -> Any:
        # 替换为你的实际同步处理逻辑
        return self.db_conn.do_work(data)

    async def submit_task(self, data: Any) -> Any:
        # 创建Future用于接收结果，将任务放入队列
        result_future = self.loop.create_future()
        await self.task_queue.put((data, result_future))
        return await result_future

class WorkerPool:
    def __init__(self, workers: list[StatefulWorker]):
        self.workers = workers
        self._next_worker_idx = 0

    async def submit_to_any_worker(self, data: Any) -> Any:
        # 轮询调度：依次分配任务到不同Worker，也可改为负载均衡策略
        worker = self.workers[self._next_worker_idx]
        self._next_worker_idx = (self._next_worker_idx + 1) % len(self.workers)
        return await worker.submit_task(data)

# 示例使用
async def main():
    # 假设此处已初始化好db1~db4数据库连接实例
    workers = [
        StatefulWorker(db1),
        StatefulWorker(db2),
        StatefulWorker(db3),
        StatefulWorker(db4)
    ]
    # 启动所有Worker的任务循环
    await asyncio.gather(*[worker.start() for worker in workers])

    # 初始化Worker池
    worker_pool = WorkerPool(workers)

    # 定义请求处理函数
    async def request(data):
        return await worker_pool.submit_to_any_worker(data)

    # 批量测试请求
    results = await asyncio.gather(
        request("hello"),
        request("hello"),
        request("hello"),
        request("hello"),
        request("hello")
    )
    for idx, res in enumerate(results):
        print(f"Result {idx+1}: {res}")

if __name__ == "__main__":
    asyncio.run(main())

方案优势 #

• 状态隔离：每个Worker持有专属资源，运行在独立线程，避免多线程共享状态的安全问题
• 无轮询异步结果：通过asyncio.Future实现结果回调，天然支持async-await，无需主动轮询任务状态
• 简洁API：不需要复杂的上下文管理器或ExitStack，代码结构清晰易维护
• 灵活调度：可轻松替换调度策略（如选择队列最短的Worker实现负载均衡）
• Web服务器兼容：完全适配FastAPI、Starlette等异步Web框架的上下文，可直接集成到请求处理流程中

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