Python搭建 * 池实现接口设置与整体调度
作者:Steven·简谈 时间:2023-05-25 11:52:03
接口模块需要用 API 来提供对外服务的接口,当然也可以直接连数据库来取,但是这样就需要知道数据库的连接信息,不太安全,而且需要配置连接,所以一个比较安全和方便的方式就是提供一个 Web API 接口,通过访问接口即可拿到可用代理
代码地址:https://github.com/Stevengz/Proxy_pool
另外三篇:
Python搭建 * 池(一)- 获取 IP
Python搭建 * 池(二)- 存储 IP
Python搭建 * 池(三)- 检测 IP
添加设置
添加模块开关变量
setting.py
# 数据库地址
HOST = '127.0.0.1'
# MySql端口
MYSQL_PORT = 3306
# MySQl用户名、密码
MYSQL_USERNAME = '***'
MYSQL_PASSWORD = '***'
# 数据库名
SQL_NAME = 'test'
# 代理等级
MAX_SCORE = 30
MIN_SCORE = 0
INITIAL_SCORE = 10
VALID_STATUS_CODES = [200, 302]
# 代理池数量界限
POOL_UPPER_THRESHOLD = 1000
# 检查周期
TESTER_CYCLE = 20
# 获取周期
GETTER_CYCLE = 300
# 测试API,建议抓哪个网站测哪个
TEST_URL = 'http://www.baidu.com'
# API配置
API_HOST = '0.0.0.0'
API_PORT = 5555
# 开关
TESTER_ENABLED = True
GETTER_ENABLED = True
API_ENABLED = True
# 最大批测试量
BATCH_TEST_SIZE = 10
TESTER_ENABLED、GETTER_ENABLED、API_ENABLED 都是布尔类型,True 或者 False。标明了测试模块、获取模块、接口模块的开关,如果为 True,则代表模块开启
定义接口
使用框架:Flask
api.py
from flask import Flask, g
from db import MySqlClient
__all__ = ['app']
app = Flask(__name__)
def get_conn():
if not hasattr(g, 'mysql'):
g.mysql = MySqlClient()
return g.mysql
@app.route('/')
def index():
return '<h2>Welcome 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 time
from multiprocessing import Process
from api import app
from getter import Getter
from tester import Tester
from db import MySqlClient
from 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() 方法运行,这样三个进程就可以并行执行,互不干扰
来源:https://blog.csdn.net/weixin_44613063/article/details/102577490
标签:Python,接口设置,整体调度
0
投稿
猜你喜欢
JavaScript经典效果集锦
2013-08-13 09:29:34
JS中Map和ForEach的区别
2024-04-29 13:19:47
pytorch显存一直变大的解决方案
2021-03-03 00:03:09
Python scikit-learn 做线性回归的示例代码
2022-05-03 11:00:54
详解python的变量
2021-01-26 20:44:20
postman批量执行接口测试的图文步骤
2023-03-14 00:22:13
学生信息管理系统python版
2023-10-17 07:32:34
解析Python的缩进规则的使用
2022-11-27 10:14:16
js禁止Backspace键使浏览器后退的实现方法
2024-04-17 09:54:05
python中利用队列asyncio.Queue进行通讯详解
2023-02-13 03:52:58
解决pandas read_csv 读取中文列标题文件报错的问题
2023-04-16 06:54:37
python时间与Unix时间戳相互转换方法详解
2023-07-11 02:09:12
k8s容器互联-flannel host-gw原理篇
2024-04-27 15:40:07
MySQL查询冗余索引和未使用过的索引操作
2024-01-22 22:04:23
使用 XML 文件记录操作日志
2008-09-05 17:13:00
正则表达式学习笔记
2008-04-15 07:44:00
python opencv实现图片旋转矩形分割
2022-01-25 06:23:24
Python命令行定时任务自动化工作流程
2023-01-01 20:52:51
python3.6使用urllib完成下载的实例
2023-08-03 21:21:13
XML+ JS创建树形菜单
2013-08-22 08:30:17