接口模块需要用 API 来提供对外服务的接口，当然也可以直接连数据库来取，但是这样就需要知道数据库的连接信息，不太安全，而且需要配置连接，所以一个比较安全和方便的方式就是提供一个 Web API 接口，通过访问接口即可拿到可用代理

代码地址：https://github.com/Stevengz/Proxy_pool

另外三篇：
Python搭建代理IP池（一）- 获取 IP
Python搭建代理IP池（二）- 存储 IP
Python搭建代理IP池（三）- 检测 IP

添加设置

添加模块开关变量

setting.py

# 数据库地址HOST = '127.0.0.1'# MySql端口MYSQL_PORT = 3306# MySQl用户名、密码MYSQL_USERNAME = '***'MYSQL_PASSWORD = '***'# 数据库名SQL_NAME = 'test'# 代理等级MAX_SCORE = 30MIN_SCORE = 0INITIAL_SCORE = 10VALID_STATUS_CODES = [200, 302]# 代理池数量界限POOL_UPPER_THRESHOLD = 1000# 检查周期TESTER_CYCLE = 20# 获取周期GETTER_CYCLE = 300# 测试API，建议抓哪个网站测哪个TEST_URL = 'http://e to Proxy Pool System</h2>'# 随机代理@app.route('/random')def get_proxy(): conn = get_conn() return conn.random()# 代理池总量@app.route('/count')def get_counts(): conn = get_conn() return str(conn.count())if __name__ == '__main__': app.run()

声明了一个 Flask 对象，定义了三个接口，分别是首页、随机代理页、获取数量页。

只需要访问对应的接口即可获取到可用代理：

调度模块

调用定义的获取、存储、检测三个模块，将这三个模块通过多进程的形式运行起来

scheduler.py

import timefrom multiprocessing import Processfrom api import appfrom getter import Getterfrom tester import Testerfrom db import MySqlClientfrom setting import *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) # 开启API def schedule_api(self): app.run(API_HOST, API_PORT) def run(self): print('代理池开始运行') if TESTER_ENABLED: tester_process = Process(target=self.schedule_tester) tester_process.start() if GETTER_ENABLED: getter_process = Process(target=self.schedule_getter) getter_process.start() if API_ENABLED: api_process = Process(target=self.schedule_api) api_process.start()if __name__ == "__main__": scheduler = Scheduler() scheduler.run()

启动入口是 run() 方法，分别判断了三个模块的开关，如果开启的话，就新建一个 Process 进程，设置好启动目标，然后调用 start() 方法运行，这样三个进程就可以并行执行，互不干扰

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。