由于公司业务要求,西瓜代理已经不满足需求,准备更换新的代理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_time
这里 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, None
这里需要传入一个参数 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: break
这里加了一些判断,如 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())
这里使用了 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": "235" }, { "valid_time": "17" }, { "valid_time": "71" }, { "valid_time": "427" }, { "valid_time": "176" }, { "valid_time": "297" }, { "valid_time": "235" }, { "valid_time": "54" }, { "valid_time": "465" }, { "valid_time": "397" }, { "valid_time": "196" }, { "valid_time": "114" }, { "valid_time": "37" }, { "valid_time": "367" }, { "valid_time": "69" }, { "valid_time": "98" }, { "valid_time": "536" }, { "valid_time": "96" }, { "valid_time": "7" }, { "valid_time": "100" }, { "valid_time": "526" }, { "valid_time": "636" }, { "valid_time": "119" }, { "valid_time": "227" }, { "valid_time": "108" }, { "valid_time": "70" }, { "valid_time": "95" }, { "valid_time": "66" }, { "valid_time": "111" }, { "valid_time": "115" }, { "valid_time": "466" }, { "valid_time": "28" }, { "valid_time": "121" }, { "valid_time": "296" }, { "valid_time": "417" }, { "valid_time": "127" }, { "valid_time": "490" }, { "valid_time": "201" }, { "valid_time": "344" }, { "valid_time": "236" }, { "valid_time": "233" }, { "valid_time": "33" }, { "valid_time": "19" }, { "valid_time": "229" }, { "valid_time": "385" }, { "valid_time": "259" }, { "valid_time": "234" }, { "valid_time": "118" }, { "valid_time": "396" }, { "valid_time": "52" }, { "valid_time": "225" }, { "valid_time": "58" }, { "valid_time": "591" }, { "valid_time": "124" }, { "valid_time": "21" }, { "valid_time": "52" }, { "valid_time": "19" }, { "valid_time": "214" }, { "valid_time": "46" }, { "valid_time": "90" }, { "valid_time": "489" }, { "valid_time": "62" }, { "valid_time": "104" }, { "valid_time": "46" }, { "valid_time": "40" }, { "valid_time": "62" }, { "valid_time": "32" }, { "valid_time": "28" }, { "valid_time": "76" }, { "valid_time": "439" }, { "valid_time": "53" }, { "valid_time": "51" }, { "valid_time": "139" }, { "valid_time": "39" }, { "valid_time": "213" }, { "valid_time": "173" }, { "valid_time": "160" }, { "valid_time": "305" }, { "valid_time": "85" }, { "valid_time": "154" }, { "valid_time": "2082" }, { "valid_time": "185" }, { "valid_time": "51" }, { "valid_time": "53" }, { "valid_time": "51" }, { "valid_time": "172" }, { "valid_time": "209" }, { "valid_time": "176" }, { "valid_time": "115" }, { "valid_time": "531" }, { "valid_time": "57" }, { "valid_time": "206" }, { "valid_time": "552" }, { "valid_time": "60" }, { "valid_time": "166" }, { "valid_time": "76" }, { "valid_time": "152" }, { "valid_time": "54" }, { "valid_time": "478" }, { "valid_time": "110" }, { "valid_time": "351" }, { "valid_time": "319" }, { "valid_time": "4" }, { "valid_time": "57" }, { "valid_time": "137" }, { "valid_time": "318" }, { "valid_time": "264" }, { "valid_time": "32" }, { "valid_time": "189" }, { "valid_time": "246" }, { "valid_time": "73" }, { "valid_time": "101" }, { "valid_time": "164" }, { "valid_time": "54" }, { "valid_time": "109" }, { "valid_time": "333" }, { "valid_time": "374" }, { "valid_time": "40" }, { "valid_time": "349" }, { "valid_time": "4" }, { "valid_time": "112" }, { "valid_time": "119" }, { "valid_time": "586" }, { "valid_time": "162" }, { "valid_time": "23" }, { "valid_time": "208" }, { "valid_time": "45" }, { "valid_time": "278" }, { "valid_time": "40" }, { "valid_time": "1191" }, { "valid_time": "43" }, { "valid_time": "478" }, { "valid_time": "178" }, { "valid_time": "168" }, { "valid_time": "90" }, { "valid_time": "181" }, { "valid_time": "476" }, { "valid_time": "353" }, { "valid_time": "349" }, { "valid_time": "533" }, { "valid_time": "648" }, { "valid_time": "38" }, { "valid_time": "549" }, { "valid_time": "298" }, { "valid_time": "296" }, { "valid_time": "334" }, { "valid_time": "42" }, { "valid_time": "363" }, { "valid_time": "235" } ] }[‘RECORDS‘] records = [int(i[‘valid_time‘]) for i in records_raw] print(records) bins = np.arange(0, 651, 10) # bins = np.arange(0, 651, 20) plt.hist(records, bins=bins, normed=0, facecolor="yellowgreen", edgecolor="black", alpha=0.7) plt.xlabel("有效时长(秒)") plt.ylabel("频数") plt.title("蜻蜓代理有效时长频率分布图") plt.xlim(0, 700) # 设置x轴分布范围 # plt.ylim(0, 0.01) # 设置y轴分布范围 plt.savefig(‘./蜻蜓代理有效时长频率分布图.jpg‘) plt.show()
7、思路参考
https://cuiqingcai.com/5094.html
https://www.zhihu.com/question/55807309?sort=created
https://www.zhihu.com/question/55807309/answer/755846298
https://www.zhihu.com/question/55807309/answer/294370242
8、谢谢
以上是为公司挑选付费代理的全过程,希望本文能够对您有所帮助。
原文地址:https://www.cnblogs.com/liangmingshen/p/11290963.html