python asyncio - RuntimeError: await wasn't used with future

Question

我想使用带有 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())

Answer 1

限制资源使用的信号量实际上是一个非常简单的概念。这类似于计算免费停车场。（汽车进入时为-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}")

python asyncio - RuntimeError: await wasn't used with future

python asyncio - RuntimeError: await wasn't used with future

python

semaphore

limit

python-asyncio