python爬虫012-IP代理池的维护-（3）检测模块

前天写了存储模块，昨天写了获取模块，今天写了检测模块。

我们通过获取模块得到了各大代理网站的ip代理地址和端口，用存储模块将得到的代理存储到redis数据库中，并用检测模块从数据库中读取IP代理进行检测，给代理打分。

检测模块的原理是这样的，从数据库中调用代理，并用这个代理去访问设定的网站，查看是否可用，以此更改代理分数。

这里用的是aiohttp异步访问，因为有的代理访问速度慢，所以需要等待一段时间，在程序等待返回结果的时候我们可以让程序去执行别的任务，加快检测的效率。

异步io我没学明白。就跟着书做了。之后的结果我们还要用一个调度模块来实现。

代码：

import aiohttp
import asyncio
import time
from saver import RedisClient
from aiohttp.client import ClientConnectorError
from aiohttp.client import ClientError


VALID_STATUS_CODES = [200]
TEST_URL = 'http://www.baidu.com'  # 应该改成要爬取的网站
BATCH_TEST_SIZE = 100


class Tester(object):
    """测试代理的可用性"""
    def __init__(self):
        """连接数据库"""
        self.redis = Redis = RedisClient()

    async def test_single_proxy(self, proxy):
        """测试单个代理

        :param proxy: 代理
        :return: None"""
        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
                print('正在测试', proxy)
                async with session.get(TEST_URL, proxy=real_proxy,
                                       timeout=15) as respone:
                    if respone.status in VALID_STATUS_CODES:
                        self.redis.max(proxy)
                        print('代理可用', proxy)
                    else:
                        self.redis.decrease(proxy)
                        print('请求响应不合法', proxy)
            except (TimeoutError, AttributeError, ClientError,
                    ClientConnectorError):
                self.redis.decrease(proxy)
                print('代理请求失败')

    def run(self):
        """测试主函数"""
        print('测试器开始运行')
        try:
            proxies = self.redis.all()
            loop = asyncio.get_event_loop()
            # 批量测试
            for i in range(0, len(proxies), BATCH_TEST_SIZE):
                test_proxies = proxies[i:BATCH_TEST_SIZE + i]
                task = [self.test_single_proxy(proxy) for proxy in test_proxies]
                loop.run_until_complete(asyncio.wait(task))
                time.sleep(5)
        except Exception as e:
            print('测试器发生故障', e.args)

异步IO等我买了《流畅的python》之后再研究吧，手边没有好的学习资源。

我不太喜欢看视频学习，更喜欢看书自己学习。不知道为什么，身边的人都更喜欢看视频学习。
我看视频学习，看的时候非常清楚，等看完了，也就完了，什么也没记住。。。

可能我比较传统吧。。但是不是啊。。。

现在一边看数据结构和算法，一边看爬虫，有点忙不过来。。。