Какой самый быстрый способ отправить 100 000 HTTP-запросов в Python?
Я открываю файл с 100 000 URL. Мне нужно отправить HTTP-запрос на каждый URL и распечатать код состояния. Я использую Python 2.6, и до сих пор смотрел на многие запутанные способы, которыми Python реализует многопоточность / параллелизм. Я даже посмотрел на библиотеку Python Concurrence, но не могу понять, как правильно написать эту программу. Кто-нибудь сталкивался с подобной проблемой? Я предполагаю, что в целом мне нужно знать, как выполнить тысячи задач в Python как можно быстрее - я полагаю, это означает "одновременно".
21 ответ
Twistedless решение:
from urlparse import urlparse
from threading import Thread
import httplib, sys
from Queue import Queue
concurrent = 200
def doWork():
while True:
url = q.get()
status, url = getStatus(url)
doSomethingWithResult(status, url)
q.task_done()
def getStatus(ourl):
try:
url = urlparse(ourl)
conn = httplib.HTTPConnection(url.netloc)
conn.request("HEAD", url.path)
res = conn.getresponse()
return res.status, ourl
except:
return "error", ourl
def doSomethingWithResult(status, url):
print status, url
q = Queue(concurrent * 2)
for i in range(concurrent):
t = Thread(target=doWork)
t.daemon = True
t.start()
try:
for url in open('urllist.txt'):
q.put(url.strip())
q.join()
except KeyboardInterrupt:
sys.exit(1)
Это немного быстрее, чем скрученное решение и использует меньше ресурсов процессора.
Все изменилось с 2010 года, когда он был опубликован, и я не пробовал все остальные ответы, но попробовал несколько, и я нашел, что это работает лучше всего для меня, используя python3.6.
Мне удалось получить около 150 уникальных доменов в секунду, работающих на AWS.
import pandas as pd
import concurrent.futures
import requests
import time
out = []
CONNECTIONS = 100
TIMEOUT = 5
tlds = open('../data/sample_1k.txt').read().splitlines()
urls = ['http://{}'.format(x) for x in tlds[1:]]
def load_url(url, timeout):
ans = requests.head(url, timeout=timeout)
return ans.status_code
with concurrent.futures.ThreadPoolExecutor(max_workers=CONNECTIONS) as executor:
future_to_url = (executor.submit(load_url, url, TIMEOUT) for url in urls)
time1 = time.time()
for future in concurrent.futures.as_completed(future_to_url):
try:
data = future.result()
except Exception as exc:
data = str(type(exc))
finally:
out.append(data)
print(str(len(out)),end="\r")
time2 = time.time()
print(f'Took {time2-time1:.2f} s')
print(pd.Series(out).value_counts())
Я знаю, что это старый вопрос, но в Python 3.7 вы можете сделать это с помощью asyncio а также aiohttp,
import asyncio
import aiohttp
from aiohttp import ClientSession, ClientConnectorError
async def fetch_html(url: str, session: ClientSession, **kwargs) -> tuple:
try:
resp = await session.request(method="GET", url=url, **kwargs)
except ClientConnectorError:
return (url, 404)
return (url, resp.status)
async def make_requests(urls: set, **kwargs) -> None:
async with ClientSession() as session:
tasks = []
for url in urls:
tasks.append(
fetch_html(url=url, session=session, **kwargs)
)
results = await asyncio.gather(*tasks)
for result in results:
print(f'{result[1]} - {str(result[0])}')
if __name__ == "__main__":
import pathlib
import sys
assert sys.version_info >= (3, 7), "Script requires Python 3.7+."
here = pathlib.Path(__file__).parent
with open(here.joinpath("urls.txt")) as infile:
urls = set(map(str.strip, infile))
asyncio.run(make_requests(urls=urls))
Вы можете прочитать больше об этом и увидеть пример здесь.
Решение с использованием асинхронной сетевой библиотеки торнадо
from tornado import ioloop, httpclient
i = 0
def handle_request(response):
print(response.code)
global i
i -= 1
if i == 0:
ioloop.IOLoop.instance().stop()
http_client = httpclient.AsyncHTTPClient()
for url in open('urls.txt'):
i += 1
http_client.fetch(url.strip(), handle_request, method='HEAD')
ioloop.IOLoop.instance().start()
Темы абсолютно не ответ здесь. Они обеспечат узкие места как процесса, так и ядра, а также ограничения пропускной способности, которые неприемлемы, если общая цель - "самый быстрый путь".
Немного twisted и его асинхронный HTTP клиент даст вам гораздо лучшие результаты.
(Примечание для себя для следующего проекта)
Решение Python 3 с использованием только
requests.Он самый простой и быстрый, не требует многопроцессорности или сложных асинхронных библиотек.
Наиболее важным аспектом является повторное использование соединений, особенно для HTTPS (для открытия TLS требуется дополнительный круговой обход). Обратите внимание, что подключение зависит от субдомена. Если вы очищаете много страниц во многих доменах, вам следует отсортировать URL-адреса, чтобы максимально увеличить повторное использование соединения.
Он будет таким же быстрым, как любой асинхронный код, при наличии достаточного количества потоков. (запросы освобождают Python GIL при ожидании ответа).
[Код производственного уровня с некоторой регистрацией и обработкой ошибок]
import logging
import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
# source: https://stackoverflow.com/a/68583332/5994461
THREAD_POOL = 16
# This is how to create a reusable connection pool with python requests.
session = requests.Session()
session.mount(
'https://',
requests.adapters.HTTPAdapter(pool_maxsize=THREAD_POOL,
max_retries=3,
pool_block=True)
)
def get(url):
response = session.get(url)
logging.info("request was completed in %s seconds [%s]", response.elapsed.total_seconds(), response.url)
if response.status_code != 200:
logging.error("request failed, error code %s [%s]", response.status_code, response.url)
if 500 <= response.status_code < 600:
# server is overloaded? give it a break
time.sleep(5)
return response
def download(urls):
with ThreadPoolExecutor(max_workers=THREAD_POOL) as executor:
# wrap in a list() to wait for all requests to complete
for response in list(executor.map(get, urls)):
if response.status_code == 200:
print(response.content)
def main():
logging.basicConfig(
format='%(asctime)s.%(msecs)03d %(levelname)-8s %(message)s',
level=logging.INFO,
datefmt='%Y-%m-%d %H:%M:%S'
)
urls = [
"https://httpstat.us/200",
"https://httpstat.us/200",
"https://httpstat.us/200",
"https://httpstat.us/404",
"https://httpstat.us/503"
]
download(urls)
if __name__ == "__main__":
main()
Используйте grequests, это комбинация запросов + модуль Gevent.
GRequests позволяет использовать запросы с Gevent, чтобы легко выполнять асинхронные HTTP-запросы.
Использование простое:
import grequests
urls = [
'http://www.heroku.com',
'http://tablib.org',
'http://httpbin.org',
'http://python-requests.org',
'http://kennethreitz.com'
]
Создайте набор неотправленных запросов:
>>> rs = (grequests.get(u) for u in urls)
Отправьте их все одновременно:
>>> grequests.map(rs)
[<Response [200]>, <Response [200]>, <Response [200]>, <Response [200]>, <Response [200]>]
Хороший подход к решению этой проблемы состоит в том, чтобы сначала написать код, необходимый для получения одного результата, а затем включить многопоточный код для распараллеливания приложения.
В идеальном мире это просто означало бы одновременный запуск 100000 потоков, которые выводят свои результаты в словарь или список для последующей обработки, но на практике вы ограничены в количестве параллельных HTTP-запросов, которые вы можете выполнить таким образом. Локально, у вас есть ограничения на количество сокетов, которые вы можете открыть одновременно, сколько потоков выполнения ваш интерпретатор Python позволит. Удаленно, вы можете быть ограничены в количестве одновременных подключений, если все запросы направлены к одному серверу или ко многим. Эти ограничения, вероятно, потребуют от вас написания сценария таким образом, чтобы опрашивать только небольшую часть URL-адресов одновременно (100, как упоминалось в другом постере, вероятно, является приличным размером пула потоков, хотя вы можете обнаружить, что вы может успешно развернуть еще много).
Вы можете использовать этот шаблон для решения вышеуказанной проблемы:
- Запустите поток, который запускает новые потоки запросов, пока количество текущих запущенных потоков (вы можете отслеживать их с помощью threading.active_count() или путем помещения объектов потока в структуру данных) не будет>= вашего максимального числа одновременных запросов (скажем, 100) затем спит в течение короткого перерыва. Этот поток должен завершиться, когда больше нет URL-адресов для обработки. Таким образом, поток будет продолжать просыпаться, запускать новые потоки и спать до тех пор, пока вы не закончите.
- Пусть потоки запроса сохранят свои результаты в некоторой структуре данных для последующего поиска и вывода. Если структура, в которой вы храните результаты, является
listили жеdictв CPython вы можете безопасно добавлять или вставлять уникальные элементы из ваших потоков без блокировок, но если вы записываете в файл или требуете более сложного взаимодействия между потоками данных, вы должны использовать блокировку взаимного исключения для защиты этого состояния от повреждения.
Я хотел бы предложить вам использовать модуль потоков. Вы можете использовать его для запуска и отслеживания запущенных потоков. Поддержка потоков в Python отсутствует, но описание вашей проблемы предполагает, что ее вполне достаточно для ваших нужд.
Наконец, если вы хотите увидеть довольно простое приложение параллельного сетевого приложения, написанное на Python, посмотрите ssh.py. Это небольшая библиотека, которая использует потоки Python для распараллеливания многих SSH-соединений. Дизайн достаточно близок к вашим требованиям, поэтому вы можете найти его хорошим ресурсом.
Если вы хотите добиться максимальной производительности, возможно, вы захотите использовать асинхронный ввод-вывод, а не потоки. Накладные расходы, связанные с тысячами потоков ОС, нетривиальны, и переключение контекста в интерпретаторе Python добавляет еще больше. Работа с потоками, безусловно, сделает работу, но я подозреваю, что асинхронный маршрут обеспечит лучшую общую производительность.
В частности, я бы предложил асинхронный веб-клиент в библиотеке Twisted ( http://www.twistedmatrix.com/). У него, по общему признанию, крутая кривая обучения, но его довольно легко использовать, если вы хорошо разбираетесь в стиле асинхронного программирования Twisted.
Руководство по асинхронному веб-клиенту Twisted доступно по адресу:
http://twistedmatrix.com/documents/current/web/howto/client.html
Решение:
from twisted.internet import reactor, threads
from urlparse import urlparse
import httplib
import itertools
concurrent = 200
finished=itertools.count(1)
reactor.suggestThreadPoolSize(concurrent)
def getStatus(ourl):
url = urlparse(ourl)
conn = httplib.HTTPConnection(url.netloc)
conn.request("HEAD", url.path)
res = conn.getresponse()
return res.status
def processResponse(response,url):
print response, url
processedOne()
def processError(error,url):
print "error", url#, error
processedOne()
def processedOne():
if finished.next()==added:
reactor.stop()
def addTask(url):
req = threads.deferToThread(getStatus, url)
req.addCallback(processResponse, url)
req.addErrback(processError, url)
added=0
for url in open('urllist.txt'):
added+=1
addTask(url.strip())
try:
reactor.run()
except KeyboardInterrupt:
reactor.stop()
Testtime:
[kalmi@ubi1:~] wc -l urllist.txt
10000 urllist.txt
[kalmi@ubi1:~] time python f.py > /dev/null
real 1m10.682s
user 0m16.020s
sys 0m10.330s
[kalmi@ubi1:~] head -n 6 urllist.txt
http://www.google.com
http://www.bix.hu
http://www.godaddy.com
http://www.google.com
http://www.bix.hu
http://www.godaddy.com
[kalmi@ubi1:~] python f.py | head -n 6
200 http://www.bix.hu
200 http://www.bix.hu
200 http://www.bix.hu
200 http://www.bix.hu
200 http://www.bix.hu
200 http://www.bix.hu
Pingtime:
bix.hu is ~10 ms away from me
godaddy.com: ~170 ms
google.com: ~30 ms
pip install requests-threads
Пример использования с использованием async/await — отправить 100 одновременных запросов
from requests_threads import AsyncSession
session = AsyncSession(n=100)
async def _main():
rs = []
for _ in range(100):
rs.append(await session.get('http://httpbin.org/get'))
print(rs)
if __name__ == '__main__':
session.run(_main)
Этот пример работает только на Python 3. Вы также можете предоставить свой собственный цикл событий asyncio!
Пример использования с использованием Twisted
from twisted.internet.defer import inlineCallbacks
from twisted.internet.task import react
from requests_threads import AsyncSession
session = AsyncSession(n=100)
@inlineCallbacks
def main(reactor):
responses = []
for i in range(100):
responses.append(session.get('http://httpbin.org/get'))
for response in responses:
r = yield response
print(r)
if __name__ == '__main__':
react(main)
Этот пример работает как на Python 2, так и на Python 3.
Может быть, это может быть полезно, мой репозиторий, один простой пример, НАПИСАНИЕ БЫСТРЫХ ASYNC HTTP-ЗАПРОСОВ В PYTHON
Вот «асинхронное» решение, которое не использует, но механизм более низкого уровня использует (в Linux):select(). (Или, может быть
asyncioиспользует
poll, или же
epoll, но это аналогичный принцип.)
Это слегка измененная версия примера из PyCurl.
(Для простоты он запрашивает один и тот же URL-адрес несколько раз, но вы можете легко изменить его, чтобы получить несколько разных URL-адресов.)
(Другая небольшая модификация может привести к получению одного и того же URL-адреса снова и снова в виде бесконечного цикла. Подсказка: изменить
while urls and handlesк
while handles, и изменить
while nprocessed<nurlsк
while 1.)
import pycurl,io,gzip,signal, time, random
signal.signal(signal.SIGPIPE, signal.SIG_IGN) # NOTE! We should ignore SIGPIPE when using pycurl.NOSIGNAL - see the libcurl tutorial for more info
NCONNS = 2 # Number of concurrent GET requests
url = 'example.com'
urls = [url for i in range(0x7*NCONNS)] # Copy the same URL over and over
# Check args
nurls = len(urls)
NCONNS = min(NCONNS, nurls)
print("\x1b[32m%s \x1b[0m(compiled against 0x%x)" % (pycurl.version, pycurl.COMPILE_LIBCURL_VERSION_NUM))
print(f'\x1b[37m{nurls} \x1b[91m@ \x1b[92m{NCONNS}\x1b[0m')
# Pre-allocate a list of curl objects
m = pycurl.CurlMulti()
m.handles = []
for i in range(NCONNS):
c = pycurl.Curl()
c.setopt(pycurl.FOLLOWLOCATION, 1)
c.setopt(pycurl.MAXREDIRS, 5)
c.setopt(pycurl.CONNECTTIMEOUT, 30)
c.setopt(pycurl.TIMEOUT, 300)
c.setopt(pycurl.NOSIGNAL, 1)
m.handles.append(c)
handles = m.handles # MUST make a copy?!
nprocessed = 0
while nprocessed<nurls:
while urls and handles: # If there is an url to process and a free curl object, add to multi stack
url = urls.pop(0)
c = handles.pop()
c.buf = io.BytesIO()
c.url = url # store some info
c.t0 = time.perf_counter()
c.setopt(pycurl.URL, c.url)
c.setopt(pycurl.WRITEDATA, c.buf)
c.setopt(pycurl.HTTPHEADER, [f'user-agent: {random.randint(0,(1<<256)-1):x}', 'accept-encoding: gzip, deflate', 'connection: keep-alive', 'keep-alive: timeout=10, max=1000'])
m.add_handle(c)
while 1: # Run the internal curl state machine for the multi stack
ret, num_handles = m.perform()
if ret!=pycurl.E_CALL_MULTI_PERFORM: break
while 1: # Check for curl objects which have terminated, and add them to the handles
nq, ok_list, ko_list = m.info_read()
for c in ok_list:
m.remove_handle(c)
t1 = time.perf_counter()
reply = gzip.decompress(c.buf.getvalue())
print(f'\x1b[33mGET \x1b[32m{t1-c.t0:.3f} \x1b[37m{len(reply):9,} \x1b[0m{reply[:32]}...') # \x1b[35m{psutil.Process(os.getpid()).memory_info().rss:,} \x1b[0mbytes')
handles.append(c)
for c, errno, errmsg in ko_list:
m.remove_handle(c)
print('\x1b[31mFAIL {c.url} {errno} {errmsg}')
handles.append(c)
nprocessed = nprocessed + len(ok_list) + len(ko_list)
if nq==0: break
m.select(1.0) # Currently no more I/O is pending, could do something in the meantime (display a progress bar, etc.). We just call select() to sleep until some more data is available.
for c in m.handles:
c.close()
m.close()
[Инструмент]
Apache Bench - это все, что вам нужно. - Компьютерная программа командной строки (CLI) для измерения производительности веб-серверов HTTP.
Хорошая запись в блоге для вас: https://www.petefreitag.com/item/689.cfm (от Пита Фрейтага )
Создайте epoll объект,
открыть много клиентских TCP-сокетов,
настроить свои буферы отправки так, чтобы они были немного больше заголовка запроса,
отправьте заголовок запроса - он должен быть немедленным, просто поместив его в буфер, зарегистрируйте сокет в epoll объект,
делать .poll на epoll obect,
читать первые 3 байта из каждого сокета из .poll,
напиши им sys.stdout с последующим \n (не сбрасывать), закройте клиентский сокет.
Ограничить количество открытых сокетов одновременно - обрабатывать ошибки при создании сокетов. Создайте новый сокет, только если другой закрыт.
Настройте пределы ОС.
Попробуйте разделить несколько (не много) процессов: это может помочь немного эффективнее использовать процессор.
Использование пула потоков - хороший вариант, и это будет довольно просто. К сожалению, в python нет стандартной библиотеки, которая делает пулы потоков очень простыми. Но вот достойная библиотека, с которой стоит начать: http://www.chrisarndt.de/projects/threadpool/
Пример кода с их сайта:
pool = ThreadPool(poolsize)
requests = makeRequests(some_callable, list_of_args, callback)
[pool.putRequest(req) for req in requests]
pool.wait()
Надеюсь это поможет.
В вашем случае многопоточность, вероятно, сработает, поскольку вы, скорее всего, будете тратить большую часть времени на ожидание ответа. В стандартной библиотеке есть полезные модули, такие как Queue, которые могут помочь.
Раньше я делал аналогичные вещи с параллельной загрузкой файлов, и это было достаточно для меня, но не в том масштабе, о котором вы говорите.
Если ваша задача была более привязана к процессору, вы можете посмотреть на многопроцессорный модуль, который позволит вам использовать больше процессоров / ядер / потоков (больше процессов, которые не будут блокировать друг друга, поскольку блокировка выполняется для каждого процесса)
Этот скрученный асинхронный веб-клиент работает довольно быстро.
#!/usr/bin/python2.7
from twisted.internet import reactor
from twisted.internet.defer import Deferred, DeferredList, DeferredLock
from twisted.internet.defer import inlineCallbacks
from twisted.web.client import Agent, HTTPConnectionPool
from twisted.web.http_headers import Headers
from pprint import pprint
from collections import defaultdict
from urlparse import urlparse
from random import randrange
import fileinput
pool = HTTPConnectionPool(reactor)
pool.maxPersistentPerHost = 16
agent = Agent(reactor, pool)
locks = defaultdict(DeferredLock)
codes = {}
def getLock(url, simultaneous = 1):
return locks[urlparse(url).netloc, randrange(simultaneous)]
@inlineCallbacks
def getMapping(url):
# Limit ourselves to 4 simultaneous connections per host
# Tweak this number, but it should be no larger than pool.maxPersistentPerHost
lock = getLock(url,4)
yield lock.acquire()
try:
resp = yield agent.request('HEAD', url)
codes[url] = resp.code
except Exception as e:
codes[url] = str(e)
finally:
lock.release()
dl = DeferredList(getMapping(url.strip()) for url in fileinput.input())
dl.addCallback(lambda _: reactor.stop())
reactor.run()
pprint(codes)
Я обнаружил, что с помощью tornado package как самый быстрый и простой способ добиться этого:
from tornado import ioloop, httpclient, gen
def main(urls):
"""
Asynchronously download the HTML contents of a list of URLs.
:param urls: A list of URLs to download.
:return: List of response objects, one for each URL.
"""
@gen.coroutine
def fetch_and_handle():
httpclient.AsyncHTTPClient.configure(None, defaults=dict(user_agent='MyUserAgent'))
http_client = httpclient.AsyncHTTPClient()
waiter = gen.WaitIterator(*[http_client.fetch(url, raise_error=False, method='HEAD')
for url in urls])
results = []
# Wait for the jobs to complete
while not waiter.done():
try:
response = yield waiter.next()
except httpclient.HTTPError as e:
print(f'Non-200 HTTP response returned: {e}')
continue
except Exception as e:
print(f'An unexpected error occurred querying: {e}')
continue
else:
print(f'URL \'{response.request.url}\' has status code <{response.code}>')
results.append(response)
return results
loop = ioloop.IOLoop.current()
web_pages = loop.run_sync(fetch_and_handle)
return web_pages
my_urls = ['url1.com', 'url2.com', 'url100000.com']
responses = main(my_urls)
print(responses[0])
Фреймворк Scrapy решит вашу проблему быстро и профессионально. Он также будет кэшировать все запросы, чтобы вы могли повторно запустить неудачные только позже.
Сохраните этот скрипт какquotes_spider.py.
# quote_spiders.py
import json
import string
import scrapy
from scrapy.crawler import CrawlerProcess
from scrapy.item import Item, Field
class TextCleaningPipeline(object):
def _clean_text(self, text):
text = text.replace('“', '').replace('”', '')
table = str.maketrans({key: None for key in string.punctuation})
clean_text = text.translate(table)
return clean_text.lower()
def process_item(self, item, spider):
item['text'] = self._clean_text(item['text'])
return item
class JsonWriterPipeline(object):
def open_spider(self, spider):
self.file = open(spider.settings['JSON_FILE'], 'a')
def close_spider(self, spider):
self.file.close()
def process_item(self, item, spider):
line = json.dumps(dict(item)) + "\n"
self.file.write(line)
return item
class QuoteItem(Item):
text = Field()
author = Field()
tags = Field()
spider = Field()
class QuoteSpider(scrapy.Spider):
name = "quotes"
def start_requests(self):
urls = [
'http://quotes.toscrape.com/page/1/',
'http://quotes.toscrape.com/page/2/',
# ...
]
for url in urls:
yield scrapy.Request(url=url, callback=self.parse)
def parse(self, response):
for quote in response.css('div.quote'):
item = QuoteItem()
item['text'] = quote.css('span.text::text').get()
item['author'] = quote.css('small.author::text').get()
item['tags'] = quote.css('div.tags a.tag::text').getall()
item['spider'] = self.name
yield item
if __name__ == '__main__':
settings = dict()
settings['USER_AGENT'] = 'Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)'
settings['HTTPCACHE_ENABLED'] = True
settings['CONCURRENT_REQUESTS'] = 20
settings['CONCURRENT_REQUESTS_PER_DOMAIN'] = 20
settings['JSON_FILE'] = 'items.jl'
settings['ITEM_PIPELINES'] = dict()
settings['ITEM_PIPELINES']['__main__.TextCleaningPipeline'] = 800
settings['ITEM_PIPELINES']['__main__.JsonWriterPipeline'] = 801
process = CrawlerProcess(settings=settings)
process.crawl(QuoteSpider)
process.start()
с последующим
$ pip install Scrapy
$ python quote_spiders.py
Для точной настройки скребка отрегулируйтеCONCURRENT_REQUESTSиCONCURRENT_REQUESTS_PER_DOMAINнастройки соответственно.
Подумайте об использовании Windmill, хотя Windmill, вероятно, не может сделать столько потоков.
Вы можете сделать это с помощью сценария Python, запущенного вручную, на 5 машинах, каждая из которых соединяется с исходящими портами 40000-60000, открывая 100 000 соединений портов.
Кроме того, это могло бы помочь выполнить пример теста с приложением QA с хорошим потоком, таким как OpenSTA, чтобы получить представление о том, сколько может обрабатывать каждый сервер.
Также попробуйте изучить использование простого Perl с классом LWP::ConnCache. Таким образом, вы, вероятно, получите больше производительности (больше подключений).
Самый простой способ - использовать встроенную в Python библиотеку потоков. Они не являются "реальными" / потоками ядра. У них есть проблемы (например, сериализация), но они достаточно хороши. Вы хотите, чтобы очередь и пул потоков. Один вариант здесь, но это тривиально, чтобы написать свой собственный. Вы не можете распараллелить все 100 000 вызовов, но вы можете запустить 100 (или около того) из них одновременно.