在获取 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, 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() 方法来访问页面
运行结果：
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。
免责声明：本站文章均来自网站采集或用户投稿，网站不提供任何软件下载或自行开发的软件！ 如有用户或公司发现本站内容信息存在侵权行为，请邮件告知！ 858582#qq.com
《魔兽世界》大逃杀！60人新游玩模式《强袭风暴》3月21日上线
暴雪近日发布了《魔兽世界》10.2.6 更新内容，新游玩模式《强袭风暴》即将于3月21 日在亚服上线，届时玩家将前往阿拉希高地展开一场 60 人大逃杀对战。
艾泽拉斯的冒险者已经征服了艾泽拉斯的大地及遥远的彼岸。他们在对抗世界上最致命的敌人时展现出过人的手腕，并且成功阻止终结宇宙等级的威胁。当他们在为即将于《魔兽世界》资料片《地心之战》中来袭的萨拉塔斯势力做战斗准备时，他们还需要在熟悉的阿拉希高地面对一个全新的敌人──那就是彼此。在《巨龙崛起》10.2.6 更新的《强袭风暴》中，玩家将会进入一个全新的海盗主题大逃杀式限时活动，其中包含极高的风险和史诗级的奖励。
《强袭风暴》不是普通的战场，作为一个独立于主游戏之外的活动，玩家可以用大逃杀的风格来体验《魔兽世界》，不分职业、不分装备（除了你在赛局中捡到的），光是技巧和战略的强弱之分就能决定出谁才是能坚持到最后的赢家。本次活动将会开放单人和双人模式，玩家在加入海盗主题的预赛大厅区域前，可以从强袭风暴角色画面新增好友。游玩游戏将可以累计名望轨迹，《巨龙崛起》和《魔兽世界：巫妖王之怒 经典版》的玩家都可以获得奖励。