我正在嘗試以這種方式進行異步 API 呼叫：
- 函式發送請求：
async def get_data(client, postdata): res = await client.post(url=_url, headers=_headers, data=postdata) return res
- 決議 JSON 的函式：
async def parse_res(client, postdata): res = await get_data(client, postdata) if bool(json.loads(res.text)['suggestions']): _oks = <...grab some JSON fields...> else: _oks = {} return _oks
- 我將這兩個函式包裝在 MAIN() 中：
async def main(_jobs): async with httpx.AsyncClient() as client: batch = [] calls = [] for job in _jobs: _postdata = '{ "query": "' job '" }' calls.append(asyncio.create_task(parse_res(client, _postdata))) batch = await asyncio.gather(*calls) return batch
然后運行 ??MAIN()
但是 API 可以快速處理大約 30-50 個（幾乎同時請求或拋出 429 HTTP 錯誤）。
所以我需要分批發送 30 個呼叫并分塊處理 10 000 個請求。
我如何分批處理 10 000（一萬）個 API 呼叫，每批 30 個？
uj5u.com熱心網友回復：
在這里派上用場的一個庫很時髦。它提供了各種用于處理序列的助手。其中之一就是大塊。如果總大小不劃分，這允許您最終將序列拆分為大小相等或更少的塊。
from funcy import chunks
result = []
for job_chunk in chunks(30, _jobs):
calls = [parse_res(client, '{ "query": "' job '" }') for job un job_chunk]
batch = await asyncio.gather(*calls)
result.extend(batch)
uj5u.com熱心網友回復：
您可以使用 Simon Hawe 的答案，但是這是一種不同的方法，無需使用外部庫
用于asyncio.Semaphore限制并發呼叫的數量，當信號量被釋放時，它將讓另一個函式運行。
import asyncio
sem = asyncio.Semaphore(30) # no. of simultaneous requests
async def get_data(client, postdata):
async with sem:
res = client.post(url=_url, headers=_headers, data=postdata)
return res
async def parse_res(client, postdata):
res = await get_data(client, postdata)
if bool(json.loads(res.text)['suggestions']):
_oks = <...grab some JSON fields...>
else:
_oks = {}
return _oks
async def main(_jobs: int):
async with httpx.AsyncClient() as client:
postdata = '{"query": "' job '"}'
calls = [
asyncio.create_task(parse_res(client, postdata)
for _ in range(_jobs)
]
return await asyncio.gather(*calls)
轉載請註明出處，本文鏈接：https://www.uj5u.com/qukuanlian/417771.html
標籤：
上一篇：異步等待未按預期回傳值
下一篇：如何逃離這個回呼地獄