4、Heritrix的多线程ToeThread和ToePool
要想更有效更快捷地抓取网页内容，则必须采用多线程。Heritirx提供了一个标准的线程池ToeThread，用于管理所有的抓取线程。
org.archive.crawler.framework
Class ToePool java.lang.Object
java.lang.ThreadGroup
org.archive.crawler.framework.ToePool
|
如前所述，ToePool的初始化是在CrawlController的初始化方法中完成的。
privatevoidsetupToePool() {
toePool = new ToePool(this);
//按order.xml中的配置，实例化并启动线程
toePool.setSize(order.getMaxToes());
}
// ToePool构造函数
public ToePool(CrawlController c) {
super("ToeThreads");
this.controller = c;
setDaemon(true);
}
|
1）真正工作的线程是如何建立的?
从上图可知，其最主要的方法是setSize(int),其源代码如下：
public void setSize(int newsize)
{
targetSize = newsize;
int difference = newsize - getToeCount();
//如果发现线程池中的实际线程数量小于应有的数量则启动新的线程
if (difference > 0) {
// 启动新线程
for(int i = 1; i <= difference; i++) {
startNewThread();
}
}
//如果线程池中的线程数量已经达到需要
else {
// must retire extra threads
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; // this toe is spared
}
// otherwise:
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方法被调用多次而传入不同的参数时，线程池会根据参数里所设定的值的大小，来决定池中所管理的线程数量的增减。
2)一个ToeThread到底是如何处理从Frontier中获得的链接？
当线程被启动后，执行的是其run（）方法中的片段。
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) {
// crawl ended (or thread was retired), so allow thread to end
} catch (Exception e) {
// everything else (including interruption)
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;
}
|
