Python构建代理IP池：接口配置与多进程调度详解

纸上星河

发布时间：2026-10-13| 4人浏览过

原创

本文详细介绍如何使用Python搭建代理IP池，重点讲解通过Flask提供Web API接口以安全获取代理IP的方法，以及利用多进程模块并行运行获取、检测和API服务模块的调度策略。包含setting.py配置说明及scheduler.py核心代码解析，帮助开发者实现高效稳定的代理池系统。

代理IP池接口模块设计

接口模块旨在通过API提供对外服务。虽然可以直接连接数据库获取数据，但这需要暴露数据库连接信息，存在安全隐患且配置繁琐。因此，提供一个Web API接口是更安全便捷的方式，用户只需访问指定接口即可获取可用的代理IP。

完整代码地址：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 = 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

setting.py中的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应用对象，并定义了三个主要接口：首页、随机代理获取页以及代理池总量查询页。

用户只需访问对应的API接口，即可直接获取到可用的代理IP。

多进程调度模块实现

调度模块

调度模块负责调用之前定义的获取、存储和检测三个子模块，并通过多进程机制将它们并行运行，以提高系统效率。

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()方法启动。这样，三个进程即可并行执行且互不干扰。