Python怎么搭建代理IP池實現檢測IP

這篇文章主要介紹Python怎么搭建代理IP池實現檢測IP，文中介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們一定要看完！

在獲取 IP 時，已經成功將各個網站的代理 IP 獲取下來了，然后就需要一個檢測模塊來對所有的代理進行一輪輪的檢測，檢測可用就設置為滿分，不可用分數就減 1，這樣就可以實時改變每個代理的可用情況，在獲取有效 IP 的時候只需要獲取分數高的 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, count, 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() 方法來訪問頁面

運行結果：

以上是“Python怎么搭建代理IP池實現檢測IP”這篇文章的所有內容，感謝各位的閱讀！希望分享的內容對大家有幫助，更多相關知識，歡迎關注億速云行業資訊頻道！