ip代理池（程序员）

无私分享全套Python爬虫干货，如果你也想学习Python，@ 私信小编获取

做爬虫时，遇到访问太频繁IP被封是难以避免的，而本地单个IP是不足以进行大规模爬取，并且自己并不想购买付费代理，在这里构建一个IP代理池是非常有必要的。

为了获取可用性ip更加方便一点，这里利用flask框架做了一个API，这个接口连接着redis数据库；当打开这个api开关时,直接从redis数据库中随机获取一个分数等级高的ip,用户可以通过访问这个接口（提供的url）就可以获取相应的IP代理。

代理池的运作关系如下图：

代码实现部分

在构建ip代理池过程中，如果使用redis数据库和flask框架，则需要提前完成相关配置。

部分代码：

class Tester(object):
def __init__(self):
self.redis =Redisclient()
async def text_single_proxy(self,proxy):
'''
测试单个代理的可用性；
:param proxy: 单个代理；
:return:
'''
conn = aiohttp.TCPConnector(verify_ssl=False)
async with aiohttp.ClientSession(connector=conn) as session:
try:
if isinstance(proxy,bytes):
proxy =proxy.decode('utf-8')
real_proxy = 'http://'+proxy
async with session.get(TEXT_URL,proxy = real_proxy,timeout = 15) as response:
if response.status in VAILD_SATTUS_CODES:
self.redis.max(proxy)
print('')
print('代理可用',proxy)
else:
self.redis.decrease_proxy(proxy)
print('请求响应码不合法',proxy)
except (TimeoutError,AttributeError):
self.redis.decrease_proxy(proxy)
print('代理请求失败')
def run(self):

'''
调用主函数；
:return:
'''
print('测试器开始运行')
try:
proxies = self.redis.get_all_prpxy()
loop =asyncio.get_event_loop()
#批量测试
for i in range(0,len(proxies),BATCH_TEXT_SIZE):
test_proxies = proxies[i:i+BATCH_TEXT_SIZE]
tasks = [self.text_single_proxy(proxy) for proxy in test_proxies]
loop.run_until_complete(asyncio.wait(tasks))
time.sleep(5)
except Exception as e:
print('测试器发生错误',e.args)

验证代理ip是否有效

可以选择在requests请求一个特定的网页，参考相应参数判断是否（200）有效。也可以用telnetlib 库里的Telnet来判断是否成功。如果遇到不成功的ip，需要等待10秒。当然你也可以更改timeout的值，但如果你改的越小，你的ip池就越小。

import telnetlib
def verify(ip,port,type):
proxies = {}
try:
telnet = telnetlib.Telnet(ip,port=port,timeout=10)
except:
print('unconnected')
else:
print('connected successfully')
proxies['type'] = type
proxies['host'] = ip
proxies['port'] = port

为了帮助大家更轻松的学好Python，我给大家分享一套Python学习资料，希望对正在学习的你有所帮助！

获取方式：关注并私信小编 “ 学习 ”，即可免费获取！