手把手教你搭建一個基于Java的分布式爬蟲系統(tǒng)

原創(chuàng)
作者：葉泳豪 2018-05-09 09:44:51

開發(fā)

后端

開發(fā)工具

分布式 在不用爬蟲框架的情況下，我經(jīng)過多方學(xué)習(xí)，嘗試實(shí)現(xiàn)了一個分布式爬蟲系統(tǒng)，并且可以將數(shù)據(jù)保存到不同地方，類似 MySQL、HBase 等。

專注于為中小企業(yè)提供成都網(wǎng)站制作、成都網(wǎng)站設(shè)計服務(wù),電腦端+手機(jī)端+微信端的三站合一,更高效的管理,為中小企業(yè)南溪免費(fèi)做網(wǎng)站提供優(yōu)質(zhì)的服務(wù)。我們立足成都，凝聚了一批互聯(lián)網(wǎng)行業(yè)人才，有力地推動了上1000家企業(yè)的穩(wěn)健成長，幫助中小企業(yè)通過網(wǎng)站建設(shè)實(shí)現(xiàn)規(guī)模擴(kuò)充和轉(zhuǎn)變。

【51CTO.com原創(chuàng)稿件】在不用爬蟲框架的情況下，我經(jīng)過多方學(xué)習(xí)，嘗試實(shí)現(xiàn)了一個分布式爬蟲系統(tǒng)，并且可以將數(shù)據(jù)保存到不同地方，類似 MySQL、HBase 等。

因?yàn)榇讼到y(tǒng)基于面向接口的編碼思想來開發(fā)，所以具有一定的擴(kuò)展性，有興趣的朋友直接看一下代碼，就能理解其設(shè)計思想。

雖然代碼目前來說很多地方還是比較緊耦合，但只要花些時間和精力，很多都是可抽取出來并且可配置化的。

因?yàn)闀r間的關(guān)系，我只寫了京東和蘇寧易購兩個網(wǎng)站的爬蟲，但是完全可以實(shí)現(xiàn)不同網(wǎng)站爬蟲的隨機(jī)調(diào)度，基于其代碼結(jié)構(gòu)，再寫國美、天貓等的商品爬取，難度不大，但是估計需要花些時間和精力。

因?yàn)樵诮馕鼍W(wǎng)頁的數(shù)據(jù)時，比如我在爬取蘇寧易購商品的價格時，價格是異步獲取的，并且其 API 是一長串的數(shù)字組合，我花了幾個小時的時間才發(fā)現(xiàn)其規(guī)律，當(dāng)然也承認(rèn)，我的經(jīng)驗(yàn)不足。

這個系統(tǒng)的設(shè)計，除了基本的數(shù)據(jù)爬取以外，更關(guān)注以下幾個方面的問題：

• 如何實(shí)現(xiàn)分布式？同一個程序打包后分發(fā)到不同的節(jié)點(diǎn)運(yùn)行時，不影響整體的數(shù)據(jù)爬取。
• 如何實(shí)現(xiàn) URL 隨機(jī)循環(huán)調(diào)度？核心是針對不同的***域名做隨機(jī)。
• 如何定時向 URL 倉庫中添加種子 URL？達(dá)到不讓爬蟲系統(tǒng)停下來的目的。
• 如何實(shí)現(xiàn)對爬蟲節(jié)點(diǎn)程序的監(jiān)控，并能夠發(fā)郵件報警？
• 如何實(shí)現(xiàn)一個隨機(jī) IP 代理庫？目的跟第 2 點(diǎn)有點(diǎn)類似，都是為了反反爬蟲。

下面會針對這個系統(tǒng)來做一個整體的基本介紹，我在代碼中都有非常詳細(xì)的注釋，有興趣的朋友可以參考一下代碼，***我會給出一些我爬蟲時的數(shù)據(jù)分析。

另外需要注意的是，這個爬蟲系統(tǒng)是基于 Java 實(shí)現(xiàn)的，但是語言本身仍然不是最重要的，有興趣的朋友可以嘗試用 Python 實(shí)現(xiàn)。

分布式爬蟲系統(tǒng)架構(gòu)

整體系統(tǒng)架構(gòu)如下：

從上面的架構(gòu)可以看出，整個系統(tǒng)主要分為三個部分：

• 爬蟲系統(tǒng)
• URL 調(diào)度系統(tǒng)
• 監(jiān)控報警系統(tǒng)

爬蟲系統(tǒng)是用來爬取數(shù)據(jù)的，因?yàn)橄到y(tǒng)設(shè)計為分布式，因此，爬蟲程序本身可以運(yùn)行在不同的服務(wù)器節(jié)點(diǎn)上。

URL 調(diào)度系統(tǒng)核心在于 URL 倉庫，所謂的 URL 倉庫其實(shí)就是用 Redis 保存了需要爬取的 URL 列表，并且在我們的 URL 調(diào)度器中根據(jù)一定的策略來消費(fèi)其中的 URL。從這個角度考慮，URL 倉庫其實(shí)也是一個 URL 隊(duì)列。

監(jiān)控報警系統(tǒng)主要是對爬蟲節(jié)點(diǎn)進(jìn)行監(jiān)控，雖然并行執(zhí)行的爬蟲節(jié)點(diǎn)中的某一個掛掉了對整體數(shù)據(jù)爬取本身沒有影響（只是降低了爬蟲的速度），但是我們還是希望能夠主動接收到節(jié)點(diǎn)掛掉的通知，而不是被動地發(fā)現(xiàn)。

下面將針對以上三個方面并結(jié)合部分代碼片段來對整個系統(tǒng)的設(shè)計思路做一些基本的介紹。

爬蟲系統(tǒng)

爬蟲系統(tǒng)是一個獨(dú)立運(yùn)行的進(jìn)程，我們把我們的爬蟲系統(tǒng)打包成 jar 包，然后分發(fā)到不同的節(jié)點(diǎn)上執(zhí)行，這樣并行爬取數(shù)據(jù)可以提高爬蟲的效率。（說明：ZooKeeper 監(jiān)控屬于監(jiān)控報警系統(tǒng)，URL 調(diào)度器屬于 URL 調(diào)度系統(tǒng)）

隨機(jī) IP 代理器

加入隨機(jī) IP 代理主要是為了反反爬蟲，因此如果有一個 IP 代理庫，并且可以在構(gòu)建 http 客戶端時隨機(jī)地使用不同的代理，那么對我們進(jìn)行反反爬蟲會有很大的幫助。

在系統(tǒng)中使用 IP 代理庫，需要先在文本文件中添加可用的代理地址信息：

  
 
 
 

   
  
  
  
# IPProxyRepository.txt 
   
  
  
  
58.60.255.104:8118 
   
  
  
  
219.135.164.245:3128 
   
  
  
  
27.44.171.27:9999 
   
  
  
  
219.135.164.245:3128 
   
  
  
  
58.60.255.104:8118 
   
  
  
  
58.252.6.165:9000 
   
  
  
  
...... 
  
 
 
 

需要注意的是，上面的代理 IP 是我在西刺代理上拿到的一些代理 IP，不一定可用，建議是自己花錢購買一批代理 IP，這樣可以節(jié)省很多時間和精力去尋找代理 IP。

然后在構(gòu)建 http 客戶端的工具類中，當(dāng)***次使用工具類時，會把這些代理 IP 加載進(jìn)內(nèi)存中，加載到 Java 的一個 HashMap：

  
 
 
 

   
  
  
  
// IP地址代理庫Map 
   
  
  
  
private static Map IPProxyRepository = new HashMap<>(); 
   
  
  
  
private static String[] keysArray = null;   // keysArray是為了方便生成隨機(jī)的代理對象 
   
  
  
  
 
   
  
  
  
/** 
   
  
  
  
     * 初次使用時使用靜態(tài)代碼塊將IP代理庫加載進(jìn)set中 
   
  
  
  
     */ 
   
  
  
  
static { 
   
  
  
  
    InputStream in = HttpUtil.class.getClassLoader().getResourceAsStream("IPProxyRepository.txt");  // 加載包含代理IP的文本 
   
  
  
  
    // 構(gòu)建緩沖流對象 
   
  
  
  
    InputStreamReader isr = new InputStreamReader(in); 
   
  
  
  
    BufferedReader bfr = new BufferedReader(isr); 
   
  
  
  
    String line = null; 
   
  
  
  
    try { 
   
  
  
  
        // 循環(huán)讀每一行，添加進(jìn)map中 
   
  
  
  
        while ((line = bfr.readLine()) != null) { 
   
  
  
  
            String[] split = line.split(":");   // 以:作為分隔符，即文本中的數(shù)據(jù)格式應(yīng)為192.168.1.1:4893 
   
  
  
  
            String host = split[0]; 
   
  
  
  
            int port = Integer.valueOf(split[1]); 
   
  
  
  
            IPProxyRepository.put(host, port); 
   
  
  
  
        } 
   
  
  
  
        Set keys = IPProxyRepository.keySet(); 
   
  
  
  
        keysArray = keys.toArray(new String[keys.size()]);  // keysArray是為了方便生成隨機(jī)的代理對象 
   
  
  
  
    } catch (IOException e) { 
   
  
  
  
        e.printStackTrace(); 
   
  
  
  
    } 
   
  
  
  
 
   
  
  
  
} 
  
 
 
 

之后，在每次構(gòu)建 http 客戶端時，都會先到 map 中看是否有代理 IP，有則使用，沒有則不使用代理：

  
 
 
 

   
  
  
  
CloseableHttpClient httpClient = null; 
   
  
  
  
HttpHost proxy = null; 
   
  
  
  
if (IPProxyRepository.size() > 0) {  // 如果ip代理地址庫不為空，則設(shè)置代理 
   
  
  
  
    proxy = getRandomProxy(); 
   
  
  
  
    httpClient = HttpClients.custom().setProxy(proxy).build();  // 創(chuàng)建httpclient對象 
   
  
  
  
} else { 
   
  
  
  
    httpClient = HttpClients.custom().build();  // 創(chuàng)建httpclient對象 
   
  
  
  
} 
   
  
  
  
HttpGet request = new HttpGet(url); // 構(gòu)建htttp get請求 
   
  
  
  
...... 
  
 
 
 

隨機(jī)代理對象則通過下面的方法生成：

  
 
 
 

   
  
  
  
/** 
   
  
  
  
     * 隨機(jī)返回一個代理對象 
   
  
  
  
     * 
   
  
  
  
     * @return 
   
  
  
  
     */ 
   
  
  
  
public static HttpHost getRandomProxy() { 
   
  
  
  
    // 隨機(jī)獲取host:port，并構(gòu)建代理對象 
   
  
  
  
    Random random = new Random(); 
   
  
  
  
    String host = keysArray[random.nextInt(keysArray.length)]; 
   
  
  
  
    int port = IPProxyRepository.get(host); 
   
  
  
  
    HttpHost proxy = new HttpHost(host, port);  // 設(shè)置http代理 
   
  
  
  
    return proxy; 
   
  
  
  
} 
  
 
 
 

這樣，通過上面的設(shè)計，基本就實(shí)現(xiàn)了隨機(jī) IP 代理器的功能，當(dāng)然，其中還有很多可以完善的地方。

比如，當(dāng)使用這個 IP 代理而請求失敗時，是否可以把這一情況記錄下來；當(dāng)超過一定次數(shù)時，再將其從代理庫中刪除，同時生成日志供開發(fā)人員或運(yùn)維人員參考，這是完全可以實(shí)現(xiàn)的，不過我就不做這一步功能了。

網(wǎng)頁下載器

網(wǎng)頁下載器就是用來下載網(wǎng)頁中的數(shù)據(jù)，主要基于下面的接口開發(fā)：

  
 
 
 

   
  
  
  
/** 
   
  
  
  
 * 網(wǎng)頁數(shù)據(jù)下載 
   
  
  
  
 */ 
   
  
  
  
public interface IDownload { 
   
  
  
  
    /** 
   
  
  
  
     * 下載給定url的網(wǎng)頁數(shù)據(jù) 
   
  
  
  
     * @param url 
   
  
  
  
     * @return 
   
  
  
  
     */ 
   
  
  
  
    public Page download(String url); 
   
  
  
  
} 
  
 
 
 

基于此，在系統(tǒng)中只實(shí)現(xiàn)了一個 http get 的下載器，但是也可以完成我們所需要的功能了：

  
 
 
 

   
  
  
  
/** 
   
  
  
  
 * 數(shù)據(jù)下載實(shí)現(xiàn)類 
   
  
  
  
 */ 
   
  
  
  
public class HttpGetDownloadImpl implements IDownload { 
   
  
  
  
 
   
  
  
  
    @Override 
   
  
  
  
    public Page download(String url) { 
   
  
  
  
        Page page = new Page(); 
   
  
  
  
        String content = HttpUtil.getHttpContent(url);  // 獲取網(wǎng)頁數(shù)據(jù) 
   
  
  
  
        page.setUrl(url); 
   
  
  
  
        page.setContent(content); 
   
  
  
  
        return page; 
   
  
  
  
    } 
   
  
  
  
} 
  
 
 
 

網(wǎng)頁解析器

網(wǎng)頁解析器就是把下載的網(wǎng)頁中我們感興趣的數(shù)據(jù)解析出來，并保存到某個對象中，供數(shù)據(jù)存儲器進(jìn)一步處理以保存到不同的持久化倉庫中，其基于下面的接口進(jìn)行開發(fā)：

  
 
 
 

   
  
  
  
/** 
   
  
  
  
 * 網(wǎng)頁數(shù)據(jù)解析 
   
  
  
  
 */ 
   
  
  
  
public interface IParser { 
   
  
  
  
    public void parser(Page page); 
   
  
  
  
} 
  
 
 
 

網(wǎng)頁解析器在整個系統(tǒng)的開發(fā)中也算是比較重頭戲的一個組件，功能不復(fù)雜，主要是代碼比較多，針對不同的商城不同的商品，對應(yīng)的解析器可能就不一樣了。

因此需要針對特別的商城的商品進(jìn)行開發(fā)，因?yàn)楹茱@然，京東用的網(wǎng)頁模板跟蘇寧易購的肯定不一樣，天貓用的跟京東用的也肯定不一樣。

所以這個完全是看自己的需要來進(jìn)行開發(fā)了，只是說，在解析器開發(fā)的過程當(dāng)中會發(fā)現(xiàn)有部分重復(fù)代碼，這時就可以把這些代碼抽象出來開發(fā)一個工具類了。

目前在系統(tǒng)中爬取的是京東和蘇寧易購的手機(jī)商品數(shù)據(jù)，因此就寫了這兩個實(shí)現(xiàn)類：

  
 
 
 

   
  
  
  
/** 
   
  
  
  
 * 解析京東商品的實(shí)現(xiàn)類 
   
  
  
  
 */ 
   
  
  
  
public class JDHtmlParserImpl implements IParser { 
   
  
  
  
    ...... 
   
  
  
  
} 
   
  
  
  
 
   
  
  
  
/** 
   
  
  
  
 * 蘇寧易購網(wǎng)頁解析 
   
  
  
  
 */ 
   
  
  
  
public class SNHtmlParserImpl implements IParser { 
   
  
  
  
    ...... 
   
  
  
  
} 
  
 
 
 

數(shù)據(jù)存儲器

數(shù)據(jù)存儲器主要是將網(wǎng)頁解析器解析出來的數(shù)據(jù)對象保存到不同的表格，而對于本次爬取的手機(jī)商品，數(shù)據(jù)對象是下面一個 Page 對象：

  
 
 
 

   
  
  
  
/** 
   
  
  
  
 * 網(wǎng)頁對象，主要包含網(wǎng)頁內(nèi)容和商品數(shù)據(jù) 
   
  
  
  
 */ 
   
  
  
  
public class Page { 
   
  
  
  
    private String content;              // 網(wǎng)頁內(nèi)容 
   
  
  
  
 
   
  
  
  
    private String id;                    // 商品Id 
   
  
  
  
    private String source;               // 商品來源 
   
  
  
  
    private String brand;                // 商品品牌 
   
  
  
  
    private String title;                // 商品標(biāo)題 
   
  
  
  
    private float price;                // 商品價格 
   
  
  
  
    private int commentCount;        // 商品評論數(shù) 
   
  
  
  
    private String url;                  // 商品地址 
   
  
  
  
    private String imgUrl;             // 商品圖片地址 
   
  
  
  
    private String params;              // 商品規(guī)格參數(shù) 
   
  
  
  
 
   
  
  
  
    private List urls = new ArrayList<>();  // 解析列表頁面時用來保存解析的商品url的容器 
   
  
  
  
} 
  
 
 
 

對應(yīng)的，在 MySQL 中，表數(shù)據(jù)結(jié)構(gòu)如下：

  
 
 
 

   
  
  
  
-- ---------------------------- 
   
  
  
  
-- Table structure for phone 
   
  
  
  
-- ---------------------------- 
   
  
  
  
DROP TABLE IF EXISTS `phone`; 
   
  
  
  
CREATE TABLE `phone` ( 
   
  
  
  
  `id` varchar(30) CHARACTER SET armscii8 NOT NULL COMMENT '商品id', 
   
  
  
  
  `source` varchar(30) NOT NULL COMMENT '商品來源，如jd suning gome等', 
   
  
  
  
  `brand` varchar(30) DEFAULT NULL COMMENT '手機(jī)品牌', 
   
  
  
  
  `title` varchar(255) DEFAULT NULL COMMENT '商品頁面的手機(jī)標(biāo)題', 
   
  
  
  
  `price` float(10,2) DEFAULT NULL COMMENT '手機(jī)價格', 
   
  
  
  
  `comment_count` varchar(30) DEFAULT NULL COMMENT '手機(jī)評論', 
   
  
  
  
  `url` varchar(500) DEFAULT NULL COMMENT '手機(jī)詳細(xì)信息地址', 
   
  
  
  
  `img_url` varchar(500) DEFAULT NULL COMMENT '圖片地址', 
   
  
  
  
  `params` text COMMENT '手機(jī)參數(shù)，json格式存儲', 
   
  
  
  
  PRIMARY KEY (`id`,`source`) 
   
  
  
  
) ENGINE=InnoDB DEFAULT CHARSET=utf8; 
  
 
 
 

而在 HBase 中的表結(jié)構(gòu)則為如下：

  
 
 
 

   
  
  
  
## cf1 存儲 id source price comment brand url 
   
  
  
  
## cf2 存儲 title params imgUrl 
   
  
  
  
create 'phone', 'cf1', 'cf2' 
   
  
  
  
 
   
  
  
  
## 在HBase shell中查看創(chuàng)建的表 
   
  
  
  
hbase(main):135:0> desc 'phone' 
   
  
  
  
Table phone is ENABLED                                                                                                 
   
  
  
  
phone                                                                                                                  
   
  
  
  
COLUMN FAMILIES DESCRIPTION                                                                                            
   
  
  
  
{NAME => 'cf1', BLOOMFILTER => 'ROW', VERSIONS => '1', IN_MEMORY => 'false', KEEP_DELETED_CELLS => 'FALSE', DATA_BLOCK 
   
  
  
  
_ENCODING => 'NONE', TTL => 'FOREVER', COMPRESSION => 'NONE', MIN_VERSIONS => '0', BLOCKCACHE => 'true', BLOCKSIZE =>  
   
  
  
  
'65536', REPLICATION_SCOPE => '0'}                                                                                     
   
  
  
  
{NAME => 'cf2', BLOOMFILTER => 'ROW', VERSIONS => '1', IN_MEMORY => 'false', KEEP_DELETED_CELLS => 'FALSE', DATA_BLOCK 
   
  
  
  
_ENCODING => 'NONE', TTL => 'FOREVER', COMPRESSION => 'NONE', MIN_VERSIONS => '0', BLOCKCACHE => 'true', BLOCKSIZE =>  
   
  
  
  
'65536', REPLICATION_SCOPE => '0'}                                                                                     
   
  
  
  
2 row(s) in 0.0350 seconds 
  
 
 
 

即在 HBase 中建立了兩個列族，分別為 cf1、cf2，其中 cf1 用來保存 id source price comment brand url 字段信息；cf2 用來保存 title params imgUrl 字段信息。

不同的數(shù)據(jù)存儲用的是不同的實(shí)現(xiàn)類，但是其都是基于下面同一個接口開發(fā)的：

  
 
 
 

   
  
  
  
/** 
   
  
  
  
 * 商品數(shù)據(jù)的存儲 
   
  
  
  
 */ 
   
  
  
  
public interface IStore { 
   
  
  
  
    public void store(Page page); 
   
  
  
  
} 
  
 
 
 

然后基于此開發(fā)了 MySQL 的存儲實(shí)現(xiàn)類、HBase 的存儲實(shí)現(xiàn)類還有控制臺的輸出實(shí)現(xiàn)類，如 MySQL 的存儲實(shí)現(xiàn)類，其實(shí)就是簡單的數(shù)據(jù)插入語句：

  
 
 
 

   
  
  
  
/** 
   
  
  
  
 * 使用dbc數(shù)據(jù)庫連接池將數(shù)據(jù)寫入mysql表中 
   
  
  
  
 */ 
   
  
  
  
public class MySQLStoreImpl implements IStore { 
   
  
  
  
    private QueryRunner queryRunner = new QueryRunner(DBCPUtil.getDataSource()); 
   
  
  
  
 
   
  
  
  
    @Override 
   
  
  
  
    public void store(Page page) { 
   
  
  
  
        String sql = "insert into phone(id, source, brand, title, price, comment_count, url, img_url, params) values(?, ?, ?, ?, ?, ?, ?, ?, ?)"; 
   
  
  
  
        try { 
   
  
  
  
            queryRunner.update(sql, page.getId(), 
   
  
  
  
                    page.getSource(), 
   
  
  
  
                    page.getBrand(), 
   
  
  
  
                    page.getTitle(), 
   
  
  
  
                    page.getPrice(), 
   
  
  
  
                    page.getCommentCount(), 
   
  
  
  
                    page.getUrl(), 
   
  
  
  
                    page.getImgUrl(), 
   
  
  
  
                    page.getParams()); 
   
  
  
  
        } catch (SQLException e) { 
   
  
  
  
            e.printStackTrace(); 
   
  
  
  
        } 
   
  
  
  
    } 
   
  
  
  
} 
  
 
 
 

而 HBase 的存儲實(shí)現(xiàn)類，則是 HBase Java API 的常用插入語句代碼：

  
 
 
 

   
  
  
  
...... 
   
  
  
  
// cf1:price 
   
  
  
  
Put pricePut = new Put(rowKey); 
   
  
  
  
// 必須要做是否為null判斷，否則會有空指針異常 
   
  
  
  
pricePut.addColumn(cf1, "price".getBytes(), page.getPrice() != null ? String.valueOf(page.getPrice()).getBytes() : "".getBytes()); 
   
  
  
  
puts.add(pricePut); 
   
  
  
  
// cf1:comment 
   
  
  
  
Put commentPut = new Put(rowKey); 
   
  
  
  
commentPut.addColumn(cf1, "comment".getBytes(), page.getCommentCount() != null ? String.valueOf(page.getCommentCount()).getBytes() : "".getBytes()); 
   
  
  
  
puts.add(commentPut); 
   
  
  
  
// cf1:brand 
   
  
  
  
Put brandPut = new Put(rowKey); 
   
  
  
  
brandPut.addColumn(cf1, "brand".getBytes(), page.getBrand() != null ? page.getBrand().getBytes() : "".getBytes()); 
   
  
  
  
puts.add(brandPut); 
   
  
  
  
...... 
  
 
 
 

當(dāng)然，至于要將數(shù)據(jù)存儲在哪個地方，在初始化爬蟲程序時，是可以手動選擇的：

  
 
 
 

   
  
  
  
// 3.注入存儲器 
   
  
  
  
iSpider.setStore(new HBaseStoreImpl()); 
  
 
 
 

目前還沒有把代碼寫成可以同時存儲在多個地方，按照目前代碼的架構(gòu)，要實(shí)現(xiàn)這一點(diǎn)也比較簡單，修改一下相應(yīng)代碼就好了。

實(shí)際上，是可以先把數(shù)據(jù)保存到 MySQL 中，然后通過 Sqoop 導(dǎo)入到 HBase 中，詳細(xì)操作可以參考我寫的 Sqoop 文章。

仍然需要注意的是，如果確定需要將數(shù)據(jù)保存到 HBase 中，請保證你有可用的集群環(huán)境，并且需要將如下配置文檔添加到 classpath 下：

  
 
 
 

   
  
  
  
core-site.xml 
   
  
  
  
hbase-site.xml 
   
  
  
  
hdfs-site.xml 
  
 
 
 

對大數(shù)據(jù)感興趣的同學(xué)可以折騰一下這一點(diǎn)，如果之前沒有接觸過的，直接使用 MySQL 存儲就好了，只需要在初始化爬蟲程序時注入 MySQL 存儲器即可：

  
 
 
 

   
  
  
  
// 3.注入存儲器 
   
  
  
  
iSpider.setStore(new MySQLStoreImpl()); 
  
 
 
 

URL 調(diào)度系統(tǒng)

URL 調(diào)度系統(tǒng)是實(shí)現(xiàn)整個爬蟲系統(tǒng)分布式的橋梁與關(guān)鍵，正是通過 URL 調(diào)度系統(tǒng)的使用，才使得整個爬蟲系統(tǒng)可以較為高效（Redis 作為存儲）隨機(jī)地獲取 URL，并實(shí)現(xiàn)整個系統(tǒng)的分布式。

URL 倉庫

通過架構(gòu)圖可以看出，所謂的 URL 倉庫不過是 Redis 倉庫，即在我們的系統(tǒng)中使用 Redis 來保存 URL 地址列表。

正是這樣，才能保證我們的程序?qū)崿F(xiàn)分布式，只要保存了 URL 是唯一的，這樣不管我們的爬蟲程序有多少個，最終保存下來的數(shù)據(jù)都是只有唯一一份的，而不會重復(fù)。

同時 URL 倉庫中的 URL 地址在獲取時的策略是通過隊(duì)列的方式來實(shí)現(xiàn)的，待會通過 URL 調(diào)度器的實(shí)現(xiàn)即可知道。

另外，在我們的 URL 倉庫中，主要保存了下面的數(shù)據(jù)：

種子 URL 列表，Redis 的數(shù)據(jù)類型為 list

種子 URL 是持久化存儲的，一定時間后，由 URL 定時器通過種子 URL 獲取 URL，并將其注入到我們的爬蟲程序需要使用的高優(yōu)先級 URL 隊(duì)列中。

這樣就可以保證我們的爬蟲程序可以源源不斷地爬取數(shù)據(jù)而不需要中止程序的執(zhí)行。

高優(yōu)先級 URL 隊(duì)列，Redis 的數(shù)據(jù)類型為 set

什么是高優(yōu)先級 URL 隊(duì)列？其實(shí)它就是用來保存列表 URL 的。那么什么是列表 URL 呢？

說白了就是一個列表中含有多個商品，以京東為例，我們打開一個手機(jī)列表：

該地址中包含的不是一個具體商品的 URL，而是包含了多個我們需要爬取的數(shù)據(jù)（手機(jī)商品）的列表。

通過對每個高級 URL 的解析，我們可以獲取到非常多的具體商品 URL，而具體的商品 URL，就是低優(yōu)先 URL，其會保存到低優(yōu)先級 URL 隊(duì)列中。

那么以這個系統(tǒng)為例，保存的數(shù)據(jù)類似如下：

  
 
 
 

   
  
  
  
jd.com.higher 
   
  
  
  
    --https://list.jd.com/list.html?cat=9987,653,655&page=1 
   
  
  
  
    ...  
   
  
  
  
suning.com.higher 
   
  
  
  
    --https://list.suning.com/0-20006-0.html 
   
  
  
  
    ... 
  
 
 
 

低優(yōu)先級 URL 隊(duì)列，Redis 的數(shù)據(jù)類型為 set

低優(yōu)先級 URL 其實(shí)就是具體某個商品的 URL，如下面一個手機(jī)商品：

通過下載該 URL 的數(shù)據(jù)，并對其進(jìn)行解析，就能夠獲取到我們想要的數(shù)據(jù)。

那么以這個系統(tǒng)為例，保存的數(shù)據(jù)類似如下：

  
 
 
 

   
  
  
  
jd.com.lower 
   
  
  
  
    --https://item.jd.com/23545806622.html 
   
  
  
  
    ... 
   
  
  
  
suning.com.lower 
   
  
  
  
    --https://product.suning.com/0000000000/690128156.html 
   
  
  
  
    ... 
  
 
 
 

URL 調(diào)度器

所謂 URL 調(diào)度器，就是 URL 倉庫 Java 代碼的調(diào)度策略，不過因?yàn)槠浜诵脑谟谡{(diào)度，所以將其放到 URL 調(diào)度器中來進(jìn)行說明，目前其調(diào)度基于以下接口開發(fā)：

  
 
 
 

   
  
  
  
/** 
   
  
  
  
 * url 倉庫 
   
  
  
  
 * 主要功能： 
   
  
  
  
 *      向倉庫中添加url（高優(yōu)先級的列表，低優(yōu)先級的商品url） 
   
  
  
  
 *      從倉庫中獲取url（優(yōu)先獲取高優(yōu)先級的url，如果沒有，再獲取低優(yōu)先級的url） 
   
  
  
  
 * 
   
  
  
  
 */ 
   
  
  
  
public interface IRepository { 
   
  
  
  
 
   
  
  
  
    /** 
   
  
  
  
     * 獲取url的方法 
   
  
  
  
     * 從倉庫中獲取url（優(yōu)先獲取高優(yōu)先級的url，如果沒有，再獲取低優(yōu)先級的url） 
   
  
  
  
     * @return 
   
  
  
  
     */ 
   
  
  
  
    public String poll(); 
   
  
  
  
 
   
  
  
  
    /** 
   
  
  
  
     * 向高優(yōu)先級列表中添加商品列表url 
   
  
  
  
     * @param highUrl 
   
  
  
  
     */ 
   
  
  
  
    public void offerHigher(String highUrl); 
   
  
  
  
 
   
  
  
  
    /** 
   
  
  
  
     * 向低優(yōu)先級列表中添加商品url 
   
  
  
  
     * @param lowUrl 
   
  
  
  
     */ 
   
  
  
  
    public void offerLower(String lowUrl); 
   
  
  
  
 
   
  
  
  
} 
  
 
 
 

其基于 Redis 作為 URL 倉庫的實(shí)現(xiàn)如下：

  
 
 
 

   
  
  
  
/** 
   
  
  
  
 * 基于Redis的全網(wǎng)爬蟲，隨機(jī)獲取爬蟲url： 
   
  
  
  
 * 
   
  
  
  
 * Redis中用來保存url的數(shù)據(jù)結(jié)構(gòu)如下： 
   
  
  
  
 * 1.需要爬取的域名集合（存儲數(shù)據(jù)類型為set，這個需要先在Redis中添加） 
   
  
  
  
 *      key 
   
  
  
  
 *          spider.website.domains 
   
  
  
  
 *      value(set) 
   
  
  
  
 *          jd.com  suning.com  gome.com 
   
  
  
  
 *      key由常量對象SpiderConstants.SPIDER_WEBSITE_DOMAINS_KEY 獲得 
   
  
  
  
 * 2.各個域名所對應(yīng)的高低優(yōu)先url隊(duì)列（存儲數(shù)據(jù)類型為list，這個由爬蟲程序解析種子url后動態(tài)添加） 
   
  
  
  
 *      key 
   
  
  
  
 *          jd.com.higher 
   
  
  
  
 *          jd.com.lower 
   
  
  
  
 *          suning.com.higher 
   
  
  
  
 *          suning.com.lower 
   
  
  
  
 *          gome.com.higher 
   
  
  
  
 *          gome.come.lower 
   
  
  
  
 *      value(list) 
   
  
  
  
 *          相對應(yīng)需要解析的url列表 
   
  
  
  
 *      key由隨機(jī)的域名 + 常量 SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX或者SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX獲得 
   
  
  
  
 * 3.種子url列表 
   
  
  
  
 *      key 
   
  
  
  
 *          spider.seed.urls 
   
  
  
  
 *      value(list) 
   
  
  
  
 *          需要爬取的數(shù)據(jù)的種子url 
   
  
  
  
 *       key由常量SpiderConstants.SPIDER_SEED_URLS_KEY獲得 
   
  
  
  
 * 
   
  
  
  
 *       種子url列表中的url會由url調(diào)度器定時向高低優(yōu)先url隊(duì)列中 
   
  
  
  
 */ 
   
  
  
  
public class RandomRedisRepositoryImpl implements IRepository { 
   
  
  
  
 
   
  
  
  
    /** 
   
  
  
  
     * 構(gòu)造方法 
   
  
  
  
     */ 
   
  
  
  
    public RandomRedisRepositoryImpl() { 
   
  
  
  
        init(); 
   
  
  
  
    } 
   
  
  
  
 
   
  
  
  
    /** 
   
  
  
  
     * 初始化方法，初始化時，先將redis中存在的高低優(yōu)先級url隊(duì)列全部刪除 
   
  
  
  
     * 否則上一次url隊(duì)列中的url沒有消耗完時，再停止啟動跑下一次，就會導(dǎo)致url倉庫中有重復(fù)的url 
   
  
  
  
     */ 
   
  
  
  
    public void init() { 
   
  
  
  
        Jedis jedis = JedisUtil.getJedis(); 
   
  
  
  
        Set domains = jedis.smembers(SpiderConstants.SPIDER_WEBSITE_DOMAINS_KEY); 
   
  
  
  
        String higherUrlKey; 
   
  
  
  
        String lowerUrlKey; 
   
  
  
  
        for(String domain : domains) { 
   
  
  
  
            higherUrlKey = domain + SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX; 
   
  
  
  
            lowerUrlKey = domain + SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX; 
   
  
  
  
            jedis.del(higherUrlKey, lowerUrlKey); 
   
  
  
  
        } 
   
  
  
  
        JedisUtil.returnJedis(jedis); 
   
  
  
  
    } 
   
  
  
  
 
   
  
  
  
    /** 
   
  
  
  
     * 從隊(duì)列中獲取url，目前的策略是： 
   
  
  
  
     *      1.先從高優(yōu)先級url隊(duì)列中獲取 
   
  
  
  
     *      2.再從低優(yōu)先級url隊(duì)列中獲取 
   
  
  
  
     *  對應(yīng)我們的實(shí)際場景，應(yīng)該是先解析完列表url再解析商品url 
   
  
  
  
     *  但是需要注意的是，在分布式多線程的環(huán)境下，肯定是不能完全保證的，因?yàn)樵谀硞€時刻高優(yōu)先級url隊(duì)列中 
   
  
  
  
     *  的url消耗完了，但實(shí)際上程序還在解析下一個高優(yōu)先級url，此時，其它線程去獲取高優(yōu)先級隊(duì)列url肯定獲取不到 
   
  
  
  
     *  這時就會去獲取低優(yōu)先級隊(duì)列中的url，在實(shí)際考慮分析時，這點(diǎn)尤其需要注意 
   
  
  
  
     * @return 
   
  
  
  
     */ 
   
  
  
  
    @Override 
   
  
  
  
    public String poll() { 
   
  
  
  
        // 從set中隨機(jī)獲取一個***域名 
   
  
  
  
        Jedis jedis = JedisUtil.getJedis(); 
   
  
  
  
        String randomDomain = jedis.srandmember(SpiderConstants.SPIDER_WEBSITE_DOMAINS_KEY);    // jd.com 
   
  
  
  
        String key = randomDomain + SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX;                // jd.com.higher 
   
  
  
  
        String url = jedis.lpop(key); 
   
  
  
  
        if(url == null) {   // 如果為null，則從低優(yōu)先級中獲取 
   
  
  
  
            key = randomDomain + SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX;    // jd.com.lower 
   
  
  
  
            url = jedis.lpop(key); 
   
  
  
  
        } 
   
  
  
  
        JedisUtil.returnJedis(jedis); 
   
  
  
  
        return url; 
   
  
  
  
    } 
   
  
  
  
 
   
  
  
  
    /** 
   
  
  
  
     * 向高優(yōu)先級url隊(duì)列中添加url 
   
  
  
  
     * @param highUrl 
   
  
  
  
     */ 
   
  
  
  
    @Override 
   
  
  
  
    public void offerHigher(String highUrl) { 
   
  
  
  
        offerUrl(highUrl, SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX); 
   
  
  
  
    } 
   
  
  
  
 
   
  
  
  
    /** 
   
  
  
  
     * 向低優(yōu)先url隊(duì)列中添加url 
   
  
  
  
     * @param lowUrl 
   
  
  
  
     */ 
   
  
  
  
    @Override 
   
  
  
  
    public void offerLower(String lowUrl) { 
   
  
  
  
        offerUrl(lowUrl, SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX); 
   
  
  
  
    } 
   
  
  
  
 
   
  
  
  
    /** 
   
  
  
  
     * 添加url的通用方法，通過offerHigher和offerLower抽象而來 
   
  
  
  
     * @param url   需要添加的url 
   
  
  
  
     * @param urlTypeSuffix  url類型后綴.higher或.lower 
   
  
  
  
     */ 
   
  
  
  
    public void offerUrl(String url, String urlTypeSuffix) { 
   
  
  
  
        Jedis jedis = JedisUtil.getJedis(); 
   
  
  
  
        String domain = SpiderUtil.getTopDomain(url);   // 獲取url對應(yīng)的***域名，如jd.com 
   
  
  
  
        String key = domain + urlTypeSuffix;            // 拼接url隊(duì)列的key，如jd.com.higher 
   
  
  
  
        jedis.lpush(key, url);                          // 向url隊(duì)列中添加url 
   
  
  
  
        JedisUtil.returnJedis(jedis); 
   
  
  
  
    } 
   
  
  
  
} 
  
 
 
 

通過代碼分析也可以知道，其核心就在如何調(diào)度 URL 倉庫（Redis）中的 URL。

URL 定時器

一段時間后，高優(yōu)先級 URL 隊(duì)列和低優(yōu)先 URL 隊(duì)列中的 URL 都會被消費(fèi)完。

為了讓程序可以繼續(xù)爬取數(shù)據(jù)，同時減少人為的干預(yù)，可以預(yù)先在 Redis 中插入種子 URL，之后定時讓 URL 定時器從種子 URL 中取出 URL 存放到高優(yōu)先級 URL 隊(duì)列中，以此達(dá)到程序定時不間斷爬取數(shù)據(jù)的目的。

URL 消費(fèi)完畢后，是否需要循環(huán)不斷爬取數(shù)據(jù)根據(jù)個人業(yè)務(wù)需求而不同，因此這一步不是必需的，只是也提供了這樣的操作。

因?yàn)槭聦?shí)上，我們需要爬取的數(shù)據(jù)也是每隔一段時間就會更新的，如果希望我們爬取的數(shù)據(jù)也跟著定時更新，那么這時定時器就有非常重要的作用了。

不過需要注意的是，一旦決定需要循環(huán)重復(fù)爬取數(shù)據(jù)，則在設(shè)計存儲器實(shí)現(xiàn)時需要考慮重復(fù)數(shù)據(jù)的問題，即重復(fù)數(shù)據(jù)應(yīng)該是更新操作。

目前在我設(shè)計的存儲器不包括這個功能，有興趣的朋友可以自己實(shí)現(xiàn)，只需要在插入數(shù)據(jù)前判斷數(shù)據(jù)庫中是否存在該數(shù)據(jù)即可。

另外需要注意的一點(diǎn)是，URL 定時器是一個獨(dú)立的進(jìn)程，需要單獨(dú)啟動。

定時器基于 Quartz 實(shí)現(xiàn)，下面是其 job 的代碼：

  
 
 
 

   
  
  
  
/** 
   
  
  
  
 * 每天定時從url倉庫中獲取種子url，添加進(jìn)高優(yōu)先級列表 
   
  
  
  
 */ 
   
  
  
  
public class UrlJob implements Job { 
   
  
  
  
 
   
  
  
  
    // log4j日志記錄 
   
  
  
  
    private Logger logger = LoggerFactory.getLogger(UrlJob.class); 
   
  
  
  
 
   
  
  
  
    @Override 
   
  
  
  
    public void execute(JobExecutionContext context) throws JobExecutionException { 
   
  
  
  
        /** 
   
  
  
  
         * 1.從指定url種子倉庫獲取種子url 
   
  
  
  
         * 2.將種子url添加進(jìn)高優(yōu)先級列表 
   
  
  
  
         */ 
   
  
  
  
        Jedis jedis = JedisUtil.getJedis(); 
   
  
  
  
        Set seedUrls = jedis.smembers(SpiderConstants.SPIDER_SEED_URLS_KEY);  // spider.seed.urls Redis數(shù)據(jù)類型為set，防止重復(fù)添加種子url 
   
  
  
  
        for(String seedUrl : seedUrls) { 
   
  
  
  
            String domain = SpiderUtil.getTopDomain(seedUrl);   // 種子url的***域名 
   
  
  
  
            jedis.sadd(domain + SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX, seedUrl); 
   
  
  
  
            logger.info("獲取種子:{}", seedUrl); 
   
  
  
  
        } 
   
  
  
  
        JedisUtil.returnJedis(jedis); 
   
  
  
  
//        System.out.println("Scheduler Job Test..."); 
   
  
  
  
    } 
   
  
  
  
 
   
  
  
  
} 
  
 
 
 

調(diào)度器的實(shí)現(xiàn)如下：

  
 
 
 

   
  
  
  
/** 
   
  
  
  
 * url定時調(diào)度器，定時向url對應(yīng)倉庫中存放種子url 
   
  
  
  
 * 
   
  
  
  
 * 業(yè)務(wù)規(guī)定：每天凌晨1點(diǎn)10分向倉庫中存放種子url 
   
  
  
  
 */ 
   
  
  
  
public class UrlJobScheduler { 
   
  
  
  
 
   
  
  
  
    public UrlJobScheduler() { 
   
  
  
  
        init(); 
   
  
  
  
    } 
   
  
  
  
 
   
  
  
  
    /** 
   
  
  
  
     * 初始化調(diào)度器 
   
  
  
  
     */ 
   
  
  
  
    public void init() { 
   
  
  
  
        try { 
   
  
  
  
            Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler(); 
   
  
  
  
 
   
  
  
  
            // 如果沒有以下start方法的執(zhí)行，則是不會開啟任務(wù)的調(diào)度 
   
  
  
  
                                                  

                                                網(wǎng)頁名稱：手把手教你搭建一個基于Java的分布式爬蟲系統(tǒng)                                                

                                                網(wǎng)頁URL：http://www.dlmjj.cn/article/dpgpddd.html