Redis数据库模块和代理源获取模块都已经完成，我们的代理池中已经有了通过Getter模块获取到代理源了，但是这些代理是否有效，可用性如何呢？接下来讲解代理池的测试模块。

设计思路

代理池中代理的数量非常多（例如本文设置为10000），对这些代理进行测试那自然是需要并发的去处理了，所以选择使用aiohttp进行代理请求测试。

开发

在MyProxyPool项目中新建tester.py：

from redisdb import RedisCli
import aiohttp
from aiohttp import ClientError, ClientConnectionError, ServerTimeoutError
import asyncio
import time
from asyncio import TimeoutError

# 测试认为有效的状态码
VALID_STATUS_CODES = [200]
# 测试网站
TEST_URL = 'http://www.baidu.com'

# 每次批量测试的数量
BATCH_TEST_SIZE = 100


class Tester:
    def __init__(self):
        self.redis = RedisCli()

    async def test_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
                print('正在测试代理', real_proxy)
                async with session.get(TEST_URL, proxy=real_proxy, timeout=15) as response:
                    if response.status in VALID_STATUS_CODES:
                        # 如果代理可用，分数置为最高
                        self.redis.max(proxy)
                        print('代理%s可用' % proxy)
                        return True
                    else:
                        # 如果代理不可用，分数减一
                        self.redis.decrease(proxy)
                        print('请求响应码不合理，代理不可用', proxy)
                        return False
            except (ClientError, ClientConnectionError, ServerTimeoutError, AttributeError, TimeoutError):
                self.redis.decrease(proxy)
                print('代理请求失败,不可用', proxy)
                return False

    def run(self):
        try:
            proxies = self.redis.all()
            loop = asyncio.get_event_loop()
            # 批量测试
            for i in range(0, len(proxies), BATCH_TEST_SIZE):
                test_proxies = proxies[i: i + BATCH_TEST_SIZE]
                tasks = [self.test_single_proxy(proxy) for proxy in test_proxies]
                loop.run_until_complete(asyncio.wait(tasks))
                # 睡眠五秒钟
                time.sleep(5)
        except Exception as e:
            print('测试错误', e)


if __name__ == '__main__':
    Tester().run()

测试网站我这里使用的是百度首页，作为一个通用的测试。如果你这个代理池是为了爬取某一网站，那你可以选择这个网站中的链接作为测试的url，这样针对性的检测，可用性更高。
这里我每五秒钟进行一次测试，每次测试BATCH_TEST_SIZE = 100个代理。

本节代码戳我