Fetcher - Nutch 22 Mar 2014

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);
    // 这里使用了inner class来定义相应的RecordWriter
    return new RecordWriter<Text, NutchWritable>(){
        private MapFile.Writer contentOut;
        private RecordWriter<Text, Parse> parseOut;
   
   
        {
            // 这里看如果Fetcher定义了输出内容，就生成相应的Content输出抽象
            if (Fetcher.isStoringContent(job)) {
                 contentOut = new MapFile.Writer(job, fs, content.toString(), Text.class, Content.class, compType, progress);
            }
   
            // 如果Fetcher对抓取的内容进行了解析，这里就定义相应的解析输出抽象
            // 注意这里使用了ParseOutputFormat的getReocrdWriter,主要是解析网页，抽取其外链接
            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);
            }
        }
    };
}

// 这个类继承自Thread，是用一个单独的线程来做的
private static class QueueFeeder extends Thread {
    private RecordReader<Text, CrawlDatum> reader; // 这里是InputFormat产生的ReocrdReader，用于读取Generate的产生的数据
    private FetchItemQueues queues;                // 这是生产者与消费者所使用的共享队列，这个对列是分层的，分一层对应一个host
    private int size;                              // 队列的大小
    private long timelimit = -1;                   // 这是一个过滤机制的策略，用于过滤所有的FetchItem
      
    // 构造方法
    public QueueFeeder(RecordReader<Text, CrawlDatum> reader,
                       FetchItemQueues queues, int size) {
        this.reader = reader;
        this.queues = queues;
        this.size = size;
        this.setDaemon(true);
        this.setName("QueueFeeder");
    }
   
    public void setTimeLimit(long tl) {
        timelimit = tl;
    }
   
    // 函数的run方法
    public void run() {
        boolean hasMore = true; // while的循环条件
        int cnt = 0;
        int timelimitcount = 0;
        while (hasMore) {
            // 这里判断是否设置了这个过滤机制，如果设置了，判断相前时间是否大于这个timelimit，如果大于timelimit，过滤所有的FetchItem
            if (System.currentTimeMillis() >= timelimit && timelimit != -1) {
                // enough .. lets's simply
                // read all the entries from the input without processing them
                try {
                    // 读出<key,value>对，过滤之
                    Text url = new Text();
                    CrawlDatum datum = new CrawlDatum();
                    hasMore = reader.next(url, datum);
                    timelimitcount++;
                } catch (IOException e) {
                    LOG.fatal("QueueFeeder error reading input, record " + cnt, e);
                    return;
                }
                continue; // 过滤之
            }
            int feed = size - queues.getTotalSize();
            // 判断剩余的队列空间是否为0
            if (feed <= 0) {
                // queues are full - spin-wait until they have some free space
                try {
                    // 休息1秒种
                    Thread.sleep(1000);
                } catch (Exception e) {};
                continue;
            } else {
                LOG.debug("-feeding " + feed + " input urls ...");
                // 如果队列还有空间(feed>0)并且recordRedder中还有数据(hasMore)
                while (feed > 0 && hasMore) {
                    try {
                        Text url = new Text();
                        CrawlDatum datum = new CrawlDatum();
                        // 读出<key,value>
                        hasMore = reader.next(url, datum);
                        if (hasMore) {       // 判断是否成功读出数据
                            queues.addFetchItem(url, datum); // 放入对列，这个队列应该是thread-safe的，下面我们可以看到
                            cnt++;       // 统计总数
                            feed--;      // 剩余队列空间减1
                        }
                    } catch (IOException e) {
                        LOG.fatal("QueueFeeder error reading input, record " + cnt, e);
                        return;
                    }
                }
            }
        }
        LOG.info("QueueFeeder finished: total " + cnt + " records + hit by time limit :" + timelimitcount);
    }
}

//从注释中我们可以看到，队列ID是由protocol+hotname或者是protocol+IP组成的
/** Create an item. Queue id will be created based on byIP
 * argument, either as a protocol + hostname pair, or protocol + IP
 * address pair.
 */
public static FetchItem create(Text url, CrawlDatum datum, boolean byIP) {
    String queueID;
    URL u = null;
    try {
        u = new URL(url.toString()); // 得到其URL
    } catch (Exception e) {
        LOG.warn("Cannot parse url: " + url, e);
        return null;
    }
    // 得到协议号
    String proto = .getProtocol().toLowerCase();
    String host;
    if (byIP) {
        // 如果是基于IP的，那得到其IP地址
        try {
            InetAddress addr = InetAddress.getByName(u.getHost());
            host = addr.getHostAddress();
        } catch (UnknownHostException e) {
            // unable to resolve it, so don't fall back to host name
            LOG.warn("Unable to resolve: " + u.getHost() + ", skipping.");
            return null;
        }
    } else {
        // 否则得到Hostname
        host = u.getHost();
        if (host == null) {
            LOG.warn("Unknown host for url: " + url + ", skipping.");
            return null;
        }
        hosthost = host.toLowerCase(); // 统一变小写
    }
    // 得成相应的队列ID号，放入FetchItemQueue中
    queueID = proto + "://" + host;
    return new FetchItem(url, u, datum, queueID);
}

FetchQueue =>
   
{
    // 用于收集相同QueueID的FetchItem, 这里使用的是线程安全的对象
    List<FetchItem> queue = Collections.synchronizedList(new LinkedList<FetchItem>());
    // 用于收集正在抓取的FetchItem
    Set<FetchItem> inProgress = Collections.synchronizedSet(new HashSet<FetchItem>());
    // 用于存储下一个FetchItem的抓取时候，如果没有到达这个时间，就返回给FetchThread为null
    AtomicLong nextFetchTime = new AtomicLong();
    // 存储抓取的出错次数
   
    AtomicInteger exceptionCounter = new AtomicInteger();
    // 存储FetchItem抓取间隔，这个配置只有当同时抓取最大线程数为1时才有用
    long crawlDelay;
    // 存储最小的抓取间隔，这个配置当同时抓取的最大线程数大于1时有用
    long minCrawlDelay;
   
    // 同时抓取的最大线程数
    int maxThreads;
    Configuration conf;
}

public FetchItem getFetchItem() {
    // 当正在抓取的FetchItem数大于同时抓取的线程数时，返回null,这是一个politness策略
    // 就是对于同一个网站，不能同时有大于maxThreads个线程在抓取，不然人家网站会认为你是在攻击它
    if (inProgress.size() >= maxThreads) return null;
    long now = System.currentTimeMillis();
    // 计算两个抓取的间隔时间，如果没有到达这个时间，就返回null，这个是保证不会有多个线程同时在抓取一个网站
    if (nextFetchTime.get() > now) return null;
    FetchItem it = null;
    // 判断队列是否为空
    if (queue.size() == 0) return null;
    try {
        // 从准备队列中移除一个FetchItem,把其放到inProcess集合中
        it = queue.remove(0);
        inProgress.add(it);
    } catch (Exception e) {
        LOG.error("Cannot remove FetchItem from queue or cannot add it to inProgress queue", e);
    }
    return it;
}

nextFetchTime = endTime + (maxThread > 1) ? minCrawlDelay : crawlDelay)

// 对配置进行检测，看一些必要的配置是否已经配置了，如http.agent.name等参数  
   
checkConfiguration();
   
// 记录fetch的开始时间
   
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);
   
}
   
// 这里对抓取的时候进行限制,在FetchItemQueue中会用到这个参数
   
// 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);
   
}
   
// 生成一个Nutch的Map-Reduce配置
   
JobConf job = new NutchJob(getConf());
   
job.setJobName("fetch " + segment);
   
// 配置抓取线程数;
   
job.setInt("fetcher.threads.fetch", threads);
   
job.set(Nutch.SEGMENT_NAME_KEY, segment.getName());
   
// 配置是否对抓取的内容进行解析
   
job.setBoolean("fetcher.parse", parsing);
   
// for politeness, don't permit parallel execution of a single task
   
job.setSpeculativeExecution(false);
   
// 配置输入的路径名
   
FileInputFormat.addInputPath(job, new Path(segment, CrawlDatum.GENERATE_DIR_NAME));
   
// 配置输入的文件格式, 这里类继承自SequenceFileInputFormat
   
// 它主要是覆盖了其getSplits方法，其作用是不对文件进行切分，以文件数量作为splits的依据
   
// 就是有几个文件，就有几个Map操作
   
job.setInputFormat(InputFormat.class);
   
// 配置Map操作的类
   
job.setMapRunnerClass(Fetcher.class);
   
// 配置输出路径
   
FileOutputFormat.setOutputPath(job, segment);
   
// 这里配置输出文件方法，这个类在前面已经分析过
   
job.setOutputFormat(FetcherOutputFormat.class);
   
// 配置输出<key,value>类型
   
job.setOutputKeyClass(Text.class);
   
job.setOutputValueClass(NutchWritable.class);
   
JobClient.runJob(job);

// 生成生产者，用于读取Generate出来的CrawlDatum，把它们放到共享队列中
   
feeder = new QueueFeeder(input, fetchQueues, threadCount * 50);
   
//feeder.setPriority((Thread.MAX_PRIORITY + Thread.NORM_PRIORITY) /2);
   
// the value of the time limit is either -1 or the time where it should finish
   
long timelimit = getConf().getLong("fetcher.timelimit", -1);
   
if (timelimit != -1) feeder.setTimeLimit(timelimit);
   
feeder.start();
   
// set non-blocking & no-robots mode for HTTP protocol plugins.
   
getConf().setBoolean(Protocol.CHECK_BLOCKING, false);
   
getConf().setBoolean(Protocol.CHECK_ROBOTS, false);
   
   
// 启动消费者线程
for (int i = 0; i < threadCount; i++) {       // spawn threads
    new FetcherThread(getConf()).start();
}
   
// select a timeout that avoids a task timeout
long timeout = getConf().getInt("mapred.task.timeout", 10*60*1000)/2;
   
// 这里用一个循环来等待线程结束
do {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {}
   
    // 这个函数是得到相前线程的抓取状态，如抓取了多少网页，多少网页抓取失败，抓取速度是多少
    reportStatus();
    LOG.info("-activeThreads=" + activeThreads + ", spinWaiting=" + spinWaiting.get()
        + ", fetchQueues.totalSize=" + fetchQueues.getTotalSize());
    // 输出抓取队列中的信息
    if (!feeder.isAlive() && fetchQueues.getTotalSize() < 5) {
    fetchQueues.dump();
    }
   
   
    // 查看timelimit的值，这里只要返回的hitByTimeLimit不为0, checkTimelimit方法会清空抓取队列中的所有数据
    // check timelimit
    if (!feeder.isAlive()) {
    int hitByTimeLimit = fetchQueues.checkTimelimit();
    if (hitByTimeLimit != 0) reporter.incrCounter("FetcherStatus",
                    "hitByTimeLimit", hitByTimeLimit);
    }
   
   
    // 查看抓取线程是否超时，如果超时，就退出等待
    // some requests seem to hang, despite all intentions
    if ((System.currentTimeMillis() - lastRequestStart.get()) > timeout) {
        if(LOG.isWarnEnabled()) {
            LOG.warn("Aborting with " + activeThreads + " hung threads.");
        }
        return;
    }
   
} while (activeThreads.get() > 0);
   
LOG.info("-activeThreads=" + activeThreads);