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

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

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

{
  "status":2  //让爬虫暂停
}

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

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