我有一个任务，要从数据库获取大量文件（I/O密集型）并处理每一个（CPU密集型）。我考虑使用生产者-消费者模式，其中获取文件的工作者（生产者）异步地获取文件并把它们放进一个队列，处理工作者（消费者）从队列中取出。我只是还不太确定中间的队列会如何工作。这和 这篇博客文章 的做法有点类似（内容是从 这里 存档的），但不是把IO密集型任务串起来再去启动多进程，而是希望处理工作者在获取到文件后就开始处理。大致就是这样：

import asyncio
import multiprocessing

async def fetch_worker(fetch_queue):
    while True:
        try:
            filename = fetch_queue.get_nowait()
            file = await fetch_file(filename)
            await queue.put(file)
            fetch_queue.task_done()
        except asyncio.QueueEmpty:
            break


async def processing_worker():

    while True:

        file = await queue.get()

        if file is None:
            queue.task_done()
            break

        await process_file(file) #??
        queue.task_done()


async def main():

    queue = multiprocessing.Queue() #???
    fetch_queue = asyncio.Queue()
    for filename in filename_list:
        fetch_queue.put_nowait(filename)

    fetch_tasks = [asyncio.create_task(fetch_worker(fetch_queue)) for _ in range(num_fetch_workers)]
    # Here it would be multiprocessing for the processing workers?

我见过以下几个解答：https://stackoverflow.com/a/56944547/32508160、Python - Combining multiprocessing and asyncio，以及 Can I somehow share an asynchronous queue with a subprocess?，但我仍然不确定如何在事件循环中以对这个生产者-消费者模式有意义的方式来部署进程池。

解决方案

import asyncio
from concurrent.futures import ProcessPoolExecutor

# --- Must be a top-level function for multiprocessing pickling ---
def process_file(file: bytes) -> str:
    # Your CPU-bound work here
    return f"processed {len(file)} bytes"


async def fetch_file(filename: str) -> bytes:
    # Your async I/O here (aiohttp, aiobotocore, etc.)
    await asyncio.sleep(0.1)  # simulate network I/O
    return b"file content of " + filename.encode()


async def fetch_and_process(
    filename: str,
    executor: ProcessPoolExecutor,
    semaphore: asyncio.Semaphore,
):
    # Fetch is purely async — no semaphore needed here
    file = await fetch_file(filename)

    # Semaphore limits concurrent CPU-bound processes
    async with semaphore:
        loop = asyncio.get_running_loop()
        result = await loop.run_in_executor(executor, process_file, file)

    return result


async def main():
    filename_list = [f"file_{i}.dat" for i in range(20)]

    num_fetch_workers = 10   # high — I/O is cheap to parallelize
    num_cpu_workers = 4      # match your CPU core count

    semaphore = asyncio.Semaphore(num_cpu_workers)

    with ProcessPoolExecutor(max_workers=num_cpu_workers) as executor:
        tasks = [
            fetch_and_process(filename, executor, semaphore)
            for filename in filename_list
        ]
        results = await asyncio.gather(*tasks)

    print(results)


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

就这样啦 ;)