如何在带累加器的while循环中应用asyncio优化分页API请求效率

原代码的核心问题是将「网络请求」和「数据处理」都放在串行循环里执行，每一步都要等上一步完全结束才会推进，导致大量时间浪费在等待上。我们可以把所有请求和后续处理任务都提交到调度器中，让调度器自动并行执行，无需手动等待单轮任务完成。

方案1：基于线程池最小改动实现（学习成本低） #

不需要引入asyncio，仅调整线程池的使用逻辑即可：把请求任务也放到线程池中执行，一次性提交所有请求任务，自动并行处理。

import concurrent.futures
import requests
import threading

supply = 3000
token_ids = []
# 加线程锁避免多线程同时写list出现数据丢失
lock = threading.Lock()

def fetch_and_process(offset):
    url = "url_1" + str(offset)
    response = requests.request("GET", url)
    a = response.json()
    assets = a["assets"]
    for an in assets:
        if an['sell_orders'] is None and an['last_sale'] is None and an['num_sales'] == 0:
            with lock:
                token_ids.append(str(an['token_id']))

if __name__ == "__main__":
    # 一次性生成所有offset参数
    offsets = range(0, supply, 50)
    # 可以根据站点限流情况调整max_workers数值，不要开太大触发反爬
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        # 提交所有请求+处理任务，线程池自动并行调度
        executor.map(fetch_and_process, offsets)
    
    print(token_ids)

改动说明： #

• 把单offset的请求+数据过滤逻辑封装成独立函数，直接提交到线程池
• 新增线程锁保证多线程写入token_ids时不会出现竞态条件导致数据丢失
• 原来每次循环都要新建销毁线程池，现在复用同一个线程池，额外减少了线程创建销毁的开销
• 注意：max_workers数值需要根据目标站点的限流规则调整，不要设置过大导致被站点封禁IP

方案2：基于asyncio+aiohttp实现异步IO（性能更高） #

如果想要更高的IO密集型任务性能，可以用异步IO实现：

import asyncio
import aiohttp

supply = 3000
token_ids = []
# 异步锁
lock = asyncio.Lock()

async def fetch_and_process(session, offset):
    url = "url_1" + str(offset)
    async with session.get(url) as response:
        a = await response.json()
        assets = a["assets"]
        for an in assets:
            if an['sell_orders'] is None and an['last_sale'] is None and an['num_sales'] == 0:
                async with lock:
                    token_ids.append(str(an['token_id']))

async def main():
    # 限制同时并发的请求数量，避免触发反爬
    connector = aiohttp.TCPConnector(limit=10)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = []
        for offset in range(0, supply, 50):
            tasks.append(fetch_and_process(session, offset))
        # 等待所有异步任务执行完成
        await asyncio.gather(*tasks)
    print(token_ids)

if __name__ == "__main__":
    asyncio.run(main())

改动说明： #

• 用异步HTTP库aiohttp替代同步的requests，充分发挥异步IO的性能优势
• 通过TCPConnector(limit)控制并发请求数，避免触发站点限流
• 异步场景下使用asyncio.Lock保证共享变量写入安全

额外优化建议 #

• 可以新增请求失败重试逻辑，避免偶发的网络波动导致任务失败
• 如果返回数据量很大，可以把token_ids的写入操作改成线程安全的队列，不需要每次加锁，性能会更好
• 可以根据站点的实际限流情况调整并发数，在不被封禁的前提下最大化拉取速度

内容的提问来源于stack exchange，提问作者Nguyen Bao Long