快捷導(dǎo)航

Python搭建代理IP池實現(xiàn)接口設(shè)置與整體調(diào)度

更新時間：2019年10月27日 15:33:07 作者：Steven·簡談

這篇文章主要介紹了Python搭建代理IP池實現(xiàn)接口設(shè)置與整體調(diào)度，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

接口模塊需要用 API 來提供對外服務(wù)的接口，當(dāng)然也可以直接連數(shù)據(jù)庫來取，但是這樣就需要知道數(shù)據(jù)庫的連接信息，不太安全，而且需要配置連接，所以一個比較安全和方便的方式就是提供一個 Web API 接口，通過訪問接口即可拿到可用代理

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

另外三篇：
Python搭建代理IP池（一）- 獲取 IP
Python搭建代理IP池（二）- 存儲 IP
Python搭建代理IP池（三）- 檢測 IP

添加設(shè)置

添加模塊開關(guān)變量

setting.py

# 數(shù)據(jù)庫地址
HOST = '127.0.0.1'
# MySql端口
MYSQL_PORT = 3306
# MySQl用戶名、密碼
MYSQL_USERNAME = '***'
MYSQL_PASSWORD = '***'
# 數(shù)據(jù)庫名
SQL_NAME = 'test'

# 代理等級
MAX_SCORE = 30
MIN_SCORE = 0
INITIAL_SCORE = 10

VALID_STATUS_CODES = [200, 302]

# 代理池數(shù)量界限
POOL_UPPER_THRESHOLD = 1000

# 檢查周期
TESTER_CYCLE = 20
# 獲取周期
GETTER_CYCLE = 300

# 測試API，建議抓哪個網(wǎng)站測哪個
TEST_URL = 'http://www.baidu.com'

# API配置
API_HOST = '0.0.0.0'
API_PORT = 5555

# 開關(guān)
TESTER_ENABLED = True
GETTER_ENABLED = True
API_ENABLED = True

# 最大批測試量
BATCH_TEST_SIZE = 10

TESTER_ENABLED、GETTER_ENABLED、API_ENABLED 都是布爾類型，True 或者 False。標(biāo)明了測試模塊、獲取模塊、接口模塊的開關(guān)，如果為 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 對象，定義了三個接口，分別是首頁、隨機代理頁、獲取數(shù)量頁。

只需要訪問對應(yīng)的接口即可獲取到可用代理：

調(diào)度模塊

調(diào)用定義的獲取、存儲、檢測三個模塊，將這三個模塊通過多進程的形式運行起來

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() 方法，分別判斷了三個模塊的開關(guān)，如果開啟的話，就新建一個 Process 進程，設(shè)置好啟動目標(biāo)，然后調(diào)用 start() 方法運行，這樣三個進程就可以并行執(zhí)行，互不干擾