limitar a concorrência e controlar as solicitações por minuto com python aiohttp?
Existe um jogo chamado guild wars 2 e ele nos dá APIs para consultar quase tudo no banco de dados do jogo. Meu objetivo é usar python asyncio e aiohttp para escrever um rastreador simples e obter todas as informações dos itens do banco de dados do jogo Guild wars 2.
Eu escrevo um programa curto, é um trabalho, mas se comporta um pouco estranho, acho que aqui está algo que não entendo sobre compor a co-rotina.
Primeiro, fiz uma solicitação com o aplicativo Postman. E, no cabeçalho da resposta, há X-Rate-Limit-Limit, 600. Então, acho que as solicitações são limitadas a 600 por minuto?
aqui está minha pergunta.
1 、 Após o término do programa. Eu verifiquei alguns arquivos JSON e eles têm o mesmo conteúdo
[{"name": "Endless Fractal Challenge Mote Tonic", "description": "Transform into a Challenge Mote for 15 minutes or until hit. You cannot move while transformed."......
o que significa que a solicitação recebeu uma resposta ruim, mas não sei por quê.
2 、 Tentei asyncio.Semaphore, mas mesmo limitando a simultaneidade em 5, a solicitação vai além de 600 logo. Então tentei controlar o tempo adicionando um time.sleep (0.2) no final da função request_item. Eu acho que o time.sleep (0.2) irá suspender todo o processo python por 0,2 segundos e, na verdade, funcionou, mas após ser executado por algum tempo, o programa trava por um longo tempo e depois dá várias tentativas falhadas. Cada nova tentativa automática ainda falhou. Estou confuso sobre esse 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 eu movo time.sleep (0.2) para o loop for dentro da função request_item, todo o programa trava. Não tenho ideia do que estava acontecendo.
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
alguém poderia explicar isso um pouco? E existe uma solução melhor? Achei que existem algumas soluções, mas não consigo testar. como, obtenha o loop.time () e suspenda todo o loop de eventos a cada 600 solicitações. Ou adicione 600 solicitações a task_list e reúna-as como um grupo, depois de concluído, asyncio.run (get_item (req_ids)) novamente com outras 600 solicitações.
aqui está todo o meu código.
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))
Respostas
Você está usando em time.sleep()vez de await asyncio.sleep(). É a execução do buraco do bloco por N segundos e fazê-lo no lugar errado.
Aqui está o que aconteceu. Quando você corre
for item_id in item_ids:
req = request_item(session, item_id)
task = asyncio.create_task(req)
task_list.append(task)
Você apenas programa sua solicitação, mas não a executa. (por exemplo, você tem 1000 item_ids) Assim, você agenda 1000 tarefas e, quando executa, await asyncio.gather(*task_list)espera que todas essas 1000 tarefas sejam executadas. Eles vão atirar de uma vez.
Mas dentro de cada tarefa que você executa time.sleep(0.2)e tem que esperar 1000 * 0,2 segundos. Lembre-se de que todas as tarefas são executadas ao mesmo tempo e, em geral, em ordem aleatória. Então, você executa a tarefa 1 e espera 0,2 segundos, depois dispara a tarefa 2 e espera 0,2 segundos, a tarefa 999 é disparada e espera 0,2 segundos e assim por diante.
A solução mais simples será esperar um minuto após disparar 600 solicitações. Você precisa desacelerar por dentro get_item. Código de exemplo (não o testo):
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 = []
Eu recomendo que você use uma biblioteca asyncio-throttle .
PS. Com limite de taxa de 600 por minuto, não acho que você precise asyncio, porque tenho quase certeza de que 600 solicitações simultâneas serão executadas em 5 a 10 segundos. Verifique duas vezes se sua solicitação 600 leva mais de 1 minuto com o clássico requestscom tópicos.