Method run

src/java/org/apache/nutch/fetcher/FetcherReducer.java:815–954

(Context context)

  }

  @Override
  public void run(Context context) throws IOException, InterruptedException {
    Configuration conf = context.getConfiguration();
    this.fetchQueues = new FetchItemQueues(conf);
    int threadCount = conf.getInt("fetcher.threads.fetch", 10);
    parse = conf.getBoolean(FetcherJob.PARSE_KEY, false);
    storingContent = conf.getBoolean("fetcher.store.content", true);
    if (parse) {
      boolean sitemap = conf.getBoolean(FetcherJob.SITEMAP, false);

      if (sitemap) {
        skipTruncated = false;
      } else {
        skipTruncated = conf.getBoolean(ParserJob.SKIP_TRUNCATED, true);
      }
      parseUtil = new ParseUtil(conf);
    }
    LOG.info("Fetcher: threads: " + threadCount);

    int maxFeedPerThread = conf.getInt("fetcher.queue.depth.multiplier", 50);
    feeder = new QueueFeeder(context, fetchQueues, threadCount
        * maxFeedPerThread);
    feeder.start();

    for (int i = 0; i < threadCount; i++) { // spawn threads
      FetcherThread ft = new FetcherThread(context, i);
      fetcherThreads.add(ft);
      ft.start();
    }
    // select a timeout that avoids a task timeout
    final long timeout = conf.getInt("mapreduce.task.timeout", 10 * 60 * 1000) / 2;

    // Used for threshold check, holds pages and bytes processed in the last sec
    float pagesLastSec;
    int bytesLastSec;

    int throughputThresholdCurrentSequence = 0;

    int throughputThresholdPages = conf.getInt(
        "fetcher.throughput.threshold.pages", -1);
    if (LOG.isInfoEnabled()) {
      LOG.info("Fetcher: throughput threshold: " + throughputThresholdPages);
    }
    int throughputThresholdSequence = conf.getInt(
        "fetcher.throughput.threshold.sequence", 5);
    if (LOG.isInfoEnabled()) {
      LOG.info("Fetcher: throughput threshold sequence: "
          + throughputThresholdSequence);
    }
    long throughputThresholdTimeLimit = System.currentTimeMillis() + (conf.getLong(
        "fetcher.throughput.threshold.check.after", 5) * 60 * 1000);

    do { // wait for threads to exit
      pagesLastSec = pages.get();
      bytesLastSec = (int) bytes.get();
      final int secondsToSleep = 5;
      try {
        Thread.sleep(secondsToSleep * 1000);