1 概述

对大数据感兴趣的同学可以折腾一下这一点，如果之前没有接触过的，直接使用MySQL存储就好了，只需要在初始化爬虫程序时注入MySQL存储器即可：

4 URL调度系统

URL调度系统是实现整个爬虫系统分布式的桥梁与关键，正是通过URL调度系统的使用，才使得整个爬虫系统可以较为高效（Redis作为存储）随机地获取url，并实现整个系统的分布式。

4.1 URL仓库

通过架构图可以看出，所谓的URL仓库不过是Redis仓库，即在我们的系统中使用Redis来保存url地址列表，正是这样，才能保证我们的程序实现分布式，只要保存了url是唯一的，这样不管我们的爬虫程序有多少个，最终保存下来的数据都是只有唯一一份的，而不会重复，是通过这样来实现分布式的。

同时url仓库中的url地址在获取时的策略是通过队列的方式来实现的，待会通过URL调度器的实现即可知道。

另外，在我们的url仓库中，主要保存了下面的数据：

• 种子URL列表

Redis的数据类型为list。

种子URL是持久化存储的，一定时间后，由URL定时器通过种子URL获取URL，并将其注入到我们的爬虫程序需要使用的高优先级URL队列中，这样就可以保存我们的爬虫程序可以源源不断地爬取数据而不需要中止程序的执行。

• 高优先级URL队列

Redis的数据类型为set。

什么是高优先级URL队列？其实它就是用来保存列表url的。

那么什么是列表url呢？

说白了就是一个列表中含有多个商品，以京东为列，我们打开一个手机列表为例：

该地址中包含的不是一个具体商品的url，而是包含了多个我们需要爬取的数据（手机商品）的列表，通过对每个高级url的解析，我们可以获取到非常多的具体商品url，而具体的商品url，就是低优先url，其会保存到低优先级URL队列中。

那么以这个系统为例，保存的数据类似如下：

• 低优先级URL队列

Redis的数据类型为set。

低优先级URL其实就是具体某个商品的URL，如下面一个手机商品：

通过下载该url的数据，并对其进行解析，就能够获取到我们想要的数据。

那么以这个系统为例，保存的数据类似如下：

4.2 URL调度器

所谓url调度器，其实说白了就是url仓库java代码的调度策略，不过因为其核心在于调度，所以将其放到URL调度器中来进行说明，目前其调度基于以下接口开发：

<span class="hljs-comment"&gt;/**
 * 获取url的方法
 * 从仓库中获取url（优先获取高优先级的url，如果没有，再获取低优先级的url）
 * <span class="hljs-doctag"&gt;@return
 */
<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;public String <span class="hljs-title"&gt;poll<span class="hljs-params"&gt;();

<span class="hljs-comment"&gt;/**
 * 向高优先级列表中添加商品列表url
 * <span class="hljs-doctag"&gt;@param highUrl
 */
<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;public <span class="hljs-keyword"&gt;void <span class="hljs-title"&gt;offerHigher<span class="hljs-params"&gt;(String highUrl);

<span class="hljs-comment"&gt;/**
 * 向低优先级列表中添加商品url
 * <span class="hljs-doctag"&gt;@param lowUrl
 */
<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;public <span class="hljs-keyword"&gt;void <span class="hljs-title"&gt;offerLower<span class="hljs-params"&gt;(String lowUrl);

}

其基于Redis作为URL仓库的实现如下：

<span class="hljs-comment"&gt;/**
 * 构造方法
 */
<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;public <span class="hljs-title"&gt;RandomRedisRepositoryImpl<span class="hljs-params"&gt;() {
    init();
}

<span class="hljs-comment"&gt;/**
 * 初始化方法，初始化时，先将redis中存在的高低优先级url队列全部删除
 * 否则上一次url队列中的url没有消耗完时，再停止启动跑下一次，就会导致url仓库中有重复的url
 */
<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;public <span class="hljs-keyword"&gt;void <span class="hljs-title"&gt;init<span class="hljs-params"&gt;() {
    Jedis jedis = JedisUtil.getJedis();
    Set<String> domains = jedis.smembers(SpiderConstants.SPIDER_WEBSITE_DOMAINS_KEY);
    String higherUrlKey;
    String lowerUrlKey;
    <span class="hljs-keyword"&gt;for(String domain : domains) {
        higherUrlKey = domain + SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX;
        lowerUrlKey = domain + SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX;
        jedis.del(higherUrlKey,lowerUrlKey);
    }
    JedisUtil.returnJedis(jedis);
}

<span class="hljs-comment"&gt;/**
 * 从队列中获取url，目前的策略是：
 *      1.先从高优先级url队列中获取
 *      2.再从低优先级url队列中获取
 *  对应我们的实际场景，应该是先解析完列表url再解析商品url
 *  但是需要注意的是，在分布式多线程的环境下，肯定是不能完全保证的，因为在某个时刻高优先级url队列中
 *  的url消耗完了，但实际上程序还在解析下一个高优先级url，此时，其它线程去获取高优先级队列url肯定获取不到
 *  这时就会去获取低优先级队列中的url，在实际考虑分析时，这点尤其需要注意
 * <span class="hljs-doctag"&gt;@return
 */
<span class="hljs-meta"&gt;@Override
<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;public String <span class="hljs-title"&gt;poll<span class="hljs-params"&gt;() {
    <span class="hljs-comment"&gt;// 从set中随机获取一个顶级域名
    Jedis jedis = JedisUtil.getJedis();
    String randomDomain = jedis.srandmember(SpiderConstants.SPIDER_WEBSITE_DOMAINS_KEY);    <span class="hljs-comment"&gt;// jd.com
    String key = randomDomain + SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX;                <span class="hljs-comment"&gt;// jd.com.higher
    String url = jedis.lpop(key);
    <span class="hljs-keyword"&gt;if(url == <span class="hljs-keyword"&gt;null) {   <span class="hljs-comment"&gt;// 如果为null，则从低优先级中获取
        key = randomDomain + SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX;    <span class="hljs-comment"&gt;// jd.com.lower
        url = jedis.lpop(key);
    }
    JedisUtil.returnJedis(jedis);
    <span class="hljs-keyword"&gt;return url;
}

<span class="hljs-comment"&gt;/**
 * 向高优先级url队列中添加url
 * <span class="hljs-doctag"&gt;@param highUrl
 */
<span class="hljs-meta"&gt;@Override
<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;public <span class="hljs-keyword"&gt;void <span class="hljs-title"&gt;offerHigher<span class="hljs-params"&gt;(String highUrl) {
    offerUrl(highUrl,SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX);
}

<span class="hljs-comment"&gt;/**
 * 向低优先url队列中添加url
 * <span class="hljs-doctag"&gt;@param lowUrl
 */
<span class="hljs-meta"&gt;@Override
<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;public <span class="hljs-keyword"&gt;void <span class="hljs-title"&gt;offerLower<span class="hljs-params"&gt;(String lowUrl) {
    offerUrl(lowUrl,SpiderConstants.SPIDER_DOMAIN_LOWER_SUFFIX);
}

<span class="hljs-comment"&gt;/**
 * 添加url的通用方法，通过offerHigher和offerLower抽象而来
 * <span class="hljs-doctag"&gt;@param url   需要添加的url
 * <span class="hljs-doctag"&gt;@param urlTypeSuffix  url类型后缀.higher或.lower
 */
<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;public <span class="hljs-keyword"&gt;void <span class="hljs-title"&gt;offerUrl<span class="hljs-params"&gt;(String url,String urlTypeSuffix) {
    Jedis jedis = JedisUtil.getJedis();
    String domain = SpiderUtil.getTopDomain(url);   <span class="hljs-comment"&gt;// 获取url对应的顶级域名，如jd.com
    String key = domain + urlTypeSuffix;            <span class="hljs-comment"&gt;// 拼接url队列的key，如jd.com.higher
    jedis.lpush(key,url);                          <span class="hljs-comment"&gt;// 向url队列中添加url
    JedisUtil.returnJedis(jedis);
}

}

通过代码分析也是可以知道，其核心就在如何调度url仓库（Redis）中的url。

4.3 URL定时器

一段时间后，高优先级URL队列和低优先URL队列中的url都会被消费完，为了让程序可以继续爬取数据，同时减少人为的干预，可以预先在Redis中插入种子url，之后定时让URL定时器从种子url中取出url定存放到高优先级URL队列中，以此达到程序定时不间断爬取数据的目的。

url消费完毕后，是否需要循环不断爬取数据根据个人业务需求而不同，因此这一步不是必需的，只是也提供了这样的操作。因为事实上，我们需要爬取的数据也是每隔一段时间就会更新的，如果希望我们爬取的数据也跟着定时更新，那么这时定时器就有非常重要的作用了。不过需要注意的是，一旦决定需要循环重复爬取数据，则在设计存储器实现时需要考虑重复数据的问题，即重复数据应该是更新操作，目前在我设计的存储器不包括这个功能，有兴趣的朋友可以自己实现，只需要在插入数据前判断数据库中是否存在该数据即可。

另外需要注意的一点是，URL定时器是一个独立的进程，需要单独启动。

定时器基于Quartz实现，下面是其job的代码：

<span class="hljs-comment"&gt;// log4j日志记录
<span class="hljs-keyword"&gt;private Logger logger = LoggerFactory.getLogger(UrlJob.class);

<span class="hljs-meta"&gt;@Override
<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;public <span class="hljs-keyword"&gt;void <span class="hljs-title"&gt;execute<span class="hljs-params"&gt;(JobExecutionContext context) <span class="hljs-keyword"&gt;throws JobExecutionException {
    <span class="hljs-comment"&gt;/**
     * 1.从指定url种子仓库获取种子url
     * 2.将种子url添加进高优先级列表
     */
    Jedis jedis = JedisUtil.getJedis();
    Set<String> seedUrls = jedis.smembers(SpiderConstants.SPIDER_SEED_URLS_KEY);  <span class="hljs-comment"&gt;// spider.seed.urls Redis数据类型为set，防止重复添加种子url
    <span class="hljs-keyword"&gt;for(String seedUrl : seedUrls) {
        String domain = SpiderUtil.getTopDomain(seedUrl);   <span class="hljs-comment"&gt;// 种子url的顶级域名
        jedis.sadd(domain + SpiderConstants.SPIDER_DOMAIN_HIGHER_SUFFIX,seedUrl);
        logger.info(<span class="hljs-string"&gt;"获取种子:{}",seedUrl);
    }
    JedisUtil.returnJedis(jedis);

}

调度器的实现如下：

<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;public <span class="hljs-title"&gt;UrlJobScheduler<span class="hljs-params"&gt;() {
    init();
}

<span class="hljs-comment"&gt;/**
 * 初始化调度器
 */
<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;public <span class="hljs-keyword"&gt;void <span class="hljs-title"&gt;init<span class="hljs-params"&gt;() {
    <span class="hljs-keyword"&gt;try {
        Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler();

        <span class="hljs-comment"&gt;// 如果没有以下start方法的执行，则是不会开启任务的调度
        scheduler.start();

        String name = <span class="hljs-string"&gt;"URL_SCHEDULER_JOB";
        String group = <span class="hljs-string"&gt;"URL_SCHEDULER_JOB_GROUP";
        JobDetail jobDetail = <span class="hljs-keyword"&gt;new JobDetail(name,group,UrlJob.class);
        String cronExpression = <span class="hljs-string"&gt;"0 10 1 * * ?";
        Trigger trigger = <span class="hljs-keyword"&gt;new CronTrigger(name,cronExpression);

        <span class="hljs-comment"&gt;// 调度任务
        scheduler.scheduleJob(jobDetail,trigger);

    } <span class="hljs-keyword"&gt;catch (SchedulerException e) {
        e.printStackTrace();
    } <span class="hljs-keyword"&gt;catch (ParseException e) {
        e.printStackTrace();
    }
}

<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;public <span class="hljs-keyword"&gt;static <span class="hljs-keyword"&gt;void <span class="hljs-title"&gt;main<span class="hljs-params"&gt;(String[] args) {
    UrlJobScheduler urlJobScheduler = <span class="hljs-keyword"&gt;new UrlJobScheduler();
    urlJobScheduler.start();
}

<span class="hljs-comment"&gt;/**
 * 定时调度任务
 * 因为我们每天要定时从指定的仓库中获取种子url，并存放到高优先级的url列表中
 * 所以是一个不间断的程序，所以不能停止
 */
<span class="hljs-function"&gt;<span class="hljs-keyword"&gt;private <span class="hljs-keyword"&gt;void <span class="hljs-title"&gt;start<span class="hljs-params"&gt;() {
    <span class="hljs-keyword"&gt;while (<span class="hljs-keyword"&gt;true) {

    }
}

}

5 监控报警系统

监控报警系统的加入主要是为了让使用者可以主动发现节点宕机，而不是被动地发现，因为实际中爬虫程序可能是持续不断运行的，并且我们会在多个节点上部署我们的爬虫程序，因此很有必要对节点进行监控，并且在节点出现问题时可以及时发现并修正，需要注意的是，监控报警系统是一个独立的进程，需要单独启动。

5.1 基本原理

首先需要先在zookeeper中创建一个/ispider节点：

 currentNodes = curator.getChildren().usingWatcher( previousNodesSet = new HashSet<>(previousNodes);
         previousNodes.size()) { 

 select count(*) from phone;
+----------+
| count(*) |
+----------+
|    12052 |
+----------+
1 row in set
mysql> select count() from phone where source='jd.com';

+----------+

| count() |

+----------+

|     9578 |

+----------+

1 row in set
mysql> select count() from phone where source='suning

.com';

+----------+

| count() |

+----------+

|     2474 |

+----------+

1 row in set

=> <span class="hljs-number">12348