由於公司業務要求,西瓜代理已經不滿足需求,準備更換新的代理IP池,所以調研測試了一下市面上的各家付費代理(免費代理可用率低故不考慮),功能限制和價格情況等如何,以便從中挑選滿足要求的代理。 1、目標站 2、情報收集 整理套餐的價格和類型,API頻率,每秒提取上限,每天提取上限,使用時長等信息: (p ...
由於公司業務要求,西瓜代理已經不滿足需求,準備更換新的代理IP池,所以調研測試了一下市面上的各家付費代理(免費代理可用率低故不考慮),功能限制和價格情況等如何,以便從中挑選滿足要求的代理。
1、目標站
2、情報收集
整理套餐的價格和類型,API頻率,每秒提取上限,每天提取上限,使用時長等信息:
(ps:這裡建表格很費勁,所以傳的圖,如果圖片看不清,請單獨打開圖片或者下載,超清哦!)
備註:
- 使用時長和可用率來源 (1)目標網站上寫的(2)聯繫客服告知的
- 類型大體上可以分為兩類,API(請求對方介面給你返回IP)和HTTP隧道(用對方軟體或者直接給對方發請求,對方用代理IP請求待爬網站,最後對方給你返回response,免去囤、驗IP的步驟)。
-
說明1:HTTP隧道代理(動態轉發)
HTTP隧道代理:接入固定代理伺服器,動態轉發請求
統一入口,隨機動態出口,每一個請求一個隨機IP。
無須切換IP,連接上他們的代理伺服器後,每一個請求都是一個隨機IP,按併發數/轉發數計費。
動態轉發就是你把你代碼的proxy設置成他們的伺服器,那麼每次請求他們會使用一個隨機的IP,每次都會變化,也就是說,你只需設置一次代理,就可以得到隨機變化的IP,免去頻繁更換代理的麻煩,省去了代理池的維護這一個步驟。 -
說明2:
1.彙總依據:以上根據知乎代回答中,所有提到的代理名稱,
以及崔慶才做的測評中可用率在8成以上的代理相結合所做的彙總。
(已剔除免費代理--西刺,可用率低/穩定性差--快代理、大象代理)
2.因業務對IP需求量較大且使用頻繁,已剔除按數量計費的套餐
3、初步篩選
剔除條件:
可接受的價格(100+ ~ 300+),每天提取上限(大於3000),且不使用隧道和軟體代理(因為線上業務和爬蟲共用,所以不能把軟體放到線上伺服器上)。
最後篩選出來這幾家需要測評的:
4、測評
我把上面的套餐都買了一遍,以供下麵的評測使用(有的買的是一天使用權有的是網站有免費測試,所以此篇文章應該至少值幾十塊錢吧哈哈)。
(1)測評指標
我關心的測試指標為:重覆率,可用率(以訪問百度為基準),響應時間,穩定性
重覆率
因為線上業務需要用代理,會有封IP的情況,而且一封就是一天,所以重覆率一定要低。這裡以500個為基準,每個IP都存入資料庫,最後查重,用重覆的數量/總數量算出重覆率。
可用率
可用率就是提取的這些代理中可以正常使用的比率。假如我們無法使用這個代理請求某個網站或者訪問超時,那麼就代表這個代理不可用,在這裡我的測試樣本大小為 500,即提取 500 個代理,看看裡面可用的比率多少。
響應速度
響應速度可以用耗費時間來衡量,即計算使用這個代理請求網站一直到得到響應所耗費的時間。時間越短,證明代理的響應速度越快,這裡同樣是 500 個樣本,計算時只對正常可用的代理做統計,計算耗費時間的平均值。
穩定性
由於爬蟲時我們需要使用大量代理,如果一個代理響應速度特別快,很快就能得到響應,而下一次請求使用的代理響應速度特別慢,等了三十秒才得到響應,那勢必會影響爬取效率,所以我們需要看下商家提供的這些代理穩定性怎樣,總不能這一個特別快,下一個又慢的不行。所以這裡我們需要統計一下耗費時間的方差,方差越大,證明穩定性越差。
(2)測評標準
測試環境
公司的生產環境
測試原則
現取現測,即取一個測一個。現在很多付費代理網站都提供了 API 介面,我們可以一次性提取多個代理,但是這樣會導致一個問題,每個代理在提取出來的時候,商家是會儘量保證它的可用性的,但過一段時間,這個代理可能就不好用了,所以假如我們一次性提取出來了 100 個代理,但是這 100 個代理並沒有同時參與測試,後面的代理就會經歷一個的等待期,過一段時間再測這些代理的話,肯定會影響後半部分代理的有效性,所以這裡我們將提取的數量統一設置成 1,即請求一次介面獲取一個代理,然後立即進行測試,這樣可以保證測試的公平性,排除了不同代理有效期的干擾。
時間計算
計算程式請求之前和得到響應之後的時間差,這裡我們使用的測試 Python 庫是 requests,所以我們就計算發起請求和得到響應之間的時間差即可,時間計算方法如下所示:
start_time = time.time() requests.get(test_url, timeout=timeout, proxies=proxies) end_time = time.time() used_time = end_time - start_timeView Code
這裡 used_time 就是使用代理請求的耗時,這樣測試的就僅僅是發起請求到得到響應的時間。
測試鏈接
測試時我們也需要使用一個穩定的且沒有反爬蟲的鏈接,這樣可以排除伺服器的干擾,這裡我們使用百度來作為測試目標。
超時限制
在測試時免不了的會遇到代理請求超時的問題,所以這裡我們也需要統一一個超時時間,這裡設置為 60 秒,如果使用代理請求百度,60 秒還沒有得到響應,那就視為該代理無效。
測試數量
要做測評,那麼樣本不能太小,如只有十幾次測試是不能輕易下結論的,這裡我選取了一個適中的測評數量 500,即每個套餐獲取 500 個代理進行測試。
(3)測評過程
主要說一下測評的代碼邏輯,首先測的時候是取一個測一個的,所以這裡定義了一個 test_proxy() 方法:
test_url = 'https://www.baidu.com/' timeout = 60 def test_proxy(proxy): try: proxies = { 'https': 'http://' + proxy } start_time = time.time() requests.get(test_url, timeout=timeout, proxies=proxies) end_time = time.time() used_time = end_time - start_time print('Proxy Valid', 'Used Time:', used_time) return True, used_time except (ProxyError, ConnectTimeout, SSLError, ReadTimeout, ConnectionError): print('Proxy Invalid:', proxy) return False, NoneView Code
這裡需要傳入一個參數 proxy,代表一個代理,即 IP 加埠組成的代理,然後這裡使用了 requests 的 proxies 參數傳遞給 get() 方法。對於代理無效的檢測,這裡判斷了 ProxyError, ConnectTimeout, SSLError, ReadTimeout, ConnectionError 這幾種異常,如果發生了這些異常統統視為代理無效,返回錯誤。如果在 timeout 60 秒內得到了響應,那麼就計算其耗費時間並返回。
在主程式里,就是獲取 API 然後統計結果了,代碼如下:
max = 500 def main(): print('Testing') used_time_list = [] valid_count = 0 total_count = 0 while True: flag, result = get_page(api_url) if flag: proxy = result.strip() if is_proxy(proxy): total_count += 1 print('Testing proxy', proxy) test_flag, test_result = test_proxy(proxy=proxy) if test_flag: valid_count += 1 used_time_list.append(test_result) stats_result(used_time_list, valid_count, total_count) time.sleep(wait) if total_count == max: breakView Code
這裡加了一些判斷,如 is_proxy() 方法判斷了獲取的是不是符合有效的代理規則,即判斷它是不是 IP 加埠的形式,這樣可以排除 API 返回一些錯誤信息的干擾。另外這裡設置了 total_count 和 valid_count 變數,只有符合代理規則的代理參與了測試,這樣才算一次有效測試,total_count 加一,如果測試可用,那麼 valid_count 加一併記錄耗費時間。最後調用了 stats_results 方法進行了統計:
import numpy as np def stats_result(used_time_list, valid_count, total_count): if not used_time_list or not total_count: return used_time_array = np.asarray(used_time_list, np.float32) print('Total Count:', total_count, 'Valid Count:', valid_count, 'Valid Percent: %.2f%%' % (valid_count * 100.0 / total_count), 'Used Time Mean:', used_time_array.mean(), 'Used Time Var', used_time_array.var())View Code
這裡使用了 Numpy 來統計了耗費時間的均值和方差,分別反映代理的響應速度和穩定性。
全部測評代碼:
import requests import time import numpy as np from requests.exceptions import ProxyError, ConnectTimeout, ReadTimeout, SSLError, ConnectionError import re test_url = 'https://www.baidu.com/' timeout = 4 max_ = 500 def get_page(url): try: headers = { 'User-Agent': 'User-Agent:Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_2) \ AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36' } response = requests.get(url, headers=headers) if response.status_code == 200: return True, response.text except ConnectionError: return False, None def is_proxy(proxy): if re.match('\d+.\d+.\d+.\d+:\d+', proxy): return True return False def test_proxy(proxy=None, proxies=None): try: if not proxies: proxies = { 'https': 'http://' + proxy } start_time = time.time() requests.get(test_url, timeout=timeout, proxies=proxies) end_time = time.time() used_time = end_time - start_time print('Proxy Valid', 'Used Time:', used_time) return True, used_time except (ProxyError, ConnectTimeout, SSLError, ReadTimeout, ConnectionError): print('Proxy Invalid:', proxy) return False, None def stats_result(test_object, used_time_list, valid_count, total_count): if not used_time_list or not total_count: return used_time_array = np.asarray(used_time_list, np.float32) print('Total Count:', total_count, 'Valid Count:', valid_count, 'Valid Percent: %.2f%%' % (valid_count * 100.0 / total_count), 'Used Time Mean:', used_time_array.mean(), 'Used Time Var', used_time_array.var()) now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) with open("%s測試記錄.txt" % test_object, "a+", encoding='utf-8') as f: f.write('\n') f.write( 'Total Count: %s, Valid Count: %s, Valid Percent: %.2f%%, Used Time Mean: %s, Used Time Var: %s 【%s】' % ( total_count, valid_count, (valid_count * 100.0 / total_count), used_time_array.mean(), used_time_array.var(), now_time))public.py
from public import get_page, test_proxy, stats_result, is_proxy import time import pymysql db = pymysql.connect(host='localhost', user='root', password='root', port=3306, db='ip_pool_test') cursor = db.cursor() a = 'http://piping.mogumiao.com/proxy/api' api_url = a + '/get_ip_bs?appKey=c33c9a92fd6c426eb01d7a94b4ee6ab2&count=1&expiryDate=0&format=2&newLine=2' wait = 1 max_ = 1500 test_object = '蘑菇代理' def save_db(ip, batch): sql = 'INSERT INTO ip_pool_test_mogu(ip, batch) values(%s, %s)' cursor.execute(sql, (ip, batch)) db.commit() print('【%s】已存入庫' % ip) def main(): print('Testing 蘑菇代理') used_time_list = [] valid_count = 0 total_count = 0 while True: flag, result = get_page(api_url) if flag: proxy = result.strip() if is_proxy(proxy): total_count += 1 print('Testing proxy', proxy) test_flag, test_result = test_proxy(proxy=proxy) if test_flag: valid_count += 1 used_time_list.append(test_result) stats_result(test_object, used_time_list, valid_count, total_count) save_db(proxy, total_count) time.sleep(wait) if total_count == max_: break if __name__ == '__main__': main()test_mogu.py
from public import get_page, test_proxy, stats_result, is_proxy import time import pymysql db = pymysql.connect(host='localhost', user='root', password='root', port=3306, db='ip_pool_test') cursor = db.cursor() api_url = 'http://http.tiqu.qingjuhe.cn/getip?num=1&type=1&pack=35619&port=11&lb=1&pb=45®ions=' wait = 1.5 max_ = 1500 test_object = '太陽代理' def save_db(ip, batch): sql = 'INSERT INTO ip_pool_test_sun(ip, batch) values(%s, %s)' cursor.execute(sql, (ip, batch)) db.commit() print('【%s】已存入庫' % ip) def main(): print('Testing 太陽代理') used_time_list = [] valid_count = 0 total_count = 0 while True: flag, result = get_page(api_url) if flag: proxy = result.strip() if is_proxy(proxy): total_count += 1 print('Testing proxy', proxy) test_flag, test_result = test_proxy(proxy=proxy) if test_flag: valid_count += 1 used_time_list.append(test_result) stats_result(test_object, used_time_list, valid_count, total_count) save_db(proxy, total_count) time.sleep(wait) if total_count == max_: break if __name__ == '__main__': main()test_sun.py
from public import get_page, test_proxy, stats_result, is_proxy import time import pymysql db = pymysql.connect(host='localhost', user='root', password='root', port=3306, db='ip_pool_test') cursor = db.cursor() api_url = 'http://api.xdaili.cn/xdaili-api//greatRecharge/getGreatIp?spiderId=e79969dc35134ab6b85dd2e490475537&orderno=YZ2019827425QSYLYn&returnType=1&count=10' wait = 5 max_ = 1000 test_object = '訊代理' def save_db(ip, batch): sql = 'INSERT INTO ip_pool_test_xun(ip, batch) values(%s, %s)' cursor.execute(sql, (ip, batch)) db.commit() print('【%s】已存入庫' % ip) def gen_ip_list(result): ip_list = result.split() return ip_list def main(): print('Testing 訊代理') used_time_list = [] valid_count = 0 total_count = 0 while True: flag, result = get_page(api_url) start = time.time() if flag: ip_list = gen_ip_list(result) for proxy in ip_list: if is_proxy(proxy): total_count += 1 print('Testing proxy', proxy) test_flag, test_result = test_proxy(proxy=proxy) if test_flag: valid_count += 1 used_time_list.append(test_result) stats_result(test_object, used_time_list, valid_count, total_count) save_db(proxy, total_count) end = time.time() used_time = end - start print('一次10個IP,測完用時:%s' % used_time) print('--' * 100) time.sleep(wait) if total_count == max_: break if __name__ == '__main__': main() # gen_ip_list()test_xdaili.py
from public import get_page, test_proxy, stats_result, is_proxy import time import pymysql db = pymysql.connect(host='localhost', user='root', password='root', port=3306, db='ip_pool_test') cursor = db.cursor() api_url = 'https://proxy.horocn.com/api/proxies?order_id=PXXN1640735352167438&num=10&format=text&line_separator=win' wait = 6 max_ = 1500 test_object = '蜻蜓代理' def save_db(ip, batch): sql = 'INSERT INTO ip_pool_test_qingting(ip, batch) values(%s, %s)' cursor.execute(sql, (ip, batch)) db.commit() print('【%s】已存入庫' % ip) def gen_ip_list(result): ip_list = result.split() return ip_list def main(): print('Testing 蜻蜓代理') used_time_list = [] valid_count = 0 total_count = 0 while True: flag, result = get_page(api_url) start = time.time() if flag: ip_list = gen_ip_list(result) for proxy in ip_list: if is_proxy(proxy): total_count += 1 print('Testing proxy', proxy) test_flag, test_result = test_proxy(proxy=proxy) if test_flag: valid_count += 1 used_time_list.append(test_result) stats_result(test_object, used_time_list, valid_count, total_count) save_db(proxy, total_count) end = time.time() used_time = end - start print('一次10個IP,測完用時:%s' % used_time) print('--' * 100) time.sleep(wait) if total_count == max_: break if __name__ == '__main__': main() # gen_ip_list()test_qingting.py
from public import get_page, test_proxy, stats_result, is_proxy import time import pymysql db = pymysql.connect(host='localhost', user='root', password='root', port=3307, db='ip_pool_test') cursor = db.cursor() a = 'http://ged.ip3366.net/api' api_url = a + '/?key=20190802162116640&getnum=10&anonymoustype=3&filter=1&area=1&order=1&proxytype=1' wait = 5 max_ = 1500 test_object = '雲代理' def save_db(ip, batch): sql = 'INSERT INTO ip_pool_test_yun(ip, batch) values(%s, %s)' cursor.execute(sql, (ip, batch)) db.commit() print('【%s】已存入庫' % ip) def gen_ip_list(result): ip_list = result.split() return ip_list def main(): print('Testing 雲代理') used_time_list = [] valid_count = 0 total_count = 0 while True: flag, result = get_page(api_url) start = time.time() if flag: ip_list = gen_ip_list(result) for proxy in ip_list: if is_proxy(proxy): total_count += 1 print('Testing proxy', proxy) test_flag, test_result = test_proxy(proxy=proxy) if test_flag: valid_count += 1 used_time_list.append(test_result) stats_result(test_object, used_time_list, valid_count, total_count) save_db(proxy, total_count) end = time.time() used_time = end - start print('一次10個IP,測完用時:%s' % used_time) print('--' * 100) time.sleep(wait) if total_count == max_: break if __name__ == '__main__': main() # gen_ip_list()test_yundaili.py
(4)測評結果
5、最終確定
根據提取上限、重覆率、可用率、響應時間,價格,最終確定選用蜻蜓代理。
6、有效時長
最後,由於線上代理IP的API更新代理池的時間策略需要調整,所以需要確定有效時長分佈是怎樣的。
用matplotlib畫了有效時長的分佈直方圖。
確定了強制更新策略是一分鐘。
import numpy as np import matplotlib import matplotlib.pyplot as plt # 設置matplotlib正常顯示中文和負號 matplotlib.rcParams['font.sans-serif'] = ['SimHei'] # 用黑體顯示中文 matplotlib.rcParams['axes.unicode_minus'] = False # 正常顯示負號 records_raw = { "RECORDS": [ { "valid_time": "175" }, { "valid_time": "53" }, { "valid_time": "271" }, { "valid_time": "414" }, { "valid_time": "54" }, { "valid_time": "76" }, { "valid_time": "231" }, { "valid_time": "343" }, { "valid_time": "403" }, { "valid_time": "101" }, { "valid_time": "359" }, { "valid_time": "194" }, { "valid_time": "233" }, { "valid_time": "388" }, { "valid_time": "184" }, { "valid_time": "246" }, { "valid_time": "32" }, { "valid_time": "171" }, { "valid_time": "405" }, { "valid_time": "153" }, { "valid_time": "281" }, { "valid_time": "534" }, { "valid_time": "416" }, { "valid_time": "406" }, { "valid_time": "