limitare la concorrenza e le richieste di controllo al minuto con python aiohttp?
C'è un gioco chiamato guild wars 2 e ci fornisce API per interrogare quasi tutto nel database del gioco. Il mio scopo è usare python asyncio e aiohttp per scrivere un semplice crawler e ottenere tutte le informazioni sugli oggetti dal database di gioco di guild wars 2.
Scrivo un breve programma, funziona, ma si comporta in modo strano, immagino che ci sia qualcosa che non capisco sulla composizione della coroutine.
Innanzitutto, ho effettuato una richiesta con l'app Postman. E, nell'intestazione della risposta, c'è X-Rate-Limit-Limit, 600. Quindi immagino che le richieste siano limitate a 600 al minuto?
ecco la mia domanda.
1 、 Al termine del programma. Ho controllato alcuni file JSON e hanno lo stesso contenuto
[{"name": "Endless Fractal Challenge Mote Tonic", "description": "Transform into a Challenge Mote for 15 minutes or until hit. You cannot move while transformed."......
il che significa che la richiesta ha ottenuto una risposta negativa, ma non so perché.
2 、 Ho provato asyncio.Semaphore, ma anche io limite la concorrenza a 5, la richiesta va ben presto oltre 600. Quindi ho provato a controllare il tempo aggiungendo un time.sleep (0.2) alla fine della funzione request_item. Immagino che time.sleep (0.2) sospenderà l'intero processo python per 0,2 secondi, e in realtà ha funzionato, ma dopo essere stato eseguito per un po 'di tempo il programma si blocca per molto tempo e poi ha dato molti tentativi falliti. Ogni nuovo tentativo automatico non è riuscito ancora. Sono confuso su questo comportamento.
async def request_item(session, item_id):
req_param_item = req_param
req_param_item['ids'] = item_id
# retry for 3 times when exception occurs.
for i in range(3):
try:
async with session.get(url_template, params=req_param_item) as response:
result = await response.json()
with open(f'item_info/{item_id}.json', 'w') as f:
json.dump(result, f)
print(item_id, 'done')
break
except Exception as e:
print(item_id, i, 'failed')
continue
time.sleep(0.2)
Quando sposto time.sleep (0.2) nel ciclo for all'interno della funzione request_item, l'intero programma si blocca. Non ho idea di cosa stesse succedendo.
async def request_item(session, item_id):
req_param_item = req_param
req_param_item['ids'] = item_id
for i in range(3):
try:
time.sleep(0.2)
async with session.get(url_template, params=req_param_item) as response:
result = await response.json()
with open(f'item_info/{item_id}.json', 'w') as f:
json.dump(result, f)
print(item_id, 'done')
break
except Exception as e:
print(item_id, i, 'failed')
continue
qualcuno potrebbe spiegarlo un po '? E c'è una soluzione migliore? Pensavo ci fossero alcune soluzioni, ma non posso testarle. come, ottenere il loop.time () e sospendere l'intero ciclo di eventi per ogni 600 richieste. Oppure, aggiungi 600 richieste a task_list e raccoglile come gruppo, dopo averlo fatto, asyncio.run (get_item (req_ids)) di nuovo con altre 600 richieste.
ecco tutto il mio codice.
import aiohttp
import asyncio
import httpx
import json
import math
import os
import time
tk = 'xxxxxxxx'
url_template = 'https://api.guildwars2.com/v2/items'
# get items list
req_param = {'access_token': tk}
item_list_resp = httpx.get(url_template, params=req_param)
items = item_list_resp.json()
async def request_item(session, item_id):
req_param_item = req_param
req_param_item['ids'] = item_id
for i in range(3):
try:
async with session.get(url_template, params=req_param_item) as response:
result = await response.json()
with open(f'item_info/{item_id}.json', 'w') as f:
json.dump(result, f)
print(item_id, 'done')
break
except Exception as e:
print(item_id, i, 'failed')
continue
# since the game API limit requests, I think it's ok to suspend program for a while
time.sleep(0.2)
async def get_item(item_ids: list):
task_list = []
async with aiohttp.ClientSession() as session:
for item_id in item_ids:
req = request_item(session, item_id)
task = asyncio.create_task(req)
task_list.append(task)
await asyncio.gather(*task_list)
asyncio.run(get_item(req_ids))
Risposte
Stai usando time.sleep()invece di await asyncio.sleep(). È un'esecuzione di block hole per N secondi e lo fa nel posto sbagliato.
Ecco cosa succede. Quando corri
for item_id in item_ids:
req = request_item(session, item_id)
task = asyncio.create_task(req)
task_list.append(task)
Devi solo programmare la tua richiesta, ma non eseguirla. (es. hai 1000 item_ids) Quindi pianifichi 1000 attività e quando esegui await asyncio.gather(*task_list)effettivamente attendi che tutte queste 1000 attività vengano eseguite. Spareranno immediatamente.
Ma all'interno di ogni attività che esegui time.sleep(0.2)e devi aspettare 1000 * 0,2 secondi. Ricorda che tutte le attività vengono eseguite contemporaneamente e in generale in ordine casuale. Quindi esegui l'attività 1 e attendi 0,2 sec, quindi spara l'attività 2 e aspetti 0,2 sec, quindi l'attività 999 spara e attendi 0,2 sec e così via.
La soluzione più semplice sarà attendere un minuto dopo aver inviato 600 richieste. Devi rallentare dentro get_item. Codice di esempio (non lo provo):
async def get_item(item_ids: list):
task_list = []
async with aiohttp.ClientSession() as session:
for n, item_id in enumerate(item_ids):
req = request_item(session, item_id)
task = asyncio.create_task(req)
task_list.append(task)
if n % 600 == 0:
await asyncio.gather(*task_list)
await asyncio.sleep(60)
task_list = []
Ti consiglio di usare una libreria asyncio-throttle .
PS. Con il limite di velocità di 600 al minuto non credo sia necessario asyncio, perché sono abbastanza sicuro che 600 richieste simultanee verranno eseguite in 5-10 secondi. Controlla due volte se la tua richiesta 600 richiede più di 1 minuto con il classico requestscon thread.