使用asyncio.gather处理协程时为什么需要create_task()及如何限制并发

1. 为什么不使用create_task()会抛出RuntimeWarning #

首先区分两个核心概念：

• 直接调用异步函数download_file(file)返回的是协程对象，本身是惰性执行的，只有被await时才会实际运行。如果协程对象直到被垃圾回收都没有被await，就会抛出你遇到的这个警告。
• asyncio.create_task()会将协程对象包装为Task，提交给当前事件循环立刻调度执行，事件循环会托管Task的整个生命周期，不需要手动await也会运行，因此不会触发未await的警告。

你之前不用create_task()时结果看似一致，是因为你测试时文件数量≤你设置的worker数10，队列里的所有协程都被worker里的await执行了；一旦文件数量超过10，队列中剩余的协程对象不会被处理，回收时就会抛出警告。

另外要特别指出：你最初的create_task()写法是完全错误的并发控制逻辑——你在往队列塞数据之前就调用create_task()，等于所有下载任务在入队时就已经被全部提交给事件循环调度，队列和worker完全没有起到限制并发的作用，如果有上千个文件会直接触发上千并发，很容易被存储接口限流或者打满带宽。

2. 基于asyncio.Queue的正确并发限制实现 #

你原来的worker逻辑错误，只会执行10个任务就退出，正确的队列模式应该将待处理的文件参数存入队列，由worker主动拉取任务执行，完整实现如下：

import asyncio
import pandas as pd

async def download_file(file):
    # 你的原有下载逻辑
    pass

async def worker(queue: asyncio.Queue, result_list: list):
    while True:
        file = await queue.get()
        try:
            df = await download_file(file)
            result_list.append(df)
        finally:
            # 标记队列任务处理完成
            queue.task_done()

async def main(files, concurrency=10):
    # 队列只存待处理的文件参数，不存协程/Task
    queue = asyncio.Queue()
    for file in files:
        queue.put_nowait(file)
    
    result_list = []
    # 启动指定数量的worker
    workers = [asyncio.create_task(worker(queue, result_list)) for _ in range(concurrency)]
    
    # 等待队列中所有任务处理完成
    await queue.join()
    
    # 关闭所有worker
    for w in workers:
        w.cancel()
    await asyncio.gather(*workers, return_exceptions=True)
    
    # 合并结果
    return pd.concat(result_list)

# 事件循环中调用
# df = await main(files, 10)

这个实现严格保证最多同时有concurrency个下载任务在运行，所有任务都会被处理，不会出现协程未await的警告。

3. 相关最佳实践 #

• 不要提前为所有异步任务调用create_task()，除非你明确不需要控制并发，否则会直接导致并发数不受限。
• 简单的并发限制场景优先选择你后续找到的信号量实现，代码更简洁易维护；队列实现更适合复杂场景，比如需要多生产者消费者、任务优先级调度、任务动态增减的场景。
• 所有协程对象必须要么被await，要么被包装为Task交由事件循环托管，避免出现未执行的协程和对应的警告。
• 使用return_exceptions=True时要注意对返回结果中的异常做判断处理，避免异常被当做正常数据传入pd.concat导致报错。

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