基于Vert.x和RxJava 2構(gòu)建通用的爬蟲框架的示例

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

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

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

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

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

一. 爬蟲框架的功能

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

1.1 Spider

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

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

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

Downloader的download方法會(huì)返回一個(gè)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>對象來實(shí)現(xiàn)后續(xù)的一系列的鏈?zhǔn)秸{(diào)用，比如將Response轉(zhuǎn)換成Page對象，再對Page對象進(jìn)行解析，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可以讓整個(gè)爬蟲框架看起來更加響應(yīng)式：）

1.2 SpiderEngine

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

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

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

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

類型：GET

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

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

類型：GET

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

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

類型：POST

參數(shù)說明：

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

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

使用框架的例子

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

    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();

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

我們能看到三個(gè)爬蟲運(yùn)行的結(jié)果。

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

案例

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

目前該程序已經(jīng)上線，可以通過詢問我的公眾號(hào)，來實(shí)時(shí)獲取這幾種數(shù)字貨幣的最新價(jià)格。

TODO

增加對登錄驗(yàn)證碼的識(shí)別
增加elasticsearch的支持

總結(jié)

這個(gè)爬蟲框架才剛剛起步，我也參考了很多優(yōu)秀的爬蟲框架。未來我會(huì)在框架中考慮增加通過截屏圖片來分析圖片中的數(shù)據(jù)。甚至?xí)Y(jié)合cv4j框架。過年前，在爬蟲框架中會(huì)優(yōu)先實(shí)現(xiàn)對登錄驗(yàn)證碼的識(shí)別。

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: