在 Python 中发送 100,000 个 HTTP 请求的最快方法是什么？

Question

我正在打开一个包含 100,000 个 URL 的文件。我需要向每个 URL 发送一个 HTTP 请求并打印状态代码。我使用的是 Python 2.6，到目前为止，我研究了 Python 实现线程/并发的许多令人困惑的方式。我什至看过 python concurrence 库，但不知道如何正确编写这个程序。有没有人遇到过类似的问题？我想通常我需要知道如何在 Python 中尽可能快地执行数千个任务 - 我想这意味着“并发”。

Answer 1

无扭曲解决方案：

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)

这个比扭曲的解决方案稍微快一点，并且使用更少的 CPU。

Answer 2

自 2010 年发布此内容以来，情况发生了很大变化，我还没有尝试所有其他答案，但我尝试了一些，我发现使用 python3.6 最适合我。

我能够在 AWS 上每秒获取大约 150 个唯一域。

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')

Answer 3

使用tornado异步网络库的解决方案

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()

此代码使用非阻塞网络 I/O，没有任何限制。它可以扩展到数万个打开的连接。它将在单个线程中运行，但比任何线程解决方案都要快。结帐non-blocking I/O

Answer 4

我知道这是一个老问题，但在 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))

您可以阅读更多相关信息并查看示例 here。

Answer 5

线程绝对不是这里的答案。如果总体目标是“最快的方式”，它们将提供进程和内核瓶颈，以及不可接受的吞吐量限制。

一点点twisted 和它的异步HTTP 客户端会给你带来更好的结果。

Answer 6

使用grequests，它是请求+Gevent模块的组合。

GRequests 允许您将 Requests 与 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]>]

Answer 7

（下一个项目的自我说明）

仅使用 requests 的 Python 3 解决方案。它最简单，速度也很快，不需要多处理或复杂的异步库。

最重要的方面是重用连接，尤其是对于 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()

Answer 8

解决此问题的一个好方法是首先编写获得一个结果所需的代码，然后合并线程代码以并行化应用程序。

在一个完美的世界中，这仅仅意味着同时启动 100,000 个线程，这些线程将它们的结果输出到字典或列表中以供以后处理，但实际上，您可以以这种方式发出多少并行 HTTP 请求是有限的。在本地，您可以同时打开多少个套接字，您的 Python 解释器将允许多少个执行线程。如果所有请求都针对一台或多台服务器，则远程连接的数量可能会受到限制。这些限制可能需要您编写脚本，以便在任何时候只轮询一小部分 URL（正如另一位海报提到的，100 可能是一个不错的线程池大小，尽管您可能会发现您可以成功部署更多）。

您可以按照此设计模式解决上述问题：

1. 启动一个线程，该线程启动新的请求线程，直到当前运行的线程数（您可以通过 threading.active_count() 或通过将线程对象推入数据结构来跟踪它们）>= 您的最大同时请求数（说 100)，然后休眠一小段时间。当没有更多要处理的 URL 时，该线程应该终止。因此，线程将不断唤醒、启动新线程并休眠，直到您完成。
2. 让请求线程将其结果存储在某个数据结构中，以供以后检索和输出。如果您存储结果的结构是 CPython 中的 list 或 dict，您可以使用 safely append or insert unique items from your threads without locks，但如果您写入文件或需要更复杂的跨线程数据交互您应该使用一个互斥锁来保护这个状态免受破坏。

我建议您使用threading 模块。您可以使用它来启动和跟踪正在运行的线程。 Python 的线程支持是裸露的，但您对问题的描述表明它完全可以满足您的需求。

最后，如果您希望看到一个用 Python 编写的并行网络应用程序的非常简单的应用程序，请查看ssh.py。这是一个使用 Python 线程并行化许多 SSH 连接的小型库。该设计非常接近您的要求，您可能会发现它是一个很好的资源。

Answer 9

如果您希望获得可能的最佳性能，您可能需要考虑使用异步 I/O 而不是线程。与数千个 OS 线程相关的开销是不小的，Python 解释器中的上下文切换在它之上增加了更多。线程肯定会完成工作，但我怀疑异步路由会提供更好的整体性能。

具体来说，我建议使用 Twisted 库 (http://www.twistedmatrix.com) 中的异步 Web 客户端。众所周知，它有一个陡峭的学习曲线，但一旦你掌握了 Twisted 的异步编程风格，它就很容易使用。

关于 Twisted 的异步 Web 客户端 API 的 HowTo 可在以下位置获得：

http://twistedmatrix.com/documents/current/web/howto/client.html

Answer 10

解决方案：

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()

测试时间：

[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

Ping 时间：

bix.hu is ~10 ms away from me
godaddy.com: ~170 ms
google.com: ~30 ms

Answer 11

使用thread pool 是一个不错的选择，并且会使这变得相当容易。不幸的是，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()

希望这会有所帮助。

Answer 12

这个扭曲的异步网络客户端运行得非常快。

#!/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)

Answer 13

创建epoll对象，
打开多个客户端 TCP 套接字，
将它们的发送缓冲区调整为比请求标头多一点，
发送一个请求头——它应该是立即的，只是放入一个缓冲区， 在epoll 对象中注册套接字，
在epoll 上做.poll 对象，
从.poll 读取每个套接字的前 3 个字节，
将它们写到sys.stdout，然后是\n（不要刷新）， 关闭客户端套接字。

限制同时打开的套接字数量——在创建套接字时处理错误。仅当另一个套接字关闭时才创建一个新套接字。
调整操作系统限制。
尝试分叉几个（不是很多）进程：这可能有助于更有效地使用 CPU。

Answer 14

我发现使用tornado 包是实现这一目标的最快和最简单的方法：

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])

Answer 15

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。你也可以提供自己的异步事件循环！

使用 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。

也许它可以帮助我的回购，一个基本的例子， WRITING FAST ASYNC HTTP REQUESTS IN PYTHON

Answer 16

这是一个 "async" 解决方案，它不使用asyncio，但较低级别的机制asyncio 使用（在Linux 上）：select()。 （或者可能asyncio 使用poll 或epoll，但原理类似。）

它是example from 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()

Answer 17

对于您的情况，线程可能会起到作用，因为您可能会花费大部分时间等待响应。标准库中有一些有用的模块，例如 Queue，可能会有所帮助。

我之前做过类似的并行下载文件的事情，这对我来说已经足够了，但它没有达到你所说的规模。

如果您的任务更受 CPU 限制，您可能需要查看 multiprocessing 模块，该模块将允许您利用更多 CPU/内核/线程（更多进程不会相互阻塞，因为锁定是每个进程）

Answer 18

考虑使用 Windmill ，尽管 Windmill 可能无法执行那么多线程。

您可以在 5 台机器上使用手动 Python 脚本来实现，每台机器使用 40000-60000 端口连接出站，打开 100,000 个端口连接。

此外，使用线程良好的 QA 应用程序（例如 OpenSTA）进行示例测试可能会有所帮助，以便了解每个服务器可以处理多少。

另外，请尝试使用带有 LWP::ConnCache 类的简单 Perl。这样你可能会获得更高的性能（更多的连接）。

Answer 19

[工具]

Apache Bench 就是你所需要的。 - 用于测量 HTTP Web 服务器性能的命令行计算机程序 (CLI)

一篇不错的博文：https://www.petefreitag.com/item/689.cfm（来自 Pete Freitag）

Answer 20

最简单的方法是使用 Python 的内置线程库。 它们不是“真正的”/内核线程它们有问题（如序列化），但已经足够好了。您需要一个队列和线程池。一种选择是here，但自己编写很简单。您无法并行处理所有 100,000 个调用，但您可以同时触发 100 个（左右）它们。