使用asyncio.Queue读取大文件URL遇阻塞问题及优化方案咨询

问题1：错误原因及修复 #

你的代码阻塞的核心原因是生产者和消费者没有并发执行：你先在with open块里循环往队列塞URL，当队列达到maxsize=5后，await queue.put()会阻塞等待队列有空位，但此时消费者逻辑（后面的while not queue.empty()循环）还没开始执行，没有任何任务在从队列取数据，导致程序彻底卡死在put操作上。

另外还有两个小问题：

• fetch函数每次请求都创建新的ClientSession，频繁建立/关闭连接会拉低效率，应该复用同一个Session。
• 单线程消费队列的方式没利用好asyncio的并发优势，应该启动多个消费任务。

修复后的代码示例：

import asyncio
from aiohttp import ClientSession

async def fetch(url, session):
    # 复用传入的session，避免重复创建连接
    async with session.get(url) as response:
        return await response.text()

async def worker(queue, session):
    # 消费者协程，持续从队列取URL处理
    while True:
        url = await queue.get()
        try:
            content = await fetch(url, session)
            print(f"{url}: {content[:15]}")
        except Exception as e:
            print(f"Error fetching {url}: {e}")
        finally:
            queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=5)
    # 创建全局ClientSession复用
    async with ClientSession() as session:
        # 启动3个消费者任务（数量可按需调整）
        tasks = [asyncio.create_task(worker(queue, session)) for _ in range(3)]
        
        # 生产者：读取URL并放入队列，此时消费者已在运行，队列满了会自动等待消费者取走数据
        with open("urls.txt", 'r') as f:
            for line in f:
                line = line.strip()
                if line:  # 跳过空行
                    await queue.put(line)
        
        # 等待队列所有任务处理完成
        await queue.join()
        
        # 终止消费者任务（worker是无限循环，需手动取消）
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)

asyncio.run(main())

问题2：更合适的解决方案 #

如果你的核心需求是控制并发抓取数量，而非必须用队列做缓冲，直接用asyncio.Semaphore控制并发数会更简洁，不需要手动管理队列：

import asyncio
from aiohttp import ClientSession

async def fetch(url, session, semaphore):
    async with semaphore:  # 用信号量限制并发执行的任务数
        async with session.get(url) as response:
            return await response.text()

async def main():
    # 限制最大并发数为5，和之前队列maxsize的效果一致
    semaphore = asyncio.Semaphore(5)
    async with ClientSession() as session:
        tasks = []
        with open("urls.txt", 'r') as f:
            for line in f:
                url = line.strip()
                if url:
                    # 直接创建抓取任务，信号量自动控制并发数
                    task = asyncio.create_task(fetch(url, session, semaphore))
                    tasks.append(task)
        # 等待所有任务完成
        await asyncio.gather(*tasks)

asyncio.run(main())

这种方式不需要维护队列，代码更简洁，同样能实现“限制并发、不一次性加载所有URL到内存”的需求——逐行读取URL并创建任务，信号量会自动控制同时运行的任务数，不会导致内存暴涨。

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