异步API响应会被存储还是即时处理？及内存优化问询

问题描述 #

我调用一款API，它对任意合法参数都会返回极长的列表。我需要针对多个参数提取每个列表的首个元素，编写了如下代码，想咨询内存占用相关问题：

import asyncio
import aiohttp

URL = 'https://api2.binance.com/api/v3/trades?symbol='

symbols = ['BTCUSDT', 'ETHBTC', 'ETHUSDT']

async def get_trades(symbols):
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(URL + symbol) for symbol in symbols]
        all_trades = await asyncio.gather(*tasks)
        trades = [(await x.json())[0] for x in all_trades]
        return trades

await get_trades(symbols)

核心疑问 #

1. 运行get_trades时，所有参数对应的完整列表会被同时存储，还是会逐个处理后立即丢弃？
2. 换句话说，all_trades是先完整创建的响应对象集合再被遍历提取首元素，还是类似生成器的抽象对象？
3. 如果所有响应会先同时存储，会占用不必要的内存，毕竟我只需要每个列表的首个元素，这种情况下该如何修改代码消除低效问题？

解答 #

内存占用现状 #

你的代码中所有响应会先被完整存储，再统一处理：

• asyncio.gather(*tasks)会等待所有HTTP请求完成，将所有响应对象存入all_trades列表，此时每个请求的完整响应内容已经被下载并保存在内存中。
• 后续的列表推导式会逐个将响应解析为完整的JSON列表，这些完整列表也会暂时占用内存，直到提取出首元素后，原列表才会被垃圾回收。但在解析和提取的过程中，所有请求的完整JSON列表会同时存在于内存中，造成不必要的内存消耗。

优化方案 #

要避免同时存储所有完整响应和JSON列表，建议将解析、提取逻辑嵌入每个异步任务中，让每个请求完成后立即处理并丢弃无用数据，只保留需要的首元素：

import asyncio
import aiohttp

URL = 'https://api2.binance.com/api/v3/trades?symbol='

symbols = ['BTCUSDT', 'ETHBTC', 'ETHUSDT']

async def fetch_first_trade(session, symbol):
    async with session.get(URL + symbol) as response:
        # 解析响应并立即提取首个元素，不保留完整列表
        trade_list = await response.json()
        return trade_list[0]

async def get_trades(symbols):
    async with aiohttp.ClientSession() as session:
        # 每个任务负责单个请求的完整处理流程
        tasks = [fetch_first_trade(session, symbol) for symbol in symbols]
        # gather直接收集所有任务返回的首元素，而非完整响应对象
        first_trades = await asyncio.gather(*tasks)
        return first_trades

await get_trades(symbols)

进一步优化（可选） #

如果API返回的列表极端庞大，连解析整个列表都会造成明显内存压力，可以使用流式JSON解析库（如ijson），在读取响应内容的过程中只提取第一个元素，无需解析完整的JSON结构。示例代码如下：

import asyncio
import aiohttp
import ijson

async def fetch_first_trade_stream(session, symbol):
    async with session.get(URL + symbol) as response:
        # 流式读取响应内容，只提取第一个元素
        parser = ijson.items(response.content, 'item')
        return await anext(parser)

async def get_trades(symbols):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_first_trade_stream(session, symbol) for symbol in symbols]
        first_trades = await asyncio.gather(*tasks)
        return first_trades

await get_trades(symbols)

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