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

问题描述：

我正在打开一个包含 100,000 个 URL 的文件。我需要向每个 URL 发送 HTTP 请求并打印状态代码。我使用的是 Python 2.6，到目前为止，我已经了解了 Python 实现线程/并发的许多令人困惑的方式。我甚至查看了 Python并发库，但无法弄清楚如何正确编写此程序。有人遇到过类似的问题吗？我想一般来说，我需要知道如何尽快在 Python 中执行数千个任务 - 我想这意味着“并发”。

解决方案 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。

解决方案 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="
")

    time2 = time.time()

print(f'Took {time2-time1:.2f} s')

解决方案 3：

我知道这是一个老问题，但是在 Python 3.7 中您可以使用asyncioand来做到这一点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))

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

解决方案 4：

使用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，没有任何限制。它可以扩展到数万个打开的连接。它将在单个线程中运行，但比任何线程解决方案都快得多。查看非阻塞 I/O

解决方案 5：

线程绝对不是解决问题的办法。如果总体目标是“最快的方式”，那么线程将带来进程和内核瓶颈，以及不可接受的吞吐量限制。

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

解决方案 6：

（提醒自己下一个项目）

仅使用 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()

解决方案 7：

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

解决方案 8：

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

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

您可以遵循此设计模式来解决上述问题：

1. 启动一个线程，该线程将启动新的请求线程，直到当前正在运行的线程数（您可以通过 threading.active_count() 或将线程对象推送到数据结构中来跟踪它们）>= 最大同时请求数（例如 100），然后短暂休眠。当没有更多 URL 需要处理时，此线程应终止。因此，线程将不断唤醒、启动新线程并休眠，直到您完成。

2. 让请求线程将其结果存储在某个数据结构中，以便稍后检索和输出。如果您存储结果的结构是list或dictCPython 中的，则可以安全地从线程中附加或插入唯一项而无需锁定，但如果您写入文件或需要更复杂的跨线程数据交互，则应使用互斥锁来保护此状态免受损坏。

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

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

解决方案 9：

如果您希望获得最佳性能，您可能需要考虑使用异步 I/O 而不是线程。数千个 OS 线程带来的开销并不小，而 Python 解释器中的上下文切换更是雪上加霜。线程肯定能完成工作，但我认为异步路由将提供更好的整体性能。

具体来说，我建议使用 Twisted 库中的异步 Web 客户端 ( http://www.twistedmatrix.com )。它确实学习起来比较困难，但一旦你掌握了 Twisted 的异步编程风格，它就很容易使用了。

Twisted 异步 Web 客户端 API 的 HowTo 可在以下位置找到：

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

解决方案 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

解决方案 11：

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。

也许这对我的仓库有帮助，一个基本的例子，
在 PYTHON 中编写快速异步 HTTP 请求

解决方案 12：

这是一个不使用 的“异步”asyncio解决方案，但较低级别的机制asyncio使用（在 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()

解决方案 13：

使用线程池是一个不错的选择，可以让这个过程变得相当简单。遗憾的是，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()

希望这有帮助。

解决方案 14：

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

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

如果您的任务更受 CPU 限制，您可能需要查看多处理模块，它将允许您使用更多的 CPU/核心/线程（由于锁定是每个进程的，因此更多进程不会相互阻塞）

解决方案 15：

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

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

解决方案 16：

创建epoll对象，

打开多个客户端 TCP 套接字，

调整它们的发送缓冲区使其比请求标头稍大，

发送请求标头 - 它应该是立即的，只需放入缓冲区，在epoll对象中注册套接字，在对象上

执行，
从每个套接字读取前 3 个字节，
将它们写入然后（不要刷新），关闭客户端套接字。.poll`epoll`
.poll
sys.stdout`
`

限制同时打开的套接字数量 — 处理创建套接字时的错误。仅当另一个套接字关闭时才创建新套接字。

调整操作系统限制。

尝试分叉成几个（不是很多）进程：这可能有助于更有效地使用 CPU。

解决方案 17：

我发现使用该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])

解决方案 18：

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)) + "
"
        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设置。

解决方案 19：

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

您可以在 5 台机器上使用手动 Python 脚本来完成此操作，每台机器使用端口 40000-60000 进行出站连接，打开 100,000 个端口连接。

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

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

解决方案 20：

[工具]

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

为您提供一篇精彩的博客文章：https ://www.petefreitag.com/item/689.cfm （来自Pete Freitag）

解决方案 21：

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