nutch源码阅读(9)-Fetch
...................................................... 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);          }        }      };  }  

?

?

?

?