Heritrix的架构

10.2  Heritrix的架构

在上一节中，详细介绍了Heritrix的使用入门。读者通过上一节的介绍，应该已经能够使用Heritrix来进行简单的网页抓取了。那么，Heritrix的内容究竟是如何工作的呢？它的设计方面有什么突出之处？

本节就将介绍Heritrix的几个主要组件，以此让读者了解其主要架构和工作方式。为后续的扩展Heritrix做一些铺垫。

10.2.1  抓取任务CrawlOrder

之所以选择从CrawlOrder这个类说起，是因为它是整个抓取工作的起点。在上一节中已经说过，一次抓取任务包括许多的属性，建立一个任务的方式有很多种，最简单的一种就是根据默认的order.xml来配置。在内存中，order使用CrawlOrder这个类来进行表示。看一下API文档中CrawlOrder的继承关系图，如图10-52所示。

图10-52  CrawlOrder类的继承关系图

从继承关系图中可以看到，CrawlOrder继承自一系列的与属性设置相关的基类。另外，它的最顶层基类是javax.management.Attribute，这是一个JMX中的类，它可以动态的反映出Java容器内某个MBean的属性变化。关于这一部分的内容不是我们所要讨论的重点，只需知道，CrawlOrder中的属性，是需要被随时读取和监测的。

那么究竟使用什么工具来读取order.xml文件中的各种属性呢。另外，一个CrawlOrder的对象又该如何构建呢？Heritrix提供了很好的工具支持对于order.xml的读取。在org.archive.crawler.settings包下有一个XMLSettingsHandler类，它可以用来帮助读取order.xml。

public XMLSettingsHandler(File orderFile) throws InvalidAttributeValueException

在XMLSettingsHandler的构造函数中，其所传入的参数orderFile正是一个经过对象封装的order.xml的File。这样，就可以直接调用其构造函数，来创建一个XMLSettingsHandler的实例，以此做为一个读取order.xml的工具。

当一个XMLSettingsHandler的实例被创建后，可以通过getOrder()方法来获取CrawlOrder的实例，这样也就可以进行下一步的工作了。

10.2.2  中央控制器CrawlController

中央控制器是一次抓取任务中的核心组件。它将决定整个抓取任务的开始和结束。CrawlController位于org.archive.crawler.framework中，在它的Field声明中，看到如下代码片段。

代码10.2

// key subcomponents which define and implement a crawl in progress

private transient CrawlOrder order;

private transient CrawlScope scope;

private transient ProcessorChainList processorChains;

private transient Frontier frontier;

private transient ToePool toePool;

private transient ServerCache serverCache;

// This gets passed into the initialize method.

private transient SettingsHandler settingsHandler;

可以看到，在CrawlController类中，定义了以下几个组件：

l  CrawlOrder：这就不用说了，因为一个抓取工作必须要有一个Order对象，它保存了对该次抓取任务中，order.xml的属性配置。

l  CrawlScope：在10.1.4节中已经介绍过了，这是决定当前的抓取范围的一个组件。

l  ProcessorChainList：从名称上很明显就能看出，它表示了处理器链，在这个列表中的每一项都可以和10.1.4节中所介绍的处理器链对应上。

l  Frontier：很明显，一次抓取任务需要设定一个Frontier，以此来不断为其每个线程提供URI。

l  ToePool：这是一个线程池，它管理了所有该抓取任务所创建的子线程。

l  ServerCache：这是一个缓存，它保存了所有在当前任务中，抓取过的Host名称和Server名称。

以上组件应该是一次正常的抓取过程中所必需的几项，它们各自的任务很独立，分工明确，但在后台中，它们之间却有着千丝万缕的联系，彼此互相做为构造函数或初始化的参数传入。

那么，究竟该如何获得CrawlController的实例，并且通过自主的编程来使用Heritrix提供的API进行一次抓任务呢？

事实上CrawlController有一个不带参数的构造函数，开发者可以直接通过它的构造函数来构造一个CrawlController的实例。但是值得注意的一点，在构造一个实例并进行抓取任务时，有几个步骤需要完成：

（1）首先构造一个XMLSettingsHandler对象，将order.xml内的属性信息装入。

（2）调用CrawlController的构造函数，构造一个CrawlController的实例。

（3）调用CrawlController的intialize(SettingsHandler)方法，初始化CrawlController实例。其中，传入的参数是在第一步是构造的XMLSettingsHandler实例。

（4）当上述3步完成后，CrawlController就已经具备运行的条件，可以开始运行了。此时，只需调用它的requestCrawlStart()方法，就可以启运线程池和Frontier，然后就可以开始不断的抓取网页了。

上述过程可以用图10-53所示。

图10-53  使用CrawlController启运抓取任务

在CrawlController的initialize()方法中，Heritrix主要做了以下几件事：

（1）从XMLSettingsHandler中取出Order。

（2）检查了用户设定的UserAgent等信息，看是否符合格式。

（3）设定了开始抓取后保存文件信息的目录结构。

（4）初始化了日志信息的记录工具。

（5）初始化了使用Berkley DB的一些工具。

（6）初始化了Scope、Frontier以及ProcessorChain。

（7）最后实例化了线程池。

在正常情况下，以上顺序不能够被随意变动，因为后一项功能的初始化很有可能需要前几项功能初始化的结果。例如线程池的初始化，必须要在先有了Frontier的实例的基础上来进行。读者可能对其中的Berkeley DB感到费解，在后面的小节将详细说明。

从图10-53中看到，最终启动抓取工作的是requestCrawlStart()方法。其代码如下。

代码10.3

public void requestCrawlStart() {

    // 初始化处理器链

    runProcessorInitialTasks();

    // 设置一下抓取状态的改变，以便能够激发一些Listeners

    // 来处理相应的事件

    sendCrawlStateChangeEvent(STARTED, CrawlJob.STATUS_PENDING);

    String jobState;

    state = RUNNING;

    jobState = CrawlJob.STATUS_RUNNING;

    sendCrawlStateChangeEvent(this.state, jobState);

    // A proper exit will change this value.

    this.sExit = CrawlJob.STATUS_FINISHED_ABNORMAL;

    // 开始日志线程

    Thread statLogger = new Thread(statistics);

    statLogger.setName("StatLogger");

    statLogger.start();

    // 启运Frontier，抓取工作开始

    frontier.start();

}

可以看到，启动抓取工作的核心就是要启动Frontier（通过调用其start()方法），以便能够开始向线程池中的工作线程提供URI，供它们抓取。

下面的代码就是BdbFrontier的父类AbstractFrontier中的start()方法和unpause()方法：

代码10.4

public void start() {

    if (((Boolean)getUncheckedAttribute(null, ATTR_PAUSE_AT_START))

.booleanValue()) {

    // 若配置文件中不允许该次抓取开始

    // 则停止

        controller.requestCrawlPause();

    } else {

    // 若允许开始，则开始

        unpause();

    }

}

synchronized public void unpause() {

    // 去除当前阻塞变量

    shouldPause = false;

    // 唤醒所有阻塞线程，开始抓取任务

    notifyAll();

}

在start()方法中，首先判断配置中的属性是否允许当前线程开始。若不允许，则令controller停止抓取。若允许开始，则简单的调用unpause()方法。unpause()方法更为简单，它首先将阻塞线程的信号量设为false，即允许线程开始活动，然后通过notifyAll()方法，唤醒线程池中所有被阻塞的线程，开始抓取。

10.2.3  Frontier链接制造工厂

Frontier在英文中的意思是“前线，领域”，在Heritrix中，它表示一种为线程提供链接的工具。它通过一些特定的算法来决定哪个链接将接下来被送入处理器链中，同时，它本身也负责一定的日志和状态报告功能。

事实上，要写出一个合格并且真正能够使用的Frontier绝非一件简单的事情，尽管有了Frontier接口，其中的方法约束了Frontier的行为，也给编码带来了一定的指示。但是其中还存在着很多问题，需要很好的设计和处理才可以解决。

在Heritrix的官方文档上，有一个Frontier的例子，在此拿出来进行一下讲解，以此来向读者说明一个最简单的Frontier都能够做什么事。以下就是这个Frontier的代码。

代码10.5

public class MyFrontier extends ModuleType implements Frontier,

        FetchStatusCodes {

    // 列表中保存了还未被抓取的链接

    List pendingURIs = new ArrayList();

   

    // 这个列表中保存了一系列的链接，它们的优先级

    // 要高于pendingURIs那个List中的任何一个链接

    // 表中的链接表示一些需要被满足的先决条件

    List prerequisites = new ArrayList();

   

    // 一个HashMap，用于存储那些已经抓取过的链接

    Map alreadyIncluded = new HashMap();

   

    // CrawlController对象

    CrawlController controller;

    // 用于标识是否一个链接正在被处理

    boolean uriInProcess = false;

   

    // 成功下载的数量

    long successCount = 0;

    // 失败的数量

    long failedCount = 0;

    // 抛弃掉链接的数量

    long disregardedCount = 0;

    // 总共下载的字节数

    long totalProcessedBytes = 0;

    // 构造函数

    public MyFrontier(String name) {

        super(Frontier.ATTR_NAME, "A simple frontier.");

    }

    // 初始化，参数为一个CrawlController

    public void initialize(CrawlController controller)

            throws FatalConfigurationException, IOException {

       

        // 注入

        this.controller = controller;

        

        // 把种子文件中的链接加入到pendingURIs中去

        this.controller.getScope().refreshSeeds();

        List seeds = this.controller.getScope().getSeedlist();

        synchronized(seeds) {

            for (Iterator i = seeds.iterator(); i.hasNext();) {

                UURI u = (UURI) i.next();

                CandidateURI caUri = new CandidateURI(u);

                caUri.setSeed();

                schedule(caUri);

            }

        }

    }

    // 该方法是给线程池中的线程调用的，用以取出下一个备处理的链接

    public synchronized CrawlURI next(int timeout) throws InterruptedException {

        if (!uriInProcess && !isEmpty()) {

            uriInProcess = true;

            CrawlURI curi;

            /*

             * 算法很简单，总是先看prerequistes队列中是否有

             * 要处理的链接，如果有，就先处理，如果没有

             * 再看pendingURIs队列中是否有链接

             * 每次在处理的时候，总是取出队列中的第一个链接

             */

            if (!prerequisites.isEmpty()) {

                curi = CrawlURI.from((CandidateURI) prerequisites.remove(0));

            } else {

                curi = CrawlURI.from((CandidateURI) pendingURIs.remove(0));

            }

            curi.setServer(controller.getServerCache().getServerFor(curi));

            return curi;

        } else {

            wait(timeout);

            return null;

        }

    }

    public boolean isEmpty() {

        return pendingURIs.isEmpty() && prerequisites.isEmpty();

    }

    // 该方法用于将新链接加入到pendingURIs队列中，等待处理

    public synchronized void schedule(CandidateURI caURI) {

        /*

         * 首先判断要加入的链接是否已经被抓取过

         * 如果已经包含在alreadyIncluded这个HashMap中

         * 则说明处理过了，即可以放弃处理

         */

        if (!alreadyIncluded.containsKey(caURI.getURIString())) {

            if(caURI.needsImmediateScheduling()) {

                prerequisites.add(caURI);

            } else {

                pendingURIs.add(caURI);

            }

            // HashMap中使用url的字符串来做为key

            // 而将实际的CadidateURI对象做为value

            alreadyIncluded.put(caURI.getURIString(), caURI);

        }

    }

    public void batchSchedule(CandidateURI caURI) {

        schedule(caURI);

    }

    public void batchFlush() {

    }

    // 一次抓取结束后所执行的操作，该操作由线程池

    // 中的线程来进行调用

    public synchronized void finished(CrawlURI cURI) {

        uriInProcess = false;

       

        // 成功下载

        if (cURI.isSuccess()) {

            successCount++;

            // 统计下载总数

            totalProcessedBytes += cURI.getContentSize();

            // 如果成功，则触发一个成功事件

            // 比如将Extractor解析出来的新URL加入队列中

            controller.fireCrawledURISuccessfulEvent(cURI);

            cURI.stripToMinimal();

        }

        // 需要推迟下载

        else if (cURI.getFetchStatus() == S_DEFERRED) {

            cURI.processingCleanup();

            alreadyIncluded.remove(cURI.getURIString());

            schedule(cURI);

        }

        // 其他状态

        else if (cURI.getFetchStatus() == S_ROBOTS_PRECLUDED

                || cURI.getFetchStatus() == S_OUT_OF_SCOPE

                || cURI.getFetchStatus() == S_BLOCKED_BY_USER

                || cURI.getFetchStatus() == S_TOO_MANY_EMBED_HOPS

                || cURI.getFetchStatus() == S_TOO_MANY_LINK_HOPS

                || cURI.getFetchStatus() == S_DELETED_BY_USER) {

            // 抛弃当前URI

            controller.fireCrawledURIDisregardEvent(cURI);

            disregardedCount++;

            cURI.stripToMinimal();

        } else {

            controller.fireCrawledURIFailureEvent(cURI);

            failedCount++;

            cURI.stripToMinimal();

        }

        cURI.processingCleanup();

    }

   

    // 返回所有已经处理过的链接数量

    public long discoveredUriCount() {

        return alreadyIncluded.size();

    }

    // 返回所有等待处理的链接的数量

    public long queuedUriCount() {

        return pendingURIs.size() + prerequisites.size();

    }

    // 返回所有已经完成的链接数量

    public long finishedUriCount() {

        return successCount + failedCount + disregardedCount;

    }

    // 返回所有成功处理的链接数量

    public long successfullyFetchedCount() {

        return successCount;

    }

    // 返回所有失败的链接数量

    public long failedFetchCount() {

        return failedCount;

    }

    // 返回所有抛弃的链接数量

    public long disregardedFetchCount() {

        return disregardedCount;

    }

    // 返回总共下载的字节数

    public long totalBytesWritten() {

        return totalProcessedBytes;

    }

    public String report() {

        return "This frontier does not return a report.";

    }

    public void importRecoverLog(String pathToLog) throws IOException {

        throw new UnsupportedOperationException();

    }

    public FrontierMarker getInitialMarker(String regexpr,

            boolean inCacheOnly) {

        return null;

    }

    public ArrayList getURIsList(FrontierMarker marker, int numberOfMatches,

            boolean verbose) throws InvalidFrontierMarkerException {

        return null;

    }

    public long deleteURIs(String match) {

        return 0;

    }

}

在Frontier中，根据笔者给出的中文注释，相信读者已经能够了解这个Frontier中的大部分玄机。以下给出详细的解释。

首先，Frontier是用来向线程提供链接的，因此，在上面的代码中，使用了两个ArrayList来保存链接。其中，第一个pendingURIs保存的是等待处理的链接，第二个prerequisites中保存的也是链接，只不过它里面的每个链接的优先级都要高于pendingURIs里的链接。通常，在prerequisites中保存的都是如DNS之类的链接，只有当这些链接被首先解析后，其后续的链接才能够被解析。

除了这两个ArrayList外，在上面的Frontier还有一个名称为alreadyIncluded的HashMap。它用于记录那些已经被处理过的链接。每当调用Frontier的schedule()方法来加入一个新的链接时，Frontier总要先检查这个正要加入到队列中的链接是不是已经被处理过了。

很显然，在分析网页的时候，会出现大量相同的链接，如果没有这种检查，很有可能造成抓取任务永远无法完成的情况。同时，在schedule()方法中还加入了一些逻辑，用于判断当前要进入队列的链接是否属于需要优先处理的，如果是，则置入prerequisites队列中，否则，就简单的加入pendingURIs中即可。

注意：Frontier中还有两个关键的方法，next()和finished()，这两个方法都是要交由抓取的线程来完成的。Next()方法的主要功能是：从等待队列中取出一个链接并返回，然后抓取线程会在它自己的run()方法中完成对这个链接的处理。而finished()方法则是在线程完成对链接的抓取和后续的一切动作后（如将链接传递经过处理器链）要执行的。它把整个处理过程中解析出的新的链接加入队列中，并且在处理完当前链接后，将之加入alreadyIncluded这个HashMap中去。

需要读者记住的是，这仅仅是一个最基础的代码，它有很多的功能缺失和性能问题，甚至可能出现重大的同步问题。不过尽管如此，它应当也起到了抛砖引玉的作用，能够从结构上揭示了一个Frontier的作用。

10.2.4  用Berkeley DB实现的BdbFrontier

简单的说，Berkeley DB就是一个HashTable，它能够按“key/value”方式来保存数据。它是由美国Sleepycat公司开发的一套开放源代码的嵌入式数据库，它为应用程序提供可伸缩的、高性能的、有事务保护功能的数据管理服务。

那么，为什么不使用一个传统的关系型数据库呢？这是因为当使用BerkeleyDB时，数据库和应用程序在相同的地址空间中运行，所以数据库操作不需要进程间的通讯。然而，当使用传统关系型数据库时，就需要在一台机器的不同进程间或在网络中不同机器间进行进程通讯，这样所花费的开销，要远远大于函数调用的开销。

另外，Berkeley DB中的所有操作都使用一组API接口。因此，不需要对某种查询语言（比如SQL）进行解析，也不用生成执行计划，这就大大提高了运行效率。

当然，做为一个数据库，最重要的功能就是事务的支持，Berkeley DB中的事务子系统就是用来为其提供事务支持的。它允许把一组对数据库的修改看作一个原子单位，这组操作要么全做，要么全不做。在默认的情况下，系统将提供严格的ACID事务属性，但是应用程序可以选择不使用系统所作的隔离保证。该子系统使用两段锁技术和先写日志策略来保证数据的正确性和一致性。这种事务的支持就要比简单的HashTable中的Synchronize要更加强大。

注意：在Heritrix中，使用的是Berkeley DB的Java版本，这种版本专门为Java语言做了优化，提供了Java的API接口以供开发者使用。

为什么Heritrix中要用到Berkeley DB呢？这就需要再回过头来看一下Frontier了。

在上一小节中，当一个链接被处理后，也即经过处理器链后，会生成很多新的链接，这些新的链接需要被Frontier的一个schedule方法加入到队列中继续处理。但是，在将这些新链接加入到队列之前，要首先做一个检查，即在alreadyIncluded这个HashMap中，查看当前要加入到队列中的链接是否在先前已经被处理过了。

当使用HashMap来存储那些已经被处理过的链接时，HashMap中的key为url，而value则为一个对url封装后的对象。很显然的，这里有几个问题。

l  对这个HashMap的读取是多线程的，因为每个线程都需要访问这个HashMap，以决定当前要加入链接是否已经存在过了。

l  对这个HashMap的写入是多线程的，每个线程在处理完毕后，都会访问这个HashMap，以写入最新处理的链接。

l  这个HashMap的容量可能很大，可以试想，一次在广域网范围上的网页抓取，可能会涉及到上十亿个URL地址，这种地址包括网页、图片、文件、多媒体对象等，所以，不可能将这么大一张表完全的置放于内存中。

综合考虑以上3点，仅用一个HashMap来保存所有的链接，显然已经不能满足“大数据量，多并发”这样的要求。因此，需要寻找一个替代的工具来解决问题。Heritrix中的BdbFrontier就采用了Berkeley DB，来解决这种URL存放的问题。事实上，BdbFrontier就是Berkeley DB Frontier的简称。

为了在BdbFrontier中使用Berkeley DB，Heritrix本身构造了一系列的类来帮助实现这个功能。这些类如下：

l  BdbFrontier

l  BdbMultipleWorkQueues

l  BdbWorkQueue

l  BdbUriUniqFilter

上述的4个类，都以Bdb3个字母开头，这表明它们都是使用到了Berkeley DB的功能。其中：

（1）BdbMultipleWorkQueues代表了一组链接队列，这些队列有各自不同的key。这样，由Key和链接队列可以形成一个“Key/Value”对，也就成为了Berkeley DB里的一条记录（DatabaseEntry）如图10-54所示。

图10-54  BdbMultipleWorkQueues示意

图10-54清楚的显示了Berkeley DB中的“key/value”形式。可以说，这就是一张Berkeley DB的数据库表。其中，数据库的一条记录包含两个部分，左边是一个由右边的所有URL链接计算出来的公共键值，右边则是一个URL的队列。

（2）BdbWorkQueue代表了一个基于Berkeley DB的队列，与BdbMutipleWorkQueues所不同的是，该队列中的所有的链接都具有相同的键值。事实上，BdbWorkQueue只是对BdbMultipleWorkQueues的封装，在构造一个BdbWorkQueue时，需传入一个健值，以此做为该Queue在数据库中的标识。事实上，在工作线程从Frontier中取出链接时，Heritrix总是先取出整个BdbWorkQueue，再从中取出第一个链接，然后将当前这个BdbWorkQueue置入一个线程安全的同步容器内，等待线程处理完毕后才将该Queue释放，以便该Queue内的其他URI可以继续被处理。

（3）BdbUriUniqFilter是一个过滤器，从名称上就能知道，它是专门用来过滤当前要进入等待队列的链接对象是否已经被抓取过。很显然，在BdbUriUniqFilter内部嵌入了一个Berkeley DB数据库用于存储所有的被抓取过的链接。它对外提供了

public void add(String key, CandidateURI value)

这样的接口，以供Frontier调用。当然，若是参数的CandidateURI已经存在于数据库中了，则该方法会禁止它加入到等待队列中去。

（4）BdbFrontier就是Heritrix中使用了Berkeley DB的链接制造工厂。它主要使用BdbUriUniqFilter，做为其判断当前要进入等待队列的链接对象是否已经被抓取过。同时，它还使用了BdbMultipleWorkQueues来做为所有等待处理的URI的容器。这些URI根据各自的内容会生成一个Hash值成为它们所在队列的键值。

在Heritrix1.10的版本中，可以说BdbFrontier是惟一一个具有实用意义的链接制造工厂了。虽然Heritrix还提供了另外两个Frontier：

org.archive.crawler.frontier.DomainSensitiveFrontier

org.archive.crawler.frontier.AdaptiveRevisitFrontier

但是，DomainSensitiveFrontier已经被废弃不再推荐使用了。而AdaptiveRevisitFrontier的算法是不管遇到什么新链接，都义无反顾的再次抓取，这显然是一种很落后的算法。因此，了解BdbFrontier的实现原理，对于更好的了解Heritrix对链接的处理有实际意义。

BdbFrontier的代码相对比较复杂，笔者在这里也只能简单将其轮廓进行介绍，读者仍须将代码仔细研读，方能把文中的点点知识串联起来，进而更好的理解Heritrix作者们的巧妙匠心。

10.2.5  Heritrix的多线程ToeThread和ToePool

想要更有效更快速的抓取网页内容，则必须采用多线程。Heritrix中提供了一个标准的线程池ToePool，它用于管理所有的抓取线程。

ToePool和ToeThread都位于org.archive.crawler.framework包中。前面已经说过，ToePool的初始化，是在CrawlController的initialize()方法中完成的。来看一下ToePool以及ToeThread是如何被初始化的。以下代码是在CrawlController中用于对ToePool进行初始化的。

// 构造函数

toePool = new ToePool(this);

// 按order.xml中的配置，实例化并启动线程

toePool.setSize(order.getMaxToes());

ToePool的构造函数很简单，如下所示：

public ToePool(CrawlController c) {

    super("ToeThreads");

    this.controller = c;

}

它仅仅是调用了父类java.lang.ThreadGroup的构造函数，同时，将注入的CrawlController赋给类变量。这样，便建立起了一个线程池的实例了。但是，那些真正的工作线程又是如何建立的呢？

下面来看一下线程池中的setSize(int)方法。从名称上看，这个方法很像是一个普通的赋值方法，但实际上，它并不是那么简单。

代码10.6

public void setSize(int newsize)

{

    targetSize = newsize;

    int difference = newsize - getToeCount();

   

    // 如果发现线程池中的实际线程数量小于应有的数量

    // 则启动新的线程

    if (difference > 0) {

        for(int i = 1; i <= difference; i++) {

            // 启动新线程

             startNewThread();

        }

    }

    // 如果线程池中的线程数量已经达到需要

    else

    {

       

        int retainedToes = targetSize;

        // 将线程池中的线程管理起来放入数组中

        Thread[] toes = this.getToes();

       

        // 循环去除多余的线程

        for (int i = 0; i < toes.length ; i++) {

             if(!(toes[i] instanceof ToeThread)) {

                     continue;

             }

             retainedToes--;

             if (retainedToes>=0) {

                     continue;

             }

             ToeThread tt = (ToeThread)toes[i];

             tt.retire();

        }

    }

}

// 用于取得所有属于当前线程池的线程

private Thread[] getToes()

{

    Thread[] toes = new Thread[activeCount()+10];

    // 由于ToePool继承自java.lang.ThreadGroup类

    // 因此当调用enumerate(Thread[] toes)方法时，

    // 实际上是将所有该ThreadGroup中开辟的线程放入

    // toes这个数组中，以备后面的管理

    this.enumerate(toes);

    return toes;

}

// 开启一个新线程

private synchronized void startNewThread()

{

    ToeThread newThread = new ToeThread(this, nextSerialNumber++);

    newThread.setPriority(DEFAULT_TOE_PRIORITY);

    newThread.start();

}

通过上面的代码可以得出这样的结论：线程池本身在创建的时候，并没有任何活动的线程实例，只有当它的setSize方法被调用时，才有可能创建新线程；如果当setSize方法被调用多次而传入不同的参数时，线程池会根据参数里所设定的值的大小，来决定池中所管理线程数量的增减。

当线程被启动后，所执行的是其run()方法中的片段。接下来，看一个ToeThread到底是如何处理从Frontier中获得的链接的。

代码10.7

public void run()

{

    String name = controller.getOrder().getCrawlOrderName();

    logger.fine(getName()+" started for order '"+name+"'");

    try {

        while ( true )

        {

             // 检查是否应该继续处理

            continueCheck();

           

            setStep(STEP_ABOUT_TO_GET_URI);

            

             // 使用Frontier的next方法从Frontier中

             // 取出下一个要处理的链接

             CrawlURI curi = controller.getFrontier().next();

            

             // 同步当前线程

                synchronized(this) {

                    continueCheck();

                    setCurrentCuri(curi);

                }

               

                /*

                 * 处理取出的链接

                 */

                processCrawlUri();

               

                setStep(STEP_ABOUT_TO_RETURN_URI);

               

                // 检查是否应该继续处理

                continueCheck();

               

                // 使用Frontier的finished()方法

                // 来对刚才处理的链接做收尾工作

                // 比如将分析得到的新的链接加入

                // 到等待队列中去

                synchronized(this) {

                    controller.getFrontier().finished(currentCuri);

                    setCurrentCuri(null);

                }

               

                // 后续的处理

                setStep(STEP_FINISHING_PROCESS);

                lastFinishTime = System.currentTimeMillis();

               // 释放链接

                controller.releaseContinuePermission();

                if(shouldRetire) {

                    break; // from while(true)

                }

        }

    } catch (EndedException e) {

    } catch (Exception e) {

        logger.log(Level.SEVERE,"Fatal exception in "+getName(),e);

    } catch (OutOfMemoryError err) {

        seriousError(err);

    } finally {

        controller.releaseContinuePermission();

    }

    setCurrentCuri(null);

   

    // 清理缓存数据

    this.httpRecorder.closeRecorders();

    this.httpRecorder = null;

    localProcessors = null;

   

    logger.fine(getName()+" finished for order '"+name+"'");

    setStep(STEP_FINISHED);

    controller.toeEnded();

    controller = null;

}

在上面的方法中，很清楚的显示了工作线程是如何从Frontier中取得下一个待处理的链接，然后对链接进行处理，并调用Frontier的finished方法来收尾、释放链接，最后清理缓存、终止单步工作等。另外，其中还有一些日志操作，主要是为了记录每次抓取的各种状态。

很显然，以上代码中，最重要的一行语句是processCrawlUri()，它是真正调用处理链来对链接进行处理的代码。其中的内容，放在下一个小节中介绍。

10.2.6  处理链和Processor

在本章第一节中介绍了设置处理器链相关的内容。从中知道，处理器链包括以下几种：

l  PreProcessor

l  Fetcher

l  Extractor

l  Writer

l  PostProcessor

为了很好的表示整个处理器链的逻辑结构，以及它们之间的链式调用关系，Heritrix设计了几个API来表示这种逻辑结构。

org.archive.crawler.framework.Processor

org.archive.crawler.framework.ProcessorChain

org.archive.crawler.framework.ProcessorChainList

下面进行详细讲解。

1．Processor类

该类代表着单个的处理器，所有的处理器都是它的子类。在Processor类中有一个process()方法，它被标识为final类型的，也就是说，它不可以被它的子类所覆盖。代码如下。

代码10.8

public final void process(CrawlURI curi) throws InterruptedException

{

    // 设置下一个处理器

    curi.setNextProcessor(getDefaultNextProcessor(curi));

    try

    {

       // 判断当前这个处理器是否为enabled

       if (!((Boolean) getAttribute(ATTR_ENABLED, curi)).booleanValue()) {

        return;

       }

    } catch (AttributeNotFoundException e) {

       logger.severe(e.getMessage());

    }

    // 如果当前的链接能够通过过滤器

    // 则调用innerProcess(curi)方法

    // 来进行处理

    if(filtersAccept(curi)) {

       innerProcess(curi);

    }

    // 如果不能通过过滤器检查，则调

    // 用innerRejectProcess(curi)来处理

    else

    {

       innerRejectProcess(curi);

    }

}

方法的含义很简单。即首先检查是否允许这个处理器处理该链接，如果允许，则检查当前处理器所自带的过滤器是否能够接受这个链接。当过滤器的检查也通过后，则调用innerProcess(curi)方法来处理，如果过滤器的检查没有通过，就使用innerRejectProcess(curi)方法处理。

其中innerProcess(curi)和innerRejectProcess(curi)方法都是protected类型的，且本身没有实现任何内容。很明显它们是留在子类中，实现具体的处理逻辑。不过大部分的子类都不会重写innerRejectProcess(curi)方法了，这是因为反正一个链接已经被当前处理器拒绝处理了，就不用再有什么逻辑了，直接跳到下一个处理器继续处理就行了。

2．ProcessorChain类

该类表示一个队列，里面包括了同种类型的几个Processor。例如，可以将一组的Extractor加入到同一个ProcessorChain中去。

在一个ProcessorChain中，有3个private类型的类变量：

private final MapType processorMap;

private ProcessorChain nextChain;

private Processor firstProcessor;

其中，processorMap中存放的是当前这个ProcessorChain中所有的Processor。nextChain的类型是ProcessorChain，它表示指向下一个处理器链的指针。而firstProcessor则是指向当前队列中的第一个处理器的指针。

3．ProcessorChainList

从名称上看，它保存了Heritrix一次抓取任务中所设定的所有处理器链，将之做为一个列表。正常情况下，一个ProcessorChainList中，应该包括有5个ProcessorChain，分别为PreProcessor链、Fetcher链、Extractor链、Writer链和PostProcessor链，而每个链中又包含有多个的Processor。这样，就将整个处理器结构合理的表示了出来。

那么，在ToeThread的processCrawlUri()方法中，又是如何来将一个链接循环经过这样一组结构的呢？请看下面的代码：

代码10.9

private void processCrawlUri() throws InterruptedException {

   // 设定当前线程的编号

   currentCuri.setThreadNumber(this.serialNumber);

   // 为当前处理的URI设定下一个ProcessorChain

   currentCuri.setNextProcessorChain(controller.getFirstProcessorChain());

  

   // 设定开始时间

   lastStartTime = System.currentTimeMillis();

   try {

  

      // 如果还有一个处理链没处理完

      while (currentCuri.nextProcessorChain() != null)

      {

         setStep(STEP_ABOUT_TO_BEGIN_CHAIN);

        

         // 将下个处理链中的第一个处理器设定为

         // 下一个处理当前链接的处理器

         currentCuri.setNextProcessor(currentCuri

                                      .nextProcessorChain().getFirstProcessor());

         // 将再下一个处理器链设定为当前链接的

         // 下一个处理器链，因为此时已经相当于

         // 把下一个处理器链置为当前处理器链了

         currentCuri.setNextProcessorChain(currentCuri

                                      .nextProcessorChain().getNextProcessorChain());

                                     

         // 开始循环处理当前处理器链中的每一个Processor

         while (currentCuri.nextProcessor() != null)

         {

            setStep(STEP_ABOUT_TO_BEGIN_PROCESSOR);

            Processor currentProcessor = getProcessor(currentCuri.nextProcessor());

            currentProcessorName = currentProcessor.getName();

            continueCheck();

            // 调用Process方法

            currentProcessor.process(currentCuri);

         }

      }

      setStep(STEP_DONE_WITH_PROCESSORS);

      currentProcessorName = "";

   }

   catch (RuntimeExceptionWrapper e) {

       // 如果是Berkeley DB的异常

       if(e.getCause() == null) {

           e.initCause(e.getDetail());

       }

       recoverableProblem(e);

   } catch (AssertionError ae) {

       recoverableProblem(ae);

   } catch (RuntimeException e) {

       recoverableProblem(e);

   } catch (StackOverflowError err) {

       recoverableProblem(err);

   } catch (Error err) {

       seriousError(err);

   }

}

代码使用了双重循环来遍历整个处理器链的结构，第一重循环首先遍历所有的处理器链，第二重循环则在链内部遍历每个Processor，然后调用它的process()方法来执行处理逻辑。

<!-- page -->