Ответ
Для оптимизации синхронных HTTP-запросов, особенно при выполнении множества обращений к одному хосту, следует применять несколько ключевых техник.
1. Использование пула соединений (Connection Pooling)
Создание нового TCP-соединения для каждого запроса — дорогостоящая операция. requests.Session решает эту проблему, переиспользуя существующие соединения.
Почему это работает: Устраняются накладные расходы на TCP handshake и TLS negotiation для каждого последующего запроса к тому же хосту.
import requests
urls = [
'https://api.example.com/data/1',
'https://api.example.com/data/2',
'https://api.example.com/data/3'
]
# Создаем сессию один раз
with requests.Session() as session:
for url in urls:
# Соединение переиспользуется
response = session.get(url)
print(f"{url}: {response.status_code}")2. Параллельное выполнение запросов с помощью потоков
Синхронные запросы по своей природе блокирующие. Пока один запрос ждет ответа от сети (I/O-bound операция), процессор простаивает. ThreadPoolExecutor позволяет выполнять несколько запросов одновременно в разных потоках.
import requests
from concurrent.futures import ThreadPoolExecutor
urls = ['https://example.com'] * 10
def fetch_url(url):
with requests.get(url) as response:
return response.status_code
# Выполняем до 5 запросов параллельно
with ThreadPoolExecutor(max_workers=5) as executor:
results = list(executor.map(fetch_url, urls))
print(results)3. Установка разумных таймаутов Чтобы избежать "зависания" приложения из-за медленного ответа сервера, всегда указывайте таймауты.
connect: время на установку соединения.read: время на ожидание первого байта ответа.
# Ожидать соединения не более 3 секунд, ответа - не более 10 секунд
response = requests.get('https://slow-service.com', timeout=(3, 10))4. Кэширование ответов
Если данные меняются редко, кэшируйте ответы, чтобы избежать повторных запросов. Для простых случаев подойдет functools.lru_cache.
from functools import lru_cache
import requests
@lru_cache(maxsize=128)
def get_user_data(user_id):
print(f"Fetching data for user {user_id}...")
return requests.get(f'https://api.example.com/users/{user_id}').json()
# Первый вызов выполнит HTTP-запрос
user1 = get_user_data(1)
# Второй вызов мгновенно вернет результат из кэша
user1_cached = get_user_data(1)Для высоконагруженных систем, где узким местом является ожидание I/O, стоит рассмотреть переход на асинхронный подход с asyncio и библиотеками aiohttp или httpx.