Python多进程与协程搭配使用出现阻塞无输出问题咨询

1. 异常原因说明 #

出现该问题是两个原因共同导致的：

• stdout缓冲问题：子进程的标准输出默认使用全缓冲模式，你代码中的print内容没有被强制刷新到控制台，所以看不到输出。
• 全局协程资源损坏：你导入的aiohttp_requests内部会预先创建绑定了父进程事件循环的全局客户端实例，Windows下multiprocessing默认使用spawn模式启动子进程时，这个全局实例的状态会被破坏，导致子进程初始化阶段直接卡死，连后续的print逻辑都无法执行。你注释掉loop.run_forever时，代码没有触发这个全局实例的初始化逻辑，所以能正常运行输出。

2. 单独多进程无异常的原因 #

你猜测的写时复制和stdout输出没有关联，看不到输出的核心是缓冲+子进程提前卡死。单独使用多进程时没有引入asyncio/aiohttp相关的全局初始化逻辑，不会触发状态损坏的问题，子进程可以正常运行，输出缓冲也能在进程退出时自动刷出，所以能看到打印内容。

3. 代码调整方案 #

按以下规则调整即可正常实现需求：

• 所有print语句添加flush=True参数，强制输出内容实时刷新到控制台
• 弃用全局的aiohttp_requests.requests实例，每个进程内部独立创建aiohttp.ClientSession，避免全局状态冲突
• 去掉子进程内额外的事件循环线程，通过asyncio.Queue实现动态任务提交，比跨线程提交任务更稳定
• 添加进程停止信号监听，避免事件循环永久运行无法退出

修改后的可运行代码如下：

import time
import asyncio
import aiohttp
import multiprocessing
from multiprocessing import Process, Queue

async def test(session):
    try:
        async with session.get('https://www.huya.com/cache.php?m=LiveList&do=getLiveListByPage&tagAll=0&page=1') as r:
            print(r.status, flush=True)
    except Exception as e:
        print(f"请求出错: {e}", flush=True)

async def task_consumer(async_queue: asyncio.Queue, session: aiohttp.ClientSession):
    while True:
        task = await async_queue.get()
        if task is None:
            # 收到停止信号，退出消费
            break
        await test(session)
        async_queue.task_done()

def processMain(task_queue: multiprocessing.Queue):
    print("enter the process", flush=True)
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    async def run():
        # 进程内独立创建session，避免全局状态冲突
        async with aiohttp.ClientSession() as session:
            async_queue = asyncio.Queue(maxsize=1000)
            # 启动协程消费者
            consumer_task = asyncio.create_task(task_consumer(async_queue, session))
            
            # 同步拉取进程间队列的任务，提交到async队列
            def sync_task_fetcher():
                while True:
                    task = task_queue.get()
                    if task is None:
                        loop.call_soon_threadsafe(async_queue.put_nowait, None)
                        break
                    loop.call_soon_threadsafe(async_queue.put_nowait, task)
            
            import threading
            threading.Thread(target=sync_task_fetcher, daemon=True).start()
            
            await consumer_task
    
    loop.run_until_complete(run())
    loop.close()
    print("the subprocess finished", flush=True)

if __name__ == "__main__":
    print('start the program', flush=True)
    # 进程间通信队列，用于动态提交任务
    task_queue = multiprocessing.Queue()
    # 可启动多个进程提升并发能力
    process = Process(target=processMain, args=(task_queue,))
    process.start()
    
    # 测试提交10个任务
    for i in range(10):
        task_queue.put(f"task_{i}")
        time.sleep(0.001)
    
    # 发送停止信号
    task_queue.put(None)
    process.join()
    print("finish", flush=True)

内容的提问来源于stack exchange，提问作者陈之文