假如有一個(gè)文件,里面有 10 萬(wàn)個(gè) url,需要對(duì)每個(gè) url 發(fā)送 http 請(qǐng)求,并打印請(qǐng)求結(jié)果的狀態(tài)碼,如何編寫(xiě)代碼盡可能快的完成這些任務(wù)呢?
Python 并發(fā)編程有很多方法,多線程的標(biāo)準(zhǔn)庫(kù) threading,concurrency,協(xié)程 asyncio,當(dāng)然還有 grequests 這種異步庫(kù),每一個(gè)都可以實(shí)現(xiàn)上述需求,下面一一用代碼實(shí)現(xiàn)一下,本文的代碼可以直接運(yùn)行,給你以后的并發(fā)編程作為參考:
隊(duì)列+多線程定義一個(gè)大小為 400 的隊(duì)列,然后開(kāi)啟 200 個(gè)線程,每個(gè)線程都是不斷的從隊(duì)列中獲取 url 并訪問(wèn)。
主線程讀取文件中的 url 放入隊(duì)列中,然后等待隊(duì)列中所有的元素都被接收和處理完畢。代碼如下:
fromthreadingimportThread
importsys
fromqueueimportQueue
importrequests
concurrent=200
defdoWork():
whileTrue:
url=q.get()
status,url=getStatus(url)
doSomethingWithResult(status,url)
q.task_done()
defgetStatus(ourl):
try:
res=requests.get(ourl)
returnres.status_code,ourl
except:
return"error",ourl
defdoSomethingWithResult(status,url):
print(status,url)
q=Queue(concurrent*2)
foriinrange(concurrent):
t=Thread(target=doWork)
t.daemon=True
t.start()
try:
forurlinopen("urllist.txt"):
q.put(url.strip())
q.join()
exceptKeyboardInterrupt:
sys.exit(1)
運(yùn)行結(jié)果如下:
有沒(méi)有 get 到新技能?
線程池
如果你使用線程池,推薦使用更高級(jí)的 concurrent.futures 庫(kù):
importconcurrent.futures
importrequests
out=[]
CONNECTIONS=100
TIMEOUT=5
urls=[]
withopen("urllist.txt")asreader:
forurlinreader:
urls.append(url.strip())
defload_url(url,timeout):
ans=requests.get(url,timeout=timeout)
returnans.status_code
withconcurrent.futures.ThreadPoolExecutor(max_workers=CONNECTIONS)asexecutor:
future_to_url=(executor.submit(load_url,url,TIMEOUT)forurlinurls)
forfutureinconcurrent.futures.as_completed(future_to_url):
try:
data=future.result()
exceptExceptionasexc:
data=str(type(exc))
finally:
out.append(data)
print(data)
協(xié)程 + aiohttp
協(xié)程也是并發(fā)非常常用的工具了:
importasyncio
fromaiohttpimportClientSession,ClientConnectorError
asyncdeffetch_html(url:str,session:ClientSession,**kwargs)->tuple:
try:
resp=awaitsession.request(method="GET",url=url,**kwargs)
exceptClientConnectorError:
return(url,404)
return(url,resp.status)
asyncdefmake_requests(urls:set,**kwargs)->None:
asyncwithClientSession()assession:
tasks=[]
forurlinurls:
tasks.append(
fetch_html(url=url,session=session,**kwargs)
)
results=awaitasyncio.gather(*tasks)
forresultinresults:
print(f'{result[1]}-{str(result[0])}')
if__name__=="__main__":
importsys
assertsys.version_info>=(3,7),"ScriptrequiresPython3.7+."
withopen("urllist.txt")asinfile:
urls=set(map(str.strip,infile))
asyncio.run(make_requests(urls=urls))
grequests[1]
這是個(gè)第三方庫(kù),目前有 3.8K 個(gè)星,就是 Requests + Gevent[2],讓異步 http 請(qǐng)求變得更加簡(jiǎn)單。Gevent 的本質(zhì)還是協(xié)程。
使用前:
pipinstallgrequests
使用起來(lái)那是相當(dāng)?shù)暮?jiǎn)單:
importgrequests
urls=[]
withopen("urllist.txt")asreader:
forurlinreader:
urls.append(url.strip())
rs=(grequests.get(u)foruinurls)
forresultingrequests.map(rs):
print(result.status_code,result.url)
注意 grequests.map(rs) 是并發(fā)執(zhí)行的。運(yùn)行結(jié)果如下:
也可以加入異常處理:
>>>defexception_handler(request,exception):
...print("Requestfailed")
>>>reqs=[
...grequests.get('http://httpbin.org/delay/1',timeout=0.001),
...grequests.get('http://fakedomain/'),
...grequests.get('http://httpbin.org/status/500')]
>>>grequests.map(reqs,exception_handler=exception_handler)
Requestfailed
Requestfailed
[None,None,500]>]
最后的話
今天分享了并發(fā) http 請(qǐng)求的幾種實(shí)現(xiàn)方式,有人說(shuō)異步(協(xié)程)性能比多線程好,其實(shí)要分場(chǎng)景看的,沒(méi)有一種方法適用所有的場(chǎng)景,筆者就曾做過(guò)一個(gè)實(shí)驗(yàn),也是請(qǐng)求 url,當(dāng)并發(fā)數(shù)量超過(guò) 500 時(shí),協(xié)程明顯變慢。
-
編程
+關(guān)注
關(guān)注
88文章
3614瀏覽量
93686 -
HTTP
+關(guān)注
關(guān)注
0文章
504瀏覽量
31197 -
URL
+關(guān)注
關(guān)注
0文章
139瀏覽量
15328 -
python
+關(guān)注
關(guān)注
56文章
4792瀏覽量
84628
原文標(biāo)題:如何用最快的方式發(fā)送 10 萬(wàn)個(gè) http 請(qǐng)求?
文章出處:【微信號(hào):LinuxHub,微信公眾號(hào):Linux愛(ài)好者】歡迎添加關(guān)注!文章轉(zhuǎn)載請(qǐng)注明出處。
發(fā)布評(píng)論請(qǐng)先 登錄
相關(guān)推薦
評(píng)論