玩轉webmagic代碼之Scheduler

webmagic上線以後，由於靈活性很強，獲得了一些爬蟲老手的歡迎，可是對於新手來講可能稍微摸不着頭腦，個人需求是這樣子，什麼模塊化，什麼靈活性，可是看了半天，我也不知道怎麼解決個人問題啊？

這裏先談談Scheduler，不單關乎框架，更可能是一些爬蟲通用的思想，但願對你們有幫助。

爲何要有Scheduler

其實Scheduler並不是webmagic首創，在scrapy以及其餘成熟爬蟲中都有相似模塊。Scheduler管理了全部待抓取的url，單個爬蟲本身是沒法控制要抓取什麼的，抓什麼都由Scheduler決定。

這樣子最大的好處就是，爬蟲自己沒有狀態，給一個url，處理一個，很是容易進行水平擴展(就是加線程、或者加機器)，並且即便單臺爬蟲宕機，也不會有什麼損失。這跟咱們在應用開發中，所說的"服務無狀態"的思想是很像的。而相反，若是在單個爬蟲線程內部，循環甚至遞歸的進行抓取，那麼這部分工做是沒法擴展的，並且宕機以後恢復會很困難。

<!-- lang: java -->
public interface Scheduler {

    public void push(Request request, Task task);

    public Request poll(Task task);

}

webmagic裏的Scheduler只有兩個接口，一個放入url，一個取出url。

玩轉Scheduler

層級關係及上下文信息

咱們這裏舉一個較複雜的例子。例如，咱們要從http://www.ip138.com/post/上抓取全國的郵編地址，最後咱們想要獲得一個樹狀結構的結果，這個結果包括省 市 縣 村/街道 郵編。這裏有兩個需求：一個是優先抓最終頁面，一個是要帶上全部前面頁面的信息。若是隨便手寫一個爬蟲，可能咱們就會用遞歸的形式寫了，那麼在webmagic裏如何作呢？

從0.2.1起，webmagic的Request，也就是保存待抓取url的對象，有兩個大的改動：

一個是支持優先級，這樣子要深度優先仍是廣度優先，均可以經過給不一樣層次設置不一樣值完成。

二是能夠在Request裏附加額外信息request.putExtra(key,value)，這個額外信息會帶到下次頁面抓取中去。

因而，咱們能夠經過給最終頁面增長高優先級，達到優先抓取的目的；同時能夠把以前抓取的信息保存到Request裏去，在最終結果中，附加上前面頁面的信息。

最終代碼在這裏，固然，其實這個例子裏，最終頁面是包含「省」、「市」信息的，這裏只是討論附加信息的可能性。

<!-- lang: java -->
public class ZipCodePageProcessor implements PageProcessor {

    private Site site = Site.me().setCharset("gb2312")
            .setSleepTime(100).addStartUrl("http://www.ip138.com/post/");

    @Override
    public void process(Page page) {
        if (page.getUrl().toString().equals("http://www.ip138.com/post/")) {
            processCountry(page);
        } else if (page.getUrl().regex("http://www\\.ip138\\.com/post/\\w+[/]?$").toString() != null) {
            processProvince(page);
        } else {
            processDistrict(page);
        }

    }

    private void processCountry(Page page) {
        List<String> provinces = page.getHtml().xpath("//*[@id=\"newAlexa\"]/table/tbody/tr/td").all();
        for (String province : provinces) {
            String link = xpath("//@href").select(province);
            String title = xpath("/text()").select(province);
            Request request = new Request(link).setPriority(0).putExtra("province", title);
            page.addTargetRequest(request);
        }
    }

    private void processProvince(Page page) {
        //這裏僅靠xpath無法精準定位，因此使用正則做爲篩選，不符合正則的會被過濾掉
        List<String> districts = page.getHtml().xpath("//body/table/tbody/tr/td").regex(".*http://www\\.ip138\\.com/post/\\w+/\\w+.*").all();
        for (String district : districts) {
            String link = xpath("//@href").select(district);
            String title = xpath("/text()").select(district);
            Request request = new Request(link).setPriority(1).putExtra("province", page.getRequest().getExtra("province")).putExtra("district", title);
            page.addTargetRequest(request);
        }
    }

    private void processDistrict(Page page) {
        String province = page.getRequest().getExtra("province").toString();
        String district = page.getRequest().getExtra("district").toString();
        List<String> counties = page.getHtml().xpath("//body/table/tbody/tr").regex(".*<td>\\d+</td>.*").all();
        String regex = "<td[^<>]*>([^<>]+)</td><td[^<>]*>([^<>]+)</td><td[^<>]*>([^<>]+)</td><td[^<>]*>([^<>]+)</td>";
        for (String county : counties) {
            String county0 = regex(regex, 1).select(county);
            String county1 = regex(regex, 2).select(county);
            String zipCode = regex(regex, 3).select(county);
            page.putField("result", StringUtils.join(new String[]{province, district,
                    county0, county1, zipCode}, "\t"));
        }
        List<String> links = page.getHtml().links().regex("http://www\\.ip138\\.com/post/\\w+/\\w+").all();
        for (String link : links) {
            page.addTargetRequest(new Request(link).setPriority(2).putExtra("province", province).putExtra("district", district));
        }

    }

    @Override
    public Site getSite() {
        return site;
    }

    public static void main(String[] args) {
        Spider.create(new ZipCodePageProcessor()).scheduler(new PriorityScheduler()).run();
    }
}

這段代碼略複雜，由於咱們其實進行了了3種頁面的抽取，論單個頁面，仍是挺簡單的:)

一樣的，咱們能夠實現一個最多抓取n層的爬蟲。經過在request.extra裏增長一個"層數"的概念便可作到，而Scheduler只需作少許定製：

<!-- lang: java -->
public class LevelLimitScheduler extends PriorityScheduler {

    private int levelLimit = 3;

    public LevelLimitScheduler(int levelLimit) {
        this.levelLimit = levelLimit;
    }

    @Override
    public synchronized void push(Request request, Task task) {
        if (((Integer) request.getExtra("_level")) <= levelLimit) {
            super.push(request, task);
        }
    }
}

按照指定URL查詢

例如我想要抓取百度某些關鍵詞查詢的結果，這個需求再簡單不過了，你能夠先新建一個Scheduler，將想要查詢的URL所有放入Scheduler以後，再啓動Spider便可：

<!-- lang: java -->
PriorityScheduler scheduler = new PriorityScheduler();
Spider spider = Spider.create(new ZipCodePageProcessor()).scheduler(scheduler);
scheduler.push(new Request("http://www.baidu.com/s?wd=webmagic"),spider);
//這裏webmagic是關鍵詞
...//其餘地址
spider.run();

按期輪詢

有一類需求是，按期檢查頁面是否更新，若是更新，則抓取最新數據。這裏包括兩個問題：

按期抓取和更新持久化數據。後者在Pipeline分享時候再說。

而按期輪詢，最簡單的方法就是按期去啓動Spider.run()。這樣子沒什麼問題，只是不夠優雅，還有一種方法是用Scheduler作按期分發，一次性把URL放進去，而後隔一段時間間隔後，再把url取出來。我這裏基於DelayQueue進行了一個實現：DelayQueueScheduler，大體思路就是這樣。

分佈式

webmagic裏有一個基於redis的RedisScheduler，能夠實現較簡單的分佈式功能。選用redis是由於redis比較輕量，同時有強大的數據結構支持。實際上更爲通用的方法是：將隊列管理和url去重拆分開來，用對應的工具去作。

url隊列，實際上很適合的載體工具就是各類消息隊列，例如JMS的實現ActiveMQ。固然若是你對關係數據庫比較熟悉，用它們來處理也是沒有問題的。

關於去重，就現成的工具來講的話，卻是沒有什麼比redis更合適了。固然，你也能夠本身構建一個去重服務，利用bloom filter等算法減小內存開銷。

玩轉webmagic系列之後會不按期更新，但願對你們有幫助。

最後依然附上 webmagic的github地址：

https://github.com/code4craft/webmagic