基于Vert.x和RxJava 2構建通用的爬蟲框架的示例

更新時間：2018年02月01日 10:28:53 作者：fengzhizi715

這篇文章主要介紹了基于Vert.x和RxJava 2構建通用的爬蟲框架的示例，小編覺得挺不錯的，現在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧

最近由于業(yè)務需要監(jiān)控一些數據，雖然市面上有很多優(yōu)秀的爬蟲框架，但是我仍然打算從頭開始實現一套完整的爬蟲框架。

在技術選型上，我沒有選擇Spring來搭建項目，而是選擇了更輕量級的Vert.x。一方面感覺Spring太重了，而Vert.x是一個基于JVM、輕量級、高性能的框架。它基于事件和異步，依托于全異步Java服務器Netty，并擴展了很多其他特性。

github地址：https://github.com/fengzhizi715/NetDiscovery

一. 爬蟲框架的功能

爬蟲框架包含爬蟲引擎(SpiderEngine)和爬蟲(Spider)。SpiderEngine可以管理多個Spider。

1.1 Spider

在Spider中，主要包含幾個組件：downloader、queue、parser、pipeline以及代理池IP(proxypool)，代理池是一個單獨的項目，我前段時間寫的，在使用爬蟲框架時經常需要切換代理IP，所以把它引入進來。

proxypool地址：https://github.com/fengzhizi715/ProxyPool

其余四個組件都是接口，在爬蟲框架中內置了一些實現，例如內置了多個下載器(downloader)包括vertx的webclient、http client、okhttp3、selenium實現的下載器。開發(fā)者可以根據自身情況來選擇使用或者自己開發(fā)全新的downloader。

Downloader的download方法會返回一個Maybe<Response>。

package com.cv4j.netdiscovery.core.downloader;
import com.cv4j.netdiscovery.core.domain.Request;
import com.cv4j.netdiscovery.core.domain.Response;
import io.reactivex.Maybe;

/**
 * Created by tony on 2017/12/23.
 */
public interface Downloader {
  Maybe<Response> download(Request request);
  void close();
}

在Spider中，通過Maybe<Response>對象來實現后續(xù)的一系列的鏈式調用，比如將Response轉換成Page對象，再對Page對象進行解析，Page解析完畢之后做一系列的pipeline操作。

         downloader.download(request)
              .observeOn(Schedulers.io())
              .map(new Function<Response, Page>() {

                @Override
                public Page apply(Response response) throws Exception {
                  Page page = new Page();
                  page.setHtml(new Html(response.getContent()));
                  page.setRequest(request);
                  page.setUrl(request.getUrl());
                  page.setStatusCode(response.getStatusCode());
                  return page;
                }
              })
              .map(new Function<Page, Page>() {

                @Override
                public Page apply(Page page) throws Exception {

                  if (parser != null) {

                    parser.process(page);
                  }

                  return page;
                }
              })
              .map(new Function<Page, Page>() {

                @Override
                public Page apply(Page page) throws Exception {

                  if (Preconditions.isNotBlank(pipelines)) {

                    pipelines.stream()
                        .forEach(pipeline -> pipeline.process(page.getResultItems()));
                  }

                  return page;
                }
              })
              .subscribe(new Consumer<Page>() {

                @Override
                public void accept(Page page) throws Exception {

                  log.info(page.getUrl());

                  if (request.getAfterRequest()!=null) {

                    request.getAfterRequest().process(page);
                  }
                }
              }, new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) throws Exception {
                  log.error(throwable.getMessage());
                }
              });

在這里使用RxJava 2可以讓整個爬蟲框架看起來更加響應式：）

1.2 SpiderEngine

SpiderEngine可以包含多個Spider，可以通過addSpider()、createSpider()來將爬蟲添加到SpiderEngine和創(chuàng)建新的Spider并添加到SpiderEngine。

在SpiderEngine中，如果調用了httpd(port)方法，還可以監(jiān)控SpiderEngine中各個Spider。

1.2.1 獲取某個爬蟲的狀態(tài)

http://localhost:{port}/netdiscovery/spider/{spiderName}

類型：GET

1.2.2 獲取SpiderEngine中所有爬蟲的狀態(tài)

http://localhost:{port}/netdiscovery/spiders/

類型：GET

1.2.3 修改某個爬蟲的狀態(tài)

http://localhost:{port}/netdiscovery/spider/{spiderName}/status

類型：POST

參數說明：

{
  "status":2  //讓爬蟲暫停
}

status	作用
2	讓爬蟲暫停
3	讓爬蟲從暫停中恢復
4	讓爬蟲停止

使用框架的例子

創(chuàng)建一個SpiderEngine，然后創(chuàng)建三個Spider，每個爬蟲每隔一定的時間去爬取一個頁面。

    SpiderEngine engine = SpiderEngine.create();
    Spider spider = Spider.create()
        .name("tony1")
        .repeatRequest(10000,"http://www.163.com")
        .initialDelay(10000);

    engine.addSpider(spider);
    Spider spider2 = Spider.create()
        .name("tony2")
        .repeatRequest(10000,"http://www.baidu.com")
        .initialDelay(10000);

    engine.addSpider(spider2);
    Spider spider3 = Spider.create()
        .name("tony3")
        .repeatRequest(10000,"http://www.126.com")
        .initialDelay(10000);
    engine.addSpider(spider3);
    engine.httpd(8080);
    engine.run();

上述程序運行一段時間之后，在瀏覽器中輸入：http://localhost:8080/netdiscovery/spiders

我們能看到三個爬蟲運行的結果。

將json格式化一下

{
  "code": 200,
  "data": [{
    "downloaderType": "VertxDownloader",
    "leftRequestSize": 0,
    "queueType": "DefaultQueue",
    "spiderName": "tony2",
    "spiderStatus": 1,
    "totalRequestSize": 7
  }, {
    "downloaderType": "VertxDownloader",
    "leftRequestSize": 0,
    "queueType": "DefaultQueue",
    "spiderName": "tony3",
    "spiderStatus": 1,
    "totalRequestSize": 7
  }, {
    "downloaderType": "VertxDownloader",
    "leftRequestSize": 0,
    "queueType": "DefaultQueue",
    "spiderName": "tony1",
    "spiderStatus": 1,
    "totalRequestSize": 7
  }],
  "message": "success"
}

案例

最近比較關注區(qū)塊鏈，因此做了一個程序來實時抓取三種數字貨幣的價格，可以通過“詢問”公眾號的方式來獲取最新的價格。