Python并发编程队列与多线程最快发送http请求方式(队列并发)

队列+多线程

定义一个大小为 400 的队列，然后开启 200 个线程，每个线程都是不断的从队列中获取 url 并访问。

主线程读取文件中的 url 放入队列中，然后等待队列中所有的元素都被接收和处理完毕。代码如下：

from threading import Thread
import sys
from queue import Queue
import requests
concurrent = 200 
def doWork():
 while True:
  url = q.get()
  status, url = getStatus(url)
  doSomethingWithResult(status, url)
  q.task_done()
def getStatus(ourl):
 try:
  res = requests.get(ourl)
  return res.status_code, 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)

运行结果如下：

有没有 get 到新技能？

线程池

如果你使用线程池，推荐使用更高级的 concurrent.futures 库：

import concurrent.futures
import requests
out = []
CONNECTIONS = 100
TIMEOUT = 5
urls = []
with open("urllist.txt") as reader:
 for url in reader:
  urls.append(url.strip())
def load_url(url, timeout):
 ans = requests.get(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)
 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(data)

协程 + aiohttp

协程也是并发非常常用的工具了：

import asyncio
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 sys
 assert sys.version_info >= (3, 7), "Script requires Python 3.7+."
 with open("urllist.txt") as infile:
  urls = set(map(str.strip, infile))
 asyncio.run(make_requests(urls=urls))

grequests

这是个第三方库，目前有 3.8K 个星，就是 Requests + Gevent，让异步 http 请求变得更加简单。Gevent 的本质还是协程。

使用前：

pip install grequests

使用起来那是相当的简单：

import grequests
urls = []
with open("urllist.txt") as reader:
 for url in reader:
  urls.append(url.strip())
rs = (grequests.get(u) for u in urls)
for result in grequests.map(rs):
 print(result.status_code, result.url)

注意 grequests.map(rs) 是并发执行的。运行结果如下：

也可以加入异常处理：

>>> def exception_handler(request, exception):
... print("Request failed")
>>> 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)
Request failed
Request failed
[None, None, <Response [500]>]