如何结合 asyncio 与 multiprocessing 实现边获取边处理文件？
可以在主进程中用 asyncio.Queue 连接文件获取和处理阶段，再通过 loop.run_in_executor() 将 CPU-bound 函数提交给 ProcessPoolExecutor。文件获取完成后就可以提交给进程池，不必等所有文件下载结束。
这里的 processing worker 是主进程中的协程，负责取文件、提交任务和等待结果；真正的 CPU 计算在进程池的子进程中执行。
为什么这里使用 asyncio.Queue？
生产者和消费者都是运行在同一个 event loop 中的协程，可以直接共享 asyncio.Queue。数据进入子进程时，由进程池负责进程间传输。
原代码中的 multiprocessing.Queue 不能直接这样使用：
await queue.put(file)
file = await queue.get()它的 put() 和 get() 是同步方法，不返回可等待对象，直接调用还可能阻塞 event loop。普通的 multiprocessing.Queue 也没有 task_done() 和 join() 这套任务计数接口。
原代码还有一个作用域问题：在 main() 内创建的局部变量 queue 不会自动对其他模块级函数可见，需要显式传入队列。
如何实现持续获取、持续处理？
下面的示例适用于 Python 3.11 及以上版本，使用 asyncio.TaskGroup 管理协程。使用前，将两个占位函数替换成实际业务逻辑：fetch_file() 负责异步获取，process_file() 负责同步 CPU 计算。
import asyncio
from concurrent.futures import ProcessPoolExecutor
async def fetch_file(filename):
# 替换为实际的异步数据库读取逻辑。
raise NotImplementedError
def process_file(file):
# 替换为实际的同步 CPU-bound 处理逻辑。
# 此函数必须定义在模块顶层，供子进程导入。
raise NotImplementedError
async def fetch_worker(fetch_queue, ready_queue):
while True:
try:
filename = fetch_queue.get_nowait()
except asyncio.QueueEmpty:
return
try:
file = await fetch_file(filename)
await ready_queue.put(file)
finally:
fetch_queue.task_done()
async def processing_worker(ready_queue, pool):
loop = asyncio.get_running_loop()
while True:
file = await ready_queue.get()
try:
if file is None:
return
result = await loop.run_in_executor(
pool, process_file, file
)
# 在这里保存或使用 result。
finally:
ready_queue.task_done()
async def main(
filename_list,
num_fetch_workers=8,
num_process_workers=4,
queue_size=16,
):
if min(num_fetch_workers, num_process_workers, queue_size) < 1:
raise ValueError("worker counts and queue_size must be positive")
fetch_queue = asyncio.Queue()
ready_queue = asyncio.Queue(maxsize=queue_size)
for filename in filename_list:
fetch_queue.put_nowait(filename)
with ProcessPoolExecutor(
max_workers=num_process_workers
) as pool:
async with asyncio.TaskGroup() as tasks:
for _ in range(num_process_workers):
tasks.create_task(
processing_worker(ready_queue, pool)
)
for _ in range(num_fetch_workers):
tasks.create_task(
fetch_worker(fetch_queue, ready_queue)
)
# 等待所有获取任务完成并将文件放入 ready_queue。
await fetch_queue.join()
# 每个消费者协程需要一个结束信号。
for _ in range(num_process_workers):
await ready_queue.put(None)
# 等待文件处理及结束信号消费完成。
await ready_queue.join()
if __name__ == "__main__":
# 替换为实际的文件标识列表。
asyncio.run(main(["example-file"]))示例中的 worker 数量和队列容量仅供参考，需要根据数据库并发限制、文件大小和计算开销调整。
为什么等待 fetch_queue.join() 不会推迟处理？
消费者协程已经提前启动。主协程执行 await fetch_queue.join() 时会让出 event loop，生产者和消费者可以继续运行。
文件进入 ready_queue 后，空闲的消费者就可以取出文件并提交给进程池。这里的 join() 用于确定发送结束信号的时机，获取和处理并非串行执行。
每个消费者要等当前进程池任务完成，才会取下一个文件。因此，同时提交但尚未完成的处理任务不会超过消费者数量。
如何限制积压和内存占用？
ready_queue 的 maxsize 限制了队列容量。队列满时，生产者会在 await ready_queue.put(file) 处等待，直到消费者腾出空间，以此限制已获取文件的积压。
不过，队列容量不是总内存上限。正在获取的文件、生产者等待入队时持有的文件、子进程中的数据以及序列化缓冲区都会占用内存。文件很大时，可以考虑让进程池接收临时文件路径等较小的标识，前提是子进程能够访问相应数据。
示例会一次性将所有文件名放入 fetch_queue。如果文件标识数量也非常多，可以增加一个异步生产阶段，逐批生成文件名。
使用时需要注意什么？
process_file()应是普通def函数。将async def函数直接交给进程池，不会自动执行其协程体。- 传入进程池的函数、参数及返回结果都必须能够被序列化。通常应传递文件内容、路径或普通数据结构，避免直接传入数据库连接或打开的文件对象。
fetch_file()必须实际使用异步 I/O。只在函数前加async，不会让同步数据库调用变成非阻塞操作。- 保留
if __name__ == "__main__":入口保护，并将进程池函数放在可导入模块的顶层。示例适合作为独立 Python 脚本运行。 - 示例用
None作为结束信号，因此实际文件数据不能使用None；如有需要，可以另行设计消息结构。 TaskGroup中任一任务抛出异常时，会取消其余协程并向外传播异常。finally中的task_done()只用于配平队列计数，不代表业务处理成功。- 取消等待进程池结果的协程，通常不能停止已经开始运行的子进程任务。退出进程池上下文时，可能仍需等待这些任务结束。
如果希望子进程长期运行自己的取队列循环，也可以采用进程间队列，但需要额外处理阻塞调用、结束信号和进程生命周期。对于当前按文件提交独立 CPU 任务的需求，asyncio.Queue 配合 ProcessPoolExecutor 就能完成流水线衔接。
备注：内容仅供参考。