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

2019-10-22 18:15:49

字體：大中小

來源：轉載

供稿：網友

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

在技術選型上，我沒有選擇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

Vert.x,RxJava2,爬蟲框架

其余四個組件都是接口，在爬蟲框架中內置了一些實現，例如內置了多個下載器(downloader)包括vertx的webclient、http client、okhttp3、selenium實現的下載器。開發者可以根據自身情況來選擇使用或者自己開發全新的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>對象來實現后續的一系列的鏈式調用，比如將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可以讓整個爬蟲框架看起來更加響應式：）

Vert.x,RxJava2,爬蟲框架

1.2 SpiderEngine

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

Vert.x,RxJava2,爬蟲框架

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

1.2.1 獲取某個爬蟲的狀態

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

類型：GET

1.2.2 獲取SpiderEngine中所有爬蟲的狀態

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

類型：GET

1.2.3 修改某個爬蟲的狀態

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

類型：POST

參數說明：

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

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

使用框架的例子

創建一個SpiderEngine，然后創建三個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

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

Vert.x,RxJava2,爬蟲框架

將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"}

案例

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

Vert.x,RxJava2,爬蟲框架