如何在web3.py中实现批量请求？适配Alchemy API速率限制

原代码的问题 #

你写的代码里，get_balances函数用for循环逐个执行await w3.eth.get_balance(address)，这会导致每一个请求都要等前一个完成才发起下一个，本质还是同步串行执行，完全没用到异步的并发能力。

解决方案思路 #

1. 分批处理：把1000个地址分成每19个一组的批次，匹配Alchemy的每秒19次调用限制
2. 并发执行单批次内请求：用asyncio.gather同时发起批次内的所有请求，实现真正的异步并发
3. 控制批次间隔：每处理完一个批次后，计算耗时，如果不足1秒就等待剩余时间，确保每秒只处理一批，避免触发API限流

修改后的代码 #

from web3 import Web3
from web3.eth import AsyncEth
import time 
import pandas as pd
import asyncio

alchemy_url = "https://eth-mainnet.g.alchemy.com/v2/zCTn-wyjipF5DvGFVNEx_XqCKZakaB57"
w3 = Web3(Web3.AsyncHTTPProvider(alchemy_url), modules={'eth': (AsyncEth,)}, middlewares=[])

start = time.time()

df = pd.read_csv('Ethereum/ethereumaddresses.csv') 
wallet_addresses = df.loc[:, 'Address'].tolist()  # 转成列表方便分批
batch_size = 19  # 每秒处理的请求数
balance_storage = []

async def fetch_balance(address):
    # 封装单个地址的余额请求函数
    balance = await w3.eth.get_balance(address)
    balance_storage.append((address, balance))
    return address, balance

async def get_balances():
    # 把地址列表分成多个批次
    for i in range(0, len(wallet_addresses), batch_size):
        batch_start_time = time.time()
        batch_addresses = wallet_addresses[i:i+batch_size]
        
        # 并发执行当前批次的所有请求
        tasks = [fetch_balance(addr) for addr in batch_addresses]
        results = await asyncio.gather(*tasks)
        
        # 打印当前批次的结果
        for addr, balance in results:
            print(addr, balance)
        
        # 计算当前批次耗时，确保每秒只处理一批
        batch_duration = time.time() - batch_start_time
        if batch_duration < 1:
            await asyncio.sleep(1 - batch_duration)

asyncio.run(get_balances())

end = time.time()
total_time = end - start
print(f"It took {total_time:.2f} seconds to make {len(wallet_addresses)} API calls")

代码说明 #

• fetch_balance：封装单个地址的余额请求逻辑，方便批量创建异步任务
• get_balances：
  • 用切片将地址列表拆分为每19个一组的批次
  • 用asyncio.gather同时执行批次内所有任务，实现异步并发处理
  • 处理完批次后，判断本次耗时，若不足1秒则休眠剩余时间，严格控制每秒只处理一批请求
• 新增balance_storage列表存储所有地址的余额结果，方便后续导出或二次处理

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