nutch fetcher详解

fetcher 是生产者和消费者的模式，生产者是QueueFeeder 不断的读取文件，消费者是
FetcherThread 不断的抓取网址 map是输入是crawl/segments/具体的segment/crawl_generate

QueueFeeder



QueueFeeder 是一个线程类，主要做了一下事情
1 根据配置属性fetcher.timelimit.mins ，这个值表示将来的一个时间，如果这个值是-1 则表明这个线程没有过期时间，如果不是-1 则每次循环都判断是否过期，如果过期了就不往下操作，

2 QueueFeeder有一个FetchItemQueues 属性 是要抓取的url的容器，查看他的容量是不是超过设置的属性size，如果超过则sleep（1000），然后重新循环
3 如果还有数据就不断的读取数据放到FetchItemQueues 里面去，容器中的队列是按ip或者域名分组，是由fetcher.threads.per.host.by.ip 这个属性决定的

FetcherThread也是一个线程类



1 FetcherThread 从fetchQueues取出一条记录，这个记录不为空往下执行
2 根据url找到协议解析的插件
3 根据2得到的协议插件如果是http协议则通过socket 得到roboots.txt文件，这个是协议文件，规定哪些url是可以访问的哪些是不可以访问的 ，如果没有这个文件，则是没有协议的
4 查看当前的url是否允许访问，如果不允许，则写入(写入参考步骤7)
5 查看roboots.txt sCrawl-delay 这个属性是否有设置，如果有，和配置fetcher.max.crawl.delay属性比较，如果不超过这个值，往下执行，否则进入下一次数据的处理
6 更加3得到的插件发送请求。根据配置 protocol.plugin.check.robots查看是不是要坚持roboots协议，是，检查，不允许访问返回。如果roboots.txt有设置crawlDelayd大于0取这个crawlDelayd，否则取fetcher.server.delay配置的值，以秒为单位，如果设置protocol.plugin.check.robots 为true，并且设置fetcher.max.crawl.delay的值大于0，则比较fetcher.max.crawl.delay 值和crawlDelayd的大小，小于，则返回。如果protocol.plugin.check.blocking 为true，就会看是否delay，发送socket请求得到返回结果

7根据返回状态，解析结果，如果为ProtocolStatus.SUCCESS ，请求成功 ，设置CrawlDatum的状态为CrawlDatum.STATUS_FETCH_SUCCESS，设置fetchtime为当前时间，设置CrawlDatum 的元数据Nutch.WRITABLE_PROTO_STATUS_KEY为ProtocolStatus，设置content的segment  metadata.set(Nutch.SEGMENT_NAME_KEY, segmentName);设置content的元数据的分数为CrawlDatum 元数据的分数，使用parseUtil解析content数据这个类会根据content的contentType，查找解析插件HtmlParser，这个会利用多线程Callable和FutureTask，返回解析结果，设置content的元数据的Nutch.FETCH_STATUS_KEY fetch—status，
写入数据key:Text  是url，value：NutchWritable ，new NutchWritable(datum)。在reduce的阶段会被写入特定的目录，如果设置fetcher.store.content
为true，写入content，key:Text  是url，value：NutchWritable ，new NutchWritable(content)，设置Parse的元数据的 segment，fetchtime，签名，设置crawldatum的签名，设置parse的元数据的分数，为content的元数据的分数，key:Text  是url，value：NutchWritable : new NutchWritable(
                    new ParseImpl(new ParseText(parse.getText()),
                                  parse.getData(), parse.isCanonical()))


9 没有显示的reduce，使用默认的reducer的,输出的目录为crawl_home/segments/当前的segment/ 下面，FetcherOutputFormat会根据不同的类型写入不同的目录，
代码如下 CrawlDatum会写在当前segment的crawl_fetch目录，Content会写在content目录，
  if (w instanceof CrawlDatum)
            fetchOut.append(key, w);
          else if (w instanceof Content)
            contentOut.append(key, w);
          else if (w instanceof Parse)
            parseOut.write(key, (Parse)w);
        }

Parse 又会根据ParseOutputFormat 写到不同目录
parse_text 目录写入key:Text url value:ParseText

得到Parse 的签名，如果转换成二进制不为空，则在crawl_parse写入，key:Text为url,value：CrawlDatum，如果db.parsemeta.to.crawldb配置的值需要值，从ParseData的元数据取出也写入 crawl_parse目录，写入
CrawlDatum newDatum = new CrawlDatum();
                newDatum.setStatus(CrawlDatum.STATUS_LINKED);
                if (reprUrl != null && !reprUrl.equals(newUrl)) {
                  newDatum.getMetaData().put(Nutch.WRITABLE_REPR_URL_KEY,
                                             new Text(reprUrl));
                }
                crawlOut.append(new Text(newUrl), newDatum);

根据外链计算分数，写入
  if (adjust != null) crawlOut.append(key, adjust);

根据db.max.outlinks.per.page这个配置，允许的最大的外链调整，进去外链调整，并对外链做normalize和filter，写入parse_data 目录key:Text url value:ParseData

写入
if (!parse.isCanonical()) {
            CrawlDatum datum = new CrawlDatum();
            datum.setStatus(CrawlDatum.STATUS_FETCH_SUCCESS);
            String timeString = parse.getData().getContentMeta().get(Nutch.FETCH_TIME_KEY);
            try {
              datum.setFetchTime(Long.parseLong(timeString));
            } catch (Exception e) {
              LOG.warn("Can't read fetch time for: " + key);
              datum.setFetchTime(System.currentTimeMillis());
            }
            crawlOut.append(key, datum);
          }