在獲取 IP 時，已經成功將各個網站的代理 IP 獲取下來了，然後就需要一個檢測模組來對所有的代理進行一輪輪的檢測，檢測可用就設定為滿分，不可用分數就減 1，這樣就可以實時改變每個代理的可用情況，在獲取有效 IP 的時候只需要獲取分數高的 IP

程式碼地址：https://github.com/Stevengz/Proxy_pool

另外三篇：
Python搭建代理IP池（一）- 獲取 IP
Python搭建代理IP池（二）- 儲存 IP
Python搭建代理IP池（四）- 介面設定與整體排程

由於代理 IP 的數量非常多，為了提高 IP 的檢測效率，這裡使用非同步請求庫 Aiohttp 來進行檢測。至於為什麼不用抓取時用的 Requests 庫，是因為它是一個同步請求庫，在發出一個請求之後需要等待網頁載入完成之後才能繼續執行程式。這個過程會阻塞在等待響應中，如果伺服器響應非常慢，一個請求就會需要十幾秒，程式不會繼續往下執行

非同步請求庫就解決了這個問題，在請求發出之後，程式可以繼續接下去執行其他的事情，當響應到達時會通知程式再去處理這個響應，這樣程式就沒有被阻塞，可以充分把時間和資源利用起來

新增設定

增加了幾個測試用的常量

setting.py

# 資料庫地址
HOST = '127.0.0.1'
# MySql埠
MYSQL_PORT = 3306
# MySQl使用者名稱、密碼
MYSQL_USERNAME = '***'
MYSQL_PASSWORD = '***'
# 資料庫名
SQL_NAME = 'test'

# 代理等級
MAX_SCORE = 30
MIN_SCORE = 0
INITIAL_SCORE = 10

# 代理池數量界限
POOL_UPPER_THRESHOLD = 1000

VALID_STATUS_CODES = [200,302]

# 測試API，建議抓哪個網站測哪個
TEST_URL = 'http://www.baidu.com'

# 最大批測試量
BATCH_TEST_SIZE = 30

VALID_STATUS_CODES 變數包含了正常的狀態碼，獲取 Response 後需要判斷響應的狀態，如果狀態碼在

VALID_STATUS_CODES 這個列表裡，則代表代理可用

定義方法

定義了一個類 Tester，init() 方法中建立了一個 MySqlClient 物件，供類中其他方法使用。接下來定義了一個 test_single_proxy() 方法，用來檢測單個代理的可用情況，其引數就是被檢測的代理

tester.py

import asyncio
import aiohttp
import time
import sys
from aiohttp import ClientError
from db import MySqlClient
from setting import *


class Tester(object):
 def __init__(self):
  self.mysql = MySqlClient()
 
 # 測試單個代理
 async def test_single_ip(self,ip):
  conn = aiohttp.TCPConnector(verify_ssl=False)
  async with aiohttp.ClientSession(connector=conn) as session:
   try:
    if isinstance(ip,bytes):
     ip = ip.decode('utf-8')
    real_ip = 'http://' + ip
    print('正在測試',ip)
    async with session.get(TEST_URL,proxy=real_ip,timeout=15,allow_redirects=False) as response:
     if response.status in VALID_STATUS_CODES:
      self.mysql.max(ip)
      print('代理可用',ip)
     else:
      self.mysql.decrease(ip)
      print('請求響應碼不合法 ',response.status,'IP',ip)
   except (ClientError,aiohttp.client_exceptions.ClientConnectorError,asyncio.TimeoutError,AttributeError):
    self.mysql.decrease(ip)
    print('代理請求失敗',ip)
 
 # 主函式
 def run(self):
  print('測試器開始執行')
  try:
   count = self.mysql.count()
   print('當前剩餘',count,'個代理')
   for i in range(0,BATCH_TEST_SIZE):
    start = i
    stop = min(i + BATCH_TEST_SIZE,count)
    print('正在測試第',start + 1,'-',stop,'個代理')
    test_ip_group = self.mysql.batch(start,stop)
    loop = asyncio.get_event_loop()
    tasks = [self.test_single_ip(ip_tuple[0]) for ip_tuple in test_ip_group]
    loop.run_until_complete(asyncio.wait(tasks))
    sys.stdout.flush()
    time.sleep(5)
  except Exception as e:
   print('測試器發生錯誤',e.args)


if __name__ == "__main__":
 test = Tester()
 test.run()

test_single_proxy() 方法前面加了 async 關鍵詞，代表這個方法是非同步的，方法內部首先建立了 Aiohttp 的 ClientSession 物件，此物件類似於 Requests 的 Session 物件，可以直接呼叫該物件的 get() 方法來訪問頁面

執行結果：

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支援我們。