快捷導(dǎo)航

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

更新時(shí)間：2019年10月27日 15:33:07 作者：Steven·簡(jiǎn)談

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

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

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

另外三篇：
Python搭建代理IP池（一）- 獲取 IP
Python搭建代理IP池（二）- 存儲(chǔ) IP
Python搭建代理IP池（三）- 檢測(cè) IP

添加設(shè)置

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

setting.py

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

# 代理等級(jí)
MAX_SCORE = 30
MIN_SCORE = 0
INITIAL_SCORE = 10

VALID_STATUS_CODES = [200, 302]

# 代理池?cái)?shù)量界限
POOL_UPPER_THRESHOLD = 1000

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

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

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

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

# 最大批測(cè)試量
BATCH_TEST_SIZE = 10

TESTER_ENABLED、GETTER_ENABLED、API_ENABLED 都是布爾類(lèi)型，True 或者 False。標(biāo)明了測(cè)試模塊、獲取模塊、接口模塊的開(kāi)關(guān)，如果為 True，則代表模塊開(kāi)啟

定義接口

使用框架：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>'

# 隨機(jī)代理
@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()

聲明了一個(gè) Flask 對(duì)象，定義了三個(gè)接口，分別是首頁(yè)、隨機(jī)代理頁(yè)、獲取數(shù)量頁(yè)。

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

調(diào)度模塊

調(diào)用定義的獲取、存儲(chǔ)、檢測(cè)三個(gè)模塊，將這三個(gè)模塊通過(guò)多進(jìn)程的形式運(yùn)行起來(lái)

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():

 # 定時(shí)測(cè)試代理
 def schedule_tester(self, cycle=TESTER_CYCLE):
  tester = Tester()
  while True:
   print('測(cè)試器開(kāi)始運(yùn)行')
   tester.run()
   time.sleep(cycle)
 
 # 定時(shí)獲取代理
 def schedule_getter(self, cycle=GETTER_CYCLE):
  getter = Getter()
  while True:
   print('開(kāi)始抓取代理')
   getter.run()
   time.sleep(cycle)
 
 # 開(kāi)啟API
 def schedule_api(self):
  app.run(API_HOST, API_PORT)
 
 def run(self):
  print('代理池開(kāi)始運(yùn)行')
  
  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()

啟動(dòng)入口是 run() 方法，分別判斷了三個(gè)模塊的開(kāi)關(guān)，如果開(kāi)啟的話，就新建一個(gè) Process 進(jìn)程，設(shè)置好啟動(dòng)目標(biāo)，然后調(diào)用 start() 方法運(yùn)行，這樣三個(gè)進(jìn)程就可以并行執(zhí)行，互不干擾