Python的几个http库性能测试

#! /usr/bin/env python3
# # -*- coding: utf-8 -*-

from gevent import monkey as curious_george
curious_george.patch_all(thread=False, select=False)

import time
import aiohttp
import asyncio
import grequests
import http.client
import httplib2
import pycurl
import requests
import urllib
import urllib3

N = 1000
TMOUT = 2

t1 = int(time.time() * 1000)
for i in range(N):
    con = http.client.HTTPConnection("127.0.0.1", port=80, timeout=TMOUT)
    con.request("GET", "/")
    con.getresponse().read()
t2 = int(time.time() * 1000)
print("http.client %dms for %d packet" % ((t2 - t1), N))

t1 = int(time.time() * 1000)
for i in range(N):
    httplib2.Http(cache=None, timeout=TMOUT).request("http://127.0.0.1:80/", "GET")
t2 = int(time.time() * 1000)
print("httplib2 %dms for %d packet" % ((t2 - t1), N))

t1 = int(time.time() * 1000)
for i in range(N):
    requests.get("http://127.0.0.1:80/", timeout=TMOUT)
t2 = int(time.time() * 1000)
print("requests %dms for %d packet" % ((t2 - t1), N))

t1 = int(time.time() * 1000)
for i in range(N):
    urllib.request.urlopen("http://127.0.0.1:80/", timeout=TMOUT).read()
t2 = int(time.time() * 1000)
print("urllib %dms for %d packet" % ((t2 - t1), N))

t1 = int(time.time() * 1000)
for i in range(N):
    urllib3.PoolManager().request("GET", "http://127.0.0.1:80/", timeout=TMOUT)
t2 = int(time.time() * 1000)
print("urllib3 %dms for %d packet" % ((t2 - t1), N))

t1 = int(time.time() * 1000)
async def test_aiohttp():
    for i in range(N):
        async with aiohttp.ClientSession() as session:
            async with session.get("http://127.0.0.1:80/") as response:
                await response.text()
asyncio.get_event_loop().run_until_complete(test_aiohttp())
t2 = int(time.time() * 1000)
print("aiohttp %dms for %d packet" % ((t2 - t1), N))

t1 = int(time.time() * 1000)
grequests.map([grequests.get("http://127.0.0.1:80/") for i in range(N)])
t2 = int(time.time() * 1000)
print("grequests %dms for %d packet" % ((t2 - t1), N))

t1 = int(time.time() * 1000)
for i in range(N):
    curl = pycurl.Curl()
    curl.setopt(pycurl.CONNECTTIMEOUT, TMOUT)
    curl.setopt(pycurl.TIMEOUT, TMOUT)
    curl.setopt(pycurl.URL, "http://127.0.0.1:80/")
    curl.setopt(pycurl.WRITEFUNCTION, lambda x: None)
    curl.perform()
t2 = int(time.time() * 1000)
print("pycurl %dms for %d packet" % ((t2 - t1), N))


def get_with_pycurl(url, proxy=None):
    curl = pycurl.Curl()
    buff = io.BytesIO()
    hdr = io.BytesIO()
    if proxy:
        if not proxy.startswith("http"):
            proxy = "http://" + proxy
        curl.setopt(pycurl.PROXY, proxy)
        curl.setopt(pycurl.PROXYTYPE_HTTP, pycurl.PROXYTYPE_HTTP)
    curl.setopt(pycurl.USERAGENT, tmp_ua)
    curl.setopt(pycurl.CONNECTTIMEOUT, TMOUT)
    curl.setopt(pycurl.TIMEOUT, TMOUT)
    curl.setopt(pycurl.URL, url)
    curl.setopt(pycurl.WRITEFUNCTION, buff.write)
    curl.setopt(pycurl.WRITEHEADER, hdr)
    try:
        curl.perform()
        curl.close()
    except: # conn fail
        return None, None
    curl.close()
    status = int(hdr.getvalue().decode().split("\r\n")[0].split(" ")[1])
    body = buff.getvalue().decode()
    return status, body