nutch源码阅读(9)-Fetch

Fetcher这个模块在Nutch中有单独一个包在实现，在org.apache.nutch.fetcher，其中有Fetcher.java, FetcherOutput 和FetcherOutputFormat来组成，看上去很简单，但其中使用到了多线程，多线程的生产者与消费者模型，MapReduce的多路径输出等方法。

下面我们来看一下Fetcher的注释，从中我们可以得到很多有用的信息。
首先，这是一种基于队列的fetcher方法，它使用了一种经典的线程模型，生产者(a-QueueFeeder)与消费者(many-FetcherThread)模型，注意，这里有多个消费者。生产者从Generate产生的fetchlists中分类得到一批FetchItemQueue，每一个FetchItmeQueue都是由一类相同host的FetchItem组成，这些FetchItem是用来描述被抓取的对象。当一个FetchItem从FetchItemQueue中取出后，QueueFeeder这个生产者会不断的向队列中加入新的FetchItem，直到这个队列满了为止或者已经没有fetchlist可读取，当队列中的所有FetchItem都被抓取完成后，所有抓取线程都会退出运行。每一个FetchItemQueue都有一套自己的抓取策略，如最大的并行抓取个数，两次抓取的间隔等，如果当FetcherThread向队列申请一个FetchItem时，FetchItemQueue发现当前的FetchItem没有满足抓取策略，那这里它就会返回null，表达当前FetchItem还没有准备好被抓取。如果这些所有FetchItem都没有准备好被抓取，那这时FetchThread就会进入等待状态，直到条件满足被促发或者是等待超时，它会认为任务已经被挂起，这时FetchThread会自动退出。

 

 

..................
..................
..................

 for (i = 0; i < depth; i++) {             // generate new segment
      Path[] segs = generator.generate(crawlDb, segments, -1, topN, System
          .currentTimeMillis());
      if (segs == null) {
        LOG.info("Stopping at depth=" + i + " - no more URLs to fetch.");
        break;
      }
      //通过之前生成的segments抓取
      fetcher.fetch(segs[0], threads);  // fetch it
      if (!Fetcher.isParsing(job)) {
        parseSegment.parse(segs[0]);    // parse it, if needed
      }
      crawlDbTool.update(crawlDb, segs, true, true); // update crawldb
    }

..................
..................
..................

 

 

 

//threads默认通过fetcher.threads.fetch设置，也可通过参数传递
 public void fetch(Path segment, int threads)
    throws IOException {
    //设置agentName
    //在http.agent.name中设置，会检查在http.robots.agents中是否存在
    checkConfiguration();
    //记录开始时间，以及segment路径
    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    long start = System.currentTimeMillis();
    if (LOG.isInfoEnabled()) {
      LOG.info("Fetcher: starting at " + sdf.format(start));
      LOG.info("Fetcher: segment: " + segment);
    }

    // set the actual time for the timelimit relative
    // to the beginning of the whole job and not of a specific task
    // otherwise it keeps trying again if a task fails
    long timelimit = getConf().getLong("fetcher.timelimit.mins", -1);
    if (timelimit != -1) {
      timelimit = System.currentTimeMillis() + (timelimit * 60 * 1000);
      LOG.info("Fetcher Timelimit set for : " + timelimit);
      getConf().setLong("fetcher.timelimit", timelimit);
    }

    // Set the time limit after which the throughput threshold feature is enabled
    timelimit = getConf().getLong("fetcher.throughput.threshold.check.after", 10);
    timelimit = System.currentTimeMillis() + (timelimit * 60 * 1000);
    getConf().setLong("fetcher.throughput.threshold.check.after", timelimit);

    int maxOutlinkDepth = getConf().getInt("fetcher.follow.outlinks.depth", -1);
    if (maxOutlinkDepth > 0) {
      LOG.info("Fetcher: following outlinks up to depth: " + Integer.toString(maxOutlinkDepth));

      int maxOutlinkDepthNumLinks = getConf().getInt("fetcher.follow.outlinks.num.links", 4);
      int outlinksDepthDivisor = getConf().getInt("fetcher.follow.outlinks.depth.divisor", 2);

      int totalOutlinksToFollow = 0;
      for (int i = 0; i < maxOutlinkDepth; i++) {
        totalOutlinksToFollow += (int)Math.floor(outlinksDepthDivisor / (i + 1) * maxOutlinkDepthNumLinks);
      }

      LOG.info("Fetcher: maximum outlinks to follow: " + Integer.toString(totalOutlinksToFollow));
    }

    JobConf job = new NutchJob(getConf());
    job.setJobName("fetch " + segment);
    //配置线程数
    job.setInt("fetcher.threads.fetch", threads);
    job.set(Nutch.SEGMENT_NAME_KEY, segment.getName());

    // for politeness, don't permit parallel execution of a single task
    job.setSpeculativeExecution(false);
    //配置输入
    FileInputFormat.addInputPath(job, new Path(segment, CrawlDatum.GENERATE_DIR_NAME));
    job.setInputFormat(InputFormat.class);
    //配置MapRunner
    job.setMapRunnerClass(Fetcher.class);
    //配置输出
    FileOutputFormat.setOutputPath(job, segment);
    job.setOutputFormat(FetcherOutputFormat.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(NutchWritable.class);

    JobClient.runJob(job);

    long end = System.currentTimeMillis();
    LOG.info("Fetcher: finished at " + sdf.format(end) + ", elapsed: " + TimingUtil.elapsedTime(start, end));
  }

 

 

public RecordWriter<Text, NutchWritable> getRecordWriter(final FileSystem fs,
                                      final JobConf job,
                                      final String name,
                                      final Progressable progress) throws IOException {
    
    //定义输出目录
    Path out = FileOutputFormat.getOutputPath(job);
    //定义抓取的输出目录
    final Path fetch =
      new Path(new Path(out, CrawlDatum.FETCH_DIR_NAME), name);
    //定义抓取内容的输出目录
    final Path content =
      new Path(new Path(out, Content.DIR_NAME), name);
    //定义压缩类型
    final CompressionType compType = SequenceFileOutputFormat.getOutputCompressionType(job);
    //定义输出对象
    final MapFile.Writer fetchOut =
      new MapFile.Writer(job, fs, fetch.toString(), Text.class, CrawlDatum.class,
          compType, progress);
    
    return new RecordWriter<Text, NutchWritable>() {
        private MapFile.Writer contentOut;
        private RecordWriter<Text, Parse> parseOut;

        {
          //如果配置"fetcher.store.content"为true，则生成content
          if (Fetcher.isStoringContent(job)) {
            contentOut = new MapFile.Writer(job, fs, content.toString(),
                                            Text.class, Content.class,
                                            compType, progress);
          }
          //如果配置"fetcher.parse"为true，而抽取出外连接
          if (Fetcher.isParsing(job)) {
            parseOut = new ParseOutputFormat().getRecordWriter(fs, job, name, progress);
          }
        }

        public void write(Text key, NutchWritable value)
          throws IOException {

          Writable w = value.get();
          // 对对象类型进行判断，调用相应的抽象输出，写到不同的文件中去 
          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);
        }

        public void close(Reporter reporter) throws IOException {
          fetchOut.close();
          if (contentOut != null) {
            contentOut.close();
          }
          if (parseOut != null) {
            parseOut.close(reporter);
          }
        }

      };

  }