httpx可以替代requests,它可以用来发送同步以及异步请求
使用方式基本和requests类似：
如：res=httpx.get('htttp://www.baidu.com')
assert res.status_code==200
*在公司遇到问题，因为公司网络使用了代理，这样写时，会报407错误
因此要把proxy加上
proxys={"http://":None}
不需要使用代理的时候
res=httpx.get('htttp://www.baidu.com',proxys=proxy)
assert res.status_code==200
异步请求：
import httpx
import asyncio
async def get_content(link):
async with httpx.AsyncClient() as client: #类似于requests的session
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
'Chrome/94.0.4606.81 Safari/537.36'}
#await client.get(login)
res = await client.get(link)
print(res.text)
links=['http://www.baidu.com','http://www.sina.com']
tasks=[]
for name, link in linklist:
task.append(get_content(link))
loop = asyncio.get_event_loop()
loop.run_until_complete(tasks)
如果是非主线程调用
loop=asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(tasks)
python3.7最新写法
import asyncio
async def main():
tasks = []
#设置link列表
for i in links
tasks.append(asyncio.create_task(get_content(i)))
await asyncio.wait(tasks)
if __name__ == '__main__':
asyncio.run(main())
使用gather
async def main():
tasks = []
#设置link列表
for i in links
tasks.append(get_content(i))
await asyncio.gather(*tasks)
*获取wait的执行结果
async def main():
tasks = []
for i in links:
tasks.append(asyncio.create_task(get_content(i)))
# 获取任务执行结果。
done, pending = await asyncio.wait(tasks)
#done, pending = await asyncio.wait(tasks, timeout=10) 设置超时时间
for task in done:
print(f"执行结果: {task.result()}")
if __name__ == '__main__':
asyncio.run(main())
*使用gather获取协程执行结果
import httpx
import asyncio
async def get_content(link):
async with httpx.AsyncClient() as client: #类似于requests的session
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
'Chrome/94.0.4606.81 Safari/537.36'}
#await client.get(login)
res = await client.get(link)
return res.status_code
async def main():
tasks = []
for i in links:
tasks.append(get_content(i))
results = await asyncio.gather(*tasks)
for result in results:
print(f"执行结果: {result}") #返回所有任务执行结果
*直接使用回调函数获取结果
- 回调函数
def callback(task):
print(f"执行结果:{task.result()}")
async def main():
tasks = []
for i in links:
task = asyncio.create_task(get_content(i))
# 注意这里，增加回调函数
task.add_done_callback(callback)
tasks.append(task)
