小白學 Python 爬蟲（31）：自己構建一個簡單的代理池

人生苦短，我用 Python

如果我的文章對您有幫助，請關注支持下作者的公眾號：極客挖掘機，您的關注，是對小編堅持原創的最大鼓勵：）

前文傳送門：

小白學 Python 爬蟲（1）：開篇

小白學 Python 爬蟲（2）：前置準備（一）基本類庫的安裝

小白學 Python 爬蟲（3）：前置準備（二）Linux基礎入門

小白學 Python 爬蟲（4）：前置準備（三）Docker基礎入門

小白學 Python 爬蟲（5）：前置準備（四）數據庫基礎

小白學 Python 爬蟲（6）：前置準備（五）爬蟲框架的安裝

小白學 Python 爬蟲（7）：HTTP 基礎

小白學 Python 爬蟲（8）：網頁基礎

小白學 Python 爬蟲（9）：爬蟲基礎

小白學 Python 爬蟲（10）：Session 和 Cookies

小白學 Python 爬蟲（11）：urllib 基礎使用（一）

小白學 Python 爬蟲（12）：urllib 基礎使用（二）

小白學 Python 爬蟲（13）：urllib 基礎使用（三）

小白學 Python 爬蟲（14）：urllib 基礎使用（四）

小白學 Python 爬蟲（15）：urllib 基礎使用（五）

小白學 Python 爬蟲（16）：urllib 實戰之爬取妹子圖

小白學 Python 爬蟲（17）：Requests 基礎使用

小白學 Python 爬蟲（18）：Requests 進階操作

小白學 Python 爬蟲（19）：Xpath 基操

小白學 Python 爬蟲（20）：Xpath 進階

小白學 Python 爬蟲（21）：解析庫 Beautiful Soup（上）

小白學 Python 爬蟲（22）：解析庫 Beautiful Soup（下）

小白學 Python 爬蟲（23）：解析庫 pyquery 入門

小白學 Python 爬蟲（24）：2019 豆瓣電影排行

小白學 Python 爬蟲（25）：爬取股票信息

小白學 Python 爬蟲（26）：為啥買不起上海二手房你都買不起

小白學 Python 爬蟲（27）：自動化測試框架 Selenium 從入門到放棄（上）

小白學 Python 爬蟲（28）：自動化測試框架 Selenium 從入門到放棄（下）

小白學 Python 爬蟲（29）：Selenium 獲取某大型電商網站商品信息

小白學 Python 爬蟲（30）：代理基礎

引言

前面的代理如果有同學動手實踐過，就會發現一個問題，現在網上的免費代理簡直太坑啦！！！

經常一屏幕好多的代理試下來，沒有幾個能用的。

當然，免費的代理嘛，連通率低、延遲高是正常的，人家畢竟是免費的。

但是這件事兒有沒有解決方案呢？

這麼天資聰穎的小編肯定是想到了辦法了呀。

先來屢屢這件事兒，其實我們要的不是連通率高，而是我們在使用的時候，能每次都用到能用的代理。

這件事兒要麼我們每次在用的時候自己手動去試，要麼~~~

我們可以寫程序讓程序自己去尋找合適的代理嘛~~~

其實這一步就是把需要我們手動做的事情變成了程序自動去完成。

代理池

先想一下這個代理池最少需要有哪些功能：

• 自動獲取代理
• 定時清除不能用的代理

這兩個是我們的核心訴求，最少要有這兩個功能，不然這個代理池也沒有存在的價值了。

根據上面兩個功能，我們來拆解程序的模塊，小編這裡定義了三個模塊，獲取模塊（獲取代理）、存儲模塊（數據庫存儲代理）、檢測模塊（定時檢查代理的可用性）。

那麼它們三者的關係就是這樣的：

這裡的存儲模塊我們使用 Mysql ，這與存儲模塊為什麼選 Mysql ，因為 Mysql 有表結構，給各位同學展示起來比較清晰，如果需要用於生產環境的話，建議是用 Redis ，提高效率。

數據庫

首先還是貼一下數據庫的表結構，之前有同學留言問過小編表結構的事情，是小編偷懶沒有貼。

本次設計使用的還是單表模式，一張表走天下就是小編本人了。

至於字段的含義小編就不介紹了，後面的註釋已經寫得比較清楚了。

存儲模塊

對於存儲模塊來講，主要的功能是要將我們獲取到的代理保存起來，上面的 Mysql 數據庫是存儲模塊的一部分。

基於 OOP 的思想，我們本次寫一個類 MysqlClient 將所有對於 Mysql 的操作封裝起來，其他模塊需要和數據庫產生交互的時候只需要調用我們在 MysqlClient 中封裝好的方法即可。

示例代碼如下：

<code>MYSQL_HOST = 
'localhost'
MYSQL_PORT = 
3306
MYSQL_USER = 
'root'
MYSQL_PASSWORD = 
'password'
MYSQL_DB =
'test'
MYSQL_CHARSET = 
'utf8mb4'

import
 pymysql
import
 uuid

class
 
MysqlClient
(object)
:

    
def
 
__init__
(self, host=MYSQL_HOST, port=MYSQL_PORT, user=MYSQL_USER, password=MYSQL_PASSWORD, database=MYSQL_DB, charset=MYSQL_CHARSET)
:

        
"""
        初始化 mysql 連接
        :param host: mysql 地址
        :param port: mysql 端口
        :param user: mysql 用戶
        :param password: mysql 密碼
        :param database: mysql scheme
        :param charset: 使用的字符集
        """
        self.conn = pymysql.connect(
            host = host,
            port = port,
            user = user,
            password = password,
            database = database,
            charset = charset
        )

    
def
 
add_proxy
(self, proxy)
:

        
"""
        新增代理
        :param proxy: 代理字典
        :return:
        """ 

        sql = 
'INSERT INTO `proxy_pool` VALUES (%(id)s, %(scheme)s, %(ip)s, %(port)s, %(status)s, %(response_time)s, now(), null )'
        data = {
            
"id"
: str(uuid.uuid1()),
            
"scheme"
: proxy[
'scheme'
],
            
"ip"
: proxy[
'ip'
],
            
"port"
: proxy[
'port'
],
            
"status"
: proxy[
'status'
],
            
"response_time"
: proxy[
'response_time'
],
        }
        self.conn.cursor().execute(sql, data)
        self.conn.commit()

    
def
 
find_all
(self)
:

        
"""
        獲取所有可用代理
        :return:
        """
        sql = 
'SELECT * FROM proxy_pool WHERE status = "1" ORDER BY update_date ASC '
        cursor = self.conn.cursor()
        cursor.execute(sql)
        res = cursor.fetchall()
        cursor.close()
        self.conn.commit()
        
return
 res

    
def
 
update_proxy
(self, proxy)
:

        
"""
        更新代理信息
        :param proxy: 需要更新的代理
        :return:
        """ 

        sql = 
'UPDATE proxy_pool SET scheme = %(scheme)s, ip = %(ip)s, port = %(port)s, status = %(status)s, response_time = %(response_time)s, update_date = now()  WHERE id = %(id)s '
        data = {
            
"id"
: proxy[
'id'
],
            
"scheme"
: proxy[
'scheme'
],
            
"ip"
: proxy[
'ip'
],
            
"port"
: proxy[
'port'
],
            
"status"
: proxy[
'status'
],
            
"response_time"
: proxy[
'response_time'
],
        }
        self.conn.cursor().execute(sql, data)
        self.conn.commit()
/<code>

在這個類中，我們首先定義了一些常量，都是和數據庫連接有關的常量，如 MYSQL_HOST 數據庫地址、 MYSQL_PORT 數據庫端口、 MYSQL_USER 數據庫用戶名、 MYSQL_PASSWORD 數據庫密碼、 MYSQL_DB 數據庫的 scheme 、 MYSQL_CHARSET 字符集。

接下來定義了一個 MysqlClient 類，定義了一些方法用以執行數據庫的相關操作。

• init()： 初始化方法，在初始化 MysqlClient 這個類時，同時初始化了 Mysql 數據庫的鏈接信息，獲得了數據庫連接 connection 。
• add_proxy()：向數據庫中添加代理，並添加相關信息，包括代理響應延時和健康狀況。
• find_all()：獲取所有數據庫可用代理，並根據更新時間正序排布，主要用於後續代理檢查。
• update_proxy()：更新代理信息，主要用戶檢查模塊檢查完代理後更新代理信息，根據取出當前代理的主鍵 id 進行更新。

獲取模塊

獲取模塊相對比較簡單，主要功能就是從各個免費代理網站上將我們所需要的代理信息抓取下來。示例如下：

<code>
import
 requests
from
 pyquery 
import
 PyQuery
from
 MysqlClient 
import
 MysqlClient
from
 VerifyProxy 
import
 VerifyProxy

class
 
CrawlProxy
(object)
:


     
def
 
__init__
(self)
:

        self.mysql = MysqlClient()
        self.verify = VerifyProxy()

    
def
 
get_page
(self, url, charset)
:

        response = requests.get(url)
        response.encoding = charset
        
return
 response.text

    
def
 
crawl_ip3366
(self, page_num = 
3
)
:

        
"""
        獲取代理 ip3366
        :param page_num:
        :return:
        """
        start_url = 
'http://www.ip3366.net/?stype=1&page={}'
        urls = [start_url.format(page) 
for
 page 
in
 range(
1
, page_num + 
1
)]
        
for
 url 
in
 urls:
            print(
'crawl:'
, url)
            html = self.get_page(url, 
'gb2312'
)
            
if
 html:
                d = PyQuery(html)
                trs = d( 
'.table-bordered tbody tr'
).items()
                
for
 tr 
in
 trs:
                    scheme = tr.find(
'td:nth-child(4)'
).text().lower()
                    ip = tr.find(
'td:nth-child(1)'
).text()
                    port = tr.find(
'td:nth-child(2)'
).text()
                    verify_result = self.verify.verify_proxy(scheme, ip, port)

                    
if
 verify_result[
"status"
] == 
'1'
:
                        proxy = {
                            
"scheme"
: scheme,
                            
"ip"
: ip,
                            
"port"
: port,
                            
"status"
: verify_result[
"status"
],
                            
"response_time"
: verify_result[
"response_time"
],
                        }
                         
                        self.mysql.add_proxy(proxy)
                        print(
'代理'
, ip, 
'連通測試已通過，已保存 Mysql'
)
                    
else
:
                        print(
'代理'
, ip, 
'連通測試未通過'
)

if
 __name__ == 
'__main__'
:
    CrawlProxy().crawl_ip3366()
/<code>

小編這裡出於示例只演示了從 ip3366 上抓取免費代理，並且在抓取到代理後，調用檢查模塊的檢查方法對當前的代理進行連通性檢查，如果連通性測試未通過則不會寫入數據庫中。

檢查模塊

檢查模塊相對也會簡單一些，功能是從數據庫中取出所有可以用的代理，進行輪詢檢查，看看是不是有代理是連不通的，如果連不通則修改連通性標記位，將此代理標記為不可用。示例代碼如下：

<code>
import
 requests
from
 MysqlClient 
import
 MysqlClient

class
 
VerifyProxy
(object)
:

    
def
 
__init__
(self)
:

        self.mysql = MysqlClient()

    
def
 
verify_proxy
(self, scheme, ip, port)
:

        
"""
        使用百度測試代理的連通性，並返回響應時長（單位：ms）
        :param scheme:
        :param ip:
        :param port:
        :return:
        """
        proxies = {
            scheme: scheme + 
'://' 
 + ip + 
':'
 + port + 
'/'
        }
        response_time = 
0
        status = 
'0'
        
try
:
            response = requests.get(scheme + 
'://www.baidu.com/get'
, proxies=proxies)
            
if
 response.ok:
                response_time = round(response.elapsed.total_seconds() * 
1000
)
                status = 
'1'
            
else
:
                response_time = 
0
                status = 
'0'
        
except
:
            
pass
        
return
 {
"response_time"
 : response_time, 
"status"
 : status}

    
def
 
verify_all
(self)
:

        
"""
        驗證住方法，從數據庫中獲取所有代理進行驗證
        :return:
        """
        results = self.mysql.find_all()
        
for
 result 
in 
 results:
            res = self.verify_proxy(result[
1
], result[
2
], result[
3
])
            proxy = {
                
"id"
: result[
0
],
                
"scheme"
: result[
1
],
                
"ip"
: result[
2
],
                
"port"
: result[
3
],
                
"status"
: res[
"status"
],
                
"response_time"
: res[
"response_time"
],
            }
            self.mysql.update_proxy(proxy)
            print(
'代理驗證成功'
)

if
 __name__ == 
'__main__'
:
    VerifyProxy().verify_all()
/<code>

小編這裡使用的是度娘進行連通性測試，如果各位同學有特殊的需要，可以使用特定的網站進行連通性測試。

小結

本篇的內容到這裡就結束了，不過有一點要說明，本篇的示例內容只能作為 DEMO 來進行測試使用，對於一個連接池來講，還有很多不完善的地方。

例如檢測模塊應該是定時啟動，自行檢測，現在是靠人手動啟動，不過這個可以使用各種系統上的定時任務來解決。

還有，現在要獲取連接信息只能自己打開數據庫從中 Copy ，這裡其實還可以加一個 API 模塊，寫成一個接口，供其他有需要使用代理的系統進行調用。

獲取模塊現在小編也只是簡單的有一個網站寫一個方法，其實可以使用 Python 高級用法，獲取到類中所有的方法名，然後調用所需要的方法。

總之，這個 DEMO 非常不完善，等小編下次有空的時候完善下，到時候還可以再來一個推送。

示例代碼

本系列的所有代碼小編都會放在代碼管理倉庫 Github 和 Gitee 上，方便大家取用。

示例代碼-Github

示例代碼-Gitee