python爬虫014-IP代理池的维护-总结调度

2019-04-10 本文已影响23人 DKider

前几天开始的IP代理池维护今天终于要见成果了。

我们一共写了4个模块：

获取模块----------->存储模块<--------------->检测模块
                      |
                      |
                      |
                      |
                      V
                   接口模块

一共四个模块：获取模块（crawler.py）、存储模块（saver.py）、检测模块（tester.py）、接口模块（api.py），其中获取模块我用了一个getter.py来调度爬虫，将数据存到数据库。

getter.py

from saver import RedisClient
from crawler import Crawler


POOL_UPPER_THRESHOLD = 10000


class Getter():
    def __init__(self):
        self.redis = RedisClient()
        self.crawler = Crawler()

    def is_over_threshold(self):
        """判断是否达到了代理池限制"""
        if self.redis.count() > POOL_UPPER_THRESHOLD:
            return True
        else:
            return False

    def run(self):
        """调度获取器"""
        print('获取器开始执行')
        if not self.is_over_threshold():
            for callback_lable in range(self.crawler.__CrawlFuncCount__):
                callback = self.crawler.__CrawlFunc__[callback_lable]
                proxies = self.crawler.get_proxies(callback)
                for proxy in proxies:
                    self.redis.add(proxy)

有了这个getter之后，我们就可以写run.py来串联所有的模块，让程序运行起来。

import time
from api import app
from getter import Getter
from tester import Tester
from multiprocessing import Process
from multiprocessing import Pool

TESTER_CYCLE = 30
GETTER_CYCLE = 30
TESTER_ENABLE = True
GETTER_ENABLE = True
API_ENABLE = True


class Scheduler():
    def schedule_tester(self, cycle=TESTER_CYCLE):
        """定时检测代理"""
        tester = Tester()
        while True:
            print('测试器开始运行')
            tester.run()
            time.sleep(cycle)

    def schedule_getter(self, cycle=GETTER_CYCLE):
        """定时获取代理"""
        getter = Getter()
        while True:
            print('获取器开始运行')
            getter.run()
            time.sleep(cycle)

    def schedule_api(self):
        """开启api"""
        app.run()

    def run(self):
        """运行代理池"""

        # 开放api
        if API_ENABLE:
            api_process = Process(target=self.schedule_api)
            api_process.start()

        # 开始循环测试
        if TESTER_ENABLE:
            tester_process = Process(target=self.schedule_tester)
            tester_process.start()

        # 开始循环获取新的代理
        if GETTER_ENABLE:
            getter_process = Process(target=self.schedule_getter)
            getter_process.start()


if __name__ == '__main__':
    p = Scheduler()
    p.run()

这里我们用了多进程，让getter、tester、和api一直运行下去，getter定时调用爬虫爬取各大免费ip代理网站，tester定时检测代理的可用性。getter将的得到代理存到数据库，tester从数据库取代理检测可用性，判定分数。api一直开启，在127.0.0.1:5000端口提供随机代理。

import time
from api import app
from getter import Getter
from tester import Tester
from multiprocessing import Process

TESTER_CYCLE = 30
GETTER_CYCLE = 30
TESTER_ENABLE = True
GETTER_ENABLE = True
API_ENABLE = True


class Scheduler():
    def schedule_tester(self, cycle=TESTER_CYCLE):
        """定时检测代理"""
        tester = Tester()
        while True:
            print('测试器开始运行')
            tester.run()
            time.sleep(cycle)

    def schedule_getter(self, cycle=GETTER_CYCLE):
        """定时获取代理"""
        getter = Getter()
        while True:
            print('获取器开始运行')
            getter.run()
            time.sleep(cycle)

    def schedule_api(self):
        """开启api"""
        app.run()

    def run(self):
        """运行代理池"""

        # 开放api
        if API_ENABLE:
            api_process = Process(target=self.schedule_api)
            api_process.start()

        # 开始循环测试
        if TESTER_ENABLE:
            tester_process = Process(target=self.schedule_tester)
            tester_process.start()

        # 开始循环获取新的代理
        if GETTER_ENABLE:
            getter_process = Process(target=self.schedule_getter)
            getter_process.start()


if __name__ == '__main__':
    p = Scheduler()
    p.run()

注意！运行前一定要开启数据库。
这个程序调度三个模块，让他们同时运行，我运行一段时间后，效果如下：