我想使用semaphore和gather()来限制api调用。我想我必须使用create_task(),但我得到一个运行时错误。"RuntimeError: await wasn't used with future". 我如何解决这个问题?
下面是代码。
import asyncio
# pip install git+https://github.com/sammchardy/python-binance.git@00dc9a978590e79d4aa02e6c75106a3632990e8d
from binance import AsyncClient
async def catch_up_aggtrades(client, symbols):
tasks = asyncio.create_task(get_historical_aggtrades(client, symbol) for symbol in symbols)
sem = asyncio.Semaphore(1)
async with sem:
await asyncio.gather(*tasks)
async def get_historical_aggtrades(client, symbol):
async for trade in client.aggregate_trade_iter(symbol, '1 day ago UTC'):
print(f"symbol {symbol}")
async def main():
client = await AsyncClient.create()
symbols = ['BTCUSDT', 'ETHUSDT', 'BNBUSDT']
await catch_up_aggtrades(client, symbols)
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
sempahore限制资源使用量其实是一个很简单的概念。它类似于计算免费停车场。汽车进入时为-1,离开时为+1)。当计数器降到零的时候,就会开始建立一个等待的汽车队列。
这意味着。
async with sem:
现行法规。
sem = asyncio.Semaphore(1)
async with sem:
await asyncio.gather(*tasks)
限制使用 asyncio.gather
到1次任务收集。它不限制任务,只是限制任务的聚集。由于 gather
反正只被调用一次,旗语不会改变任何东西。
你的程序可能会改成(包括评论中解决的问题)。
LIMIT = 1
async def catch_up_aggtrades(client, symbols):
sem = asyncio.Semaphore(LIMIT)
tasks = [asyncio.create_task(get_historical_aggtrades(client, symbol, sem)) for symbol in symbols]
await asyncio.gather(*tasks)
async def get_historical_aggtrades(client, symbol, sem):
async with sem:
async for trade in client.aggregate_trade_iter(symbol, '1 day ago UTC'):
print(f"symbol {symbol}")