在 Python 众多的 HTTP 客户端中，最有名的莫过于requests、aiohttp和httpx。
在不借助其他第三方库的情况下，requests只能发送同步请求；aiohttp只能发送异步请求；httpx既能发送同步请求，又能发送异步请求。
那么怎么选择呢
只发同步请求用requests，但可配合多线程变异步。只发异步请求用aiohttp，但可以配合await变同步。httpx可以发同步请求也可以异步，但是请求速度同步略差于requests，异步略差于aiohttp
Asyncio 的强大。但是，任何一种方案都不是完美的，都存在一定的局限性，Asyncio 同样如此。
实际使用中，想用好 Asyncio，特别是发挥其强大的功能，很多情况下必须得有相应的 Python 库支持。
比如
requests 库并不兼容 Asyncio，而 aiohttp 库兼容。
requests
这里先说requests
安装依赖
pip install requests
响应
响应的类型
#获取接口返回的字符串数据r.text#获取接口返回的json数据，即直接将json格式的数据转换为json对象r.json()#获取接口返回的二进制数据，假设二进制数据如果为图片可以继续转换成图片r.content#获取原始套接字,使用r.raw请在 requests 请求中加上参数 stream=Truer.raw
获取请求响应的其他信息
#获取状态码r.status_code#获取请求的urlr.url#获取指定cookies信息r.cookies['token']#获取访问服务器返回给我们的响应头部信息r.headers#获取指定访问服务器返回给我们的响应头部信息r.headers['Content-Type']#获取发送到服务器的请求的头部的信息r.request.headers
请求
GET请求
get请求：
res = requests.get(url,data=data,cookies=cookie,headers=header,verify=False,files=file)
data可传可不传，data是字典格式。
如果url是https的话，加上verify=False。如果url是http的话，可不加。
示例1
import requestsif __name__ == '__main__':r = requests.get("")print(r.text)
示例2
import requestsif __name__ == '__main__':data = {'username': 'admin', 'passwd': '123456'}r = requests.get("/login.json", params=data)print(r.status_code)print(r.json()["obj"])
POST请求
url_post = "/login.json"#不包含任何参数的请求r = requests.post(url_post)#不包含任何参数的请求,设置超时10s，timeout不设置则默认60sr = requests.post(url_post,timeout=10)#携带参数的请求，dict_param为参数字典，默认data=dict_param，使用data=则表示post的是form请求#即 application/x-www-form-urlencoded 。r = requests.post(url_post, data=dict_param)#携带参数的请求，dict_param为参数字典，使用json=则表示post的是json请求r = requests.post(url_post, json=dict_param)#携带参数的请求，body传字符串，这里是JSON字符串。r = requests.post(url_post, data=json.dumps(dict_param))#携带参数的请求，dict_param为参数字典，设置超时10s,并携带headers属性r = requests.post(url_post,data=dict_param,timeout=10,headers={'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X)'})#post请求上传文件url = 'http://apihost/upload/post'files = {'file': open('report.xls', 'rb')}r = requests.post(url, files=files)
其他类型请求
r = requests.put(url, data =dict_param)r = requests.delete(url)r = requests.head(url)r = requests.options(url)
代理
跨域的时候可以考虑代理访问，不管是post请求还是get请求，只需要添加proxies即可。
客户端开发时不用考虑跨域问题，没有必要设置代理访问。
proxies = {"http": "http://10.10.1.10:3128","https": "http://10.10.1.10:1080",}requests.get(url_get, proxies=proxies)
查看代理是否有效
和telnet作用一样
import telnetlibif __name__ == '__main__':try:telnetlib.Telnet('110.242.68.4', port='80', timeout=3)except:print('ip无效！')else:print('ip有效！')
异步请求
aiohttp 的代码与 httpx 异步模式的代码重合度90%，只不过把AsyncClient换成了ClientSession，
另外，在使用 httpx 时，当你await client.post时就已经发送了请求。但是当使用aiohttp时，只有在awiat resp.json()时才会真正发送请求。
aiohttp
import aiohttpimport asyncioasync def main():async with aiohttp.ClientSession() as client:resp = await client.post('/login.json',json={'ts': '-01-20 13:14:15'})result = await resp.json()print(result)asyncio.run(main())
httpx
import httpximport asyncioasync def main():async with httpx.AsyncClient() as client:resp = await client.post('/login.json',json={'ts': '-01-20 13:14:15'})result = resp.json()print(result)asyncio.run(main())
JSON
字符串转对象
import json# 一些 JSON:x = '{ "name":"Bill", "age":63, "city":"Seatle"}'# 解析 x:y = json.loads(x)# 结果是 Python 字典：print(y["age"])
对象转字符串
import json# Python 对象（字典）：x = {"name": "Bill","age": 63,"city": "Seatle"}# 转换为 JSON：y = json.dumps(x)# 结果是 JSON 字符串：print(y)
当 Python 转换为 JSON 时，Python 对象会被转换为 JSON（JavaScript）等效项：
异步IO（协程）和请求
pip install aiohttp
简单示例
import asyncioasync def test():await asyncio.sleep(3)return "123"async def main():result = await test()print(result)if __name__ == '__main__':asyncio.run(main())
异步请求
import asyncioimport aiohttpimport timeasync def download_one(url):async with aiohttp.ClientSession() as session:async with session.get(url) as resp:print('Read {} from {}'.format(resp.content_length, url))async def download_all(sites):tasks = [asyncio.ensure_future(download_one(site)) for site in sites]await asyncio.gather(*tasks)def main():sites = ['/index.html','/login.json','/userlist.json']start_time = time.perf_counter()loop = asyncio.get_event_loop()try:loop.run_until_complete(download_all(sites))finally:if loop.is_running():loop.close()end_time = time.perf_counter()print('Download {} sites in {} seconds'.format(len(sites), end_time - start_time))if __name__ == '__main__':main()
多线程
from concurrent.futures import ThreadPoolExecutorimport threading# 定义一个准备作为线程任务的函数def action(num):print(threading.current_thread().name)return num+100# 创建一个包含4条线程的线程池with ThreadPoolExecutor(max_workers=3) as pool:future1 = pool.submit(action, 1000)def get_result(future):print(f"单个任务返回:{future.result()}")# 为future1添加线程完成的回调函数future1.add_done_callback(get_result)print('------------------------------')# 使用线程执行map计算results = pool.map(action, (50, 100, 150))for r in results:print(f"多个任务返回:{r}")
异步 IO/多进程/多线程对比
异步 IO(asyncio)、多进程(multiprocessing)、多线程(multithreading)
IO 密集型应用CPU等待IO时间远大于CPU 自身运行时间，太浪费；
常见的 IO 密集型业务包括：浏览器交互、磁盘请求、网络爬虫、数据库请求等
Python 世界对于 IO 密集型场景的并发提升有 3 种方法：多进程、多线程、异步 IO(asyncio);
理论上讲asyncio是性能最高的，原因如下：
进程、线程会有CPU上下文切换进程、线程需要内核态和用户态的交互，性能开销大；而协程对内核透明的,只在用户态运行进程、线程并不可以无限创建，最佳实践一般是 CPU*2；而协程并发能力强，并发上限理论上取决于操作系统IO多路复用(Linux下是 epoll)可注册的文件描述符的极限
那asyncio的实际表现是否如理论上那么强，到底强多少呢？我构建了如下测试场景：
访问500台 DB，并sleep 100ms模拟业务查询
方法 1；顺序串行一台台执行方法 2：多进程方法 3：多线程方法 4：asyncio方法 5：asyncio+uvloop
最后的asyncio+uvloop和官方asyncio 最大不同是用 Cython+libuv 重新实现了asyncio 的事件循环(event loop)部分,
官方测试性能是 node.js的 2 倍，持平 golang。
以下测试代码需要 Pyhton3.7+：
顺序串行一台台执行
import recordsuser = "root"pwd = "123456"port = 3306hosts = [] # 500台 db列表def query(host):conn = records.Database(f'mysql+pymysql://{user}:{pwd}@{host}:{port}/mysql?charset=utf8mb4')rows = conn.query('select sleep(0.1);')print(rows[0])def main():for h in hosts:query(h)# main entranceif __name__ == '__main__':main()
多进程
from concurrent import futuresimport recordsuser = "root"pwd = "123456"port = 3306hosts = [] # 500台 db列表def query(host):conn = records.Database(f'mysql+pymysql://{user}:{pwd}@{host}:{port}/mysql?charset=utf8mb4')rows = conn.query('select sleep(0.1);')print(rows[0])def main():with futures.ProcessPoolExecutor() as executor:for future in executor.map(query,hosts):pass# main entranceif __name__ == '__main__':main()
多线程
from concurrent import futuresimport recordsuser = "root"pwd = "123456"port = 3306hosts = [] # 500台 db列表def query(host):conn = records.Database(f'mysql+pymysql://{user}:{pwd}@{host}:{port}/mysql?charset=utf8mb4')rows = conn.query('select sleep(0.1);')print(rows[0])def main():with .ThreadPoolExecutor() as executor:for future in executor.map(query,hosts):pass# main entranceif __name__ == '__main__':main()
asyncio
import asynciofrom databases import Database user = "root"pwd = "123456"port = 3306hosts = [] # 500台 db列表async def query(host):DATABASE_URL = f'mysql+pymysql://{user}:{pwd}@{host}:{port}/mysql?charset=utf8mb4'async with Database(DATABASE_URL) as database:query = 'select sleep(0.1);'rows = await database.fetch_all(query=query)print(rows[0])async def main():tasks = [asyncio.create_task(query(host)) for host in hosts]await asyncio.gather(*tasks)# main entranceif __name__ == '__main__':asyncio.run(main())
asyncio+uvloop
import asyncioimport uvloopfrom databases import Databaseuser = "root"pwd = "123456"port = 3306hosts = [] # 500台 db列表async def query(host):DATABASE_URL = f'mysql+pymysql://{user}:{pwd}@{host}:{port}/mysql?charset=utf8mb4'async with Database(DATABASE_URL) as database:query = 'select sleep(0.1);'rows = await database.fetch_all(query=query)print(rows[0])async def main():tasks = [asyncio.create_task(query(host)) for host in hosts]await asyncio.gather(*tasks)# main entranceif __name__ == '__main__':uvloop.install()asyncio.run(main())
运行时间对比
可以看出: 无论多进程、多进程还是asyncio都能大幅提升IO 密集型场景下的并发，但asyncio+uvloop性能最高，运行时间只有原始串行运行时间的 1/90，相差快 2 个数量级了！
