深入学习Heritrix---解析Frontier(链接工厂)

http://caixinbao1.blog.163.com/blog/static/161494162009730115520760/

Frontier是Heritrix最核心的组成部分之一,也是最复杂的组成部分.它主要功能是为处理链接的线程提供URL,并负责链接处理完成后的一些后续调度操作.并且为了提高效率,它在内部使用了Berkeley DB.本节将对它的内部机理进行详细解剖.

在Heritrix的官方文档上有一个Frontier的例子,虽然很简单,但是它却解释Frontier实现的基本原理.在这里就不讨论,有兴趣的读者可以参考相应文档.但是不得不提它的三个核心方法:
(1)next(int timeout):为处理线程提供一个链接.Heritrix的所有处理线程(ToeThread)都是通过调用该方法获取链接的.

(2)schedule(CandidateURI caURI):调度待处理的链接.
(3)finished(CrawlURI cURI):完成一个已处理的链接.
整体结构如下:



BdbMultipleWorkQueues:
它是对Berkeley DB的简单封装.在内部有一个Berkeley Database,存放所有待处理的链接.




Code


package org.archive.crawler.frontier;


public class BdbMultipleWorkQueues


{


//存放所有待处理的URL的数据库


private Database pendingUrisDB = null;






//由key获取一个链接


public CrawlURI get(DatabaseEntry headKey)


throws DatabaseException {


DatabaseEntry result = new DatabaseEntry();




// From Linda Lee of sleepycat:


// "You want to check the status returned from Cursor.getSearchKeyRange


// to make sure that you have OperationStatus.SUCCESS. In that case,


// you have found a valid data record, and result.getData()


// (called by internally by the binding code, in this case) will be


// non-null. The other possible status return is


// OperationStatus.NOTFOUND, in which case no data record matched


// the criteria. "


//由key获取相应的链接


OperationStatus status = getNextNearestItem(headKey, result);


CrawlURI retVal = null;


if (status != OperationStatus.SUCCESS) {


LOGGER.severe("See '1219854 NPE je-2.0 "


+ "entryToObject

'. OperationStatus "


+ " was not SUCCESS: "


+ status


+ ", headKey "


+ BdbWorkQueue.getPrefixClassKey(headKey.getData()));


return null;


}


try {


retVal = (CrawlURI)crawlUriBinding.entryToObject(result);


} catch (RuntimeExceptionWrapper rw) {


LOGGER.log(


Level.SEVERE,


"expected object missing in queue " +


BdbWorkQueue.getPrefixClassKey(headKey.getData()),


rw);


return null;


}


retVal.setHolderKey(headKey);


return retVal;//返回链接


}




//从等处理列表获取一个链接


protected OperationStatus getNextNearestItem(DatabaseEntry headKey,


DatabaseEntry result) throws DatabaseException {


Cursor cursor = null;


OperationStatus status;


try {


//打开游标


cursor = this.pendingUrisDB.openCursor(null, null);


// get cap; headKey at this point should always point to


// a queue-beginning cap entry (zero-length value)


status = cursor.getSearchKey(headKey, result, null);


if(status!=OperationStatus.SUCCESS || result.getData().length > 0) {


// cap missing


throw new DatabaseException("bdb queue cap missing");


}


// get next item (real first item of queue)


status = cursor.getNext(headKey,result,null);


} finally {


if(cursor!=null) {


cursor.close();


}


}


return status;


}




/**


* Put the given CrawlURI in at the appropriate place.


* 添加URL到数据库


* @param curi


* @throws DatabaseException


*/


public void put(CrawlURI curi, boolean overwriteIfPresent)


throws DatabaseException {


DatabaseEntry insertKey = (DatabaseEntry)curi.getHolderKey();


if (insertKey == null) {


insertKey = calculateInsertKey(curi);


curi.setHolderKey(insertKey);


}


DatabaseEntry value = new DatabaseEntry();


crawlUriBinding.objectToEntry(curi, value);


// Output tally on avg. size if level is FINE or greater.


if (LOGGER.isLoggable(Level.FINE)) {


tallyAverageEntrySize(curi, value);


}


OperationStatus status;


if(overwriteIfPresent) {


//添加


status = pendingUrisDB.put(null, insertKey, value);


} else {


status = pendingUrisDB.putNoOverwrite(null, insertKey, value);


}


if(status!=OperationStatus.SUCCESS) {


LOGGER.severe("failed; "+status+ " "+curi);


}


}




}

BdbWorkQueue:
代表一个链接队列,该队列中所有的链接都具有相同的键值.它实际上是通过调用BdbMultipleWorkQueues的get方法从等处理链接数据库中取得一个链接的.


Code
package org.archive.crawler.frontier;

public class BdbWorkQueue extends WorkQueue
implements Comparable, Serializabl
{
//获取一个URL
protected CrawlURI peekItem(final WorkQueueFrontier frontier)
throws IOException {
/**
* 关键:从BdbFrontier中返回pendingUris
*/
final BdbMultipleWorkQueues queues = ((BdbFrontier) frontier)
.getWorkQueues();

DatabaseEntry key = new DatabaseEntry(origin);
CrawlURI curi = null;
int tries = 1;
while(true) {
try {
//获取链接
curi = queues.get(key);
} catch (DatabaseException e) {
LOGGER.log(Level.SEVERE,"peekItem failure; retrying",e);
}



return curi;
}
}
WorkQueueFrontier:
实现了最核心的三个方法.


Code
public CrawlURI next()
throws InterruptedException, EndedException {
while (true) { // loop left only by explicit return or exception
long now = System.currentTimeMillis();

// Do common checks for pause, terminate, bandwidth-hold
preNext(now);

synchronized(readyClassQueues) {
int activationsNeeded = targetSizeForReadyQueues() - readyClassQueues.size();
while(activationsNeeded > 0 && !inactiveQueues.isEmpty()) {
activateInactiveQueue();
activationsNeeded--;
}
}

WorkQueue readyQ = null;
Object key = readyClassQueues.poll(DEFAULT_WAIT,TimeUnit.MILLISECONDS);
if (key != null) {
readyQ = (WorkQueue)this.allQueues.get(key);
}
if (readyQ != null) {
while(true) { // loop left by explicit return or break on empty
CrawlURI curi = null;
synchronized(readyQ) {
/**取出一个URL,最终从子类BdbFrontier的
* pendingUris中取出一个链接
*/
curi = readyQ.peek(this);
if (curi != null) {
// check if curi belongs in different queue
String currentQueueKey = getClassKey(curi);
if (currentQueueKey.equals(curi.getClassKey())) {
// curi was in right queue, emit
noteAboutToEmit(curi, readyQ);
//加入正在处理队列中
inProcessQueues.add(readyQ);
return curi; //返回
}
// URI's assigned queue has changed since it
// was queued (eg because its IP has become
// known). Requeue to new queue.
curi.setClassKey(currentQueueKey);
readyQ.dequeue(this);//出队列
decrementQueuedCount(1);
curi.setHolderKey(null);
// curi will be requeued to true queue after lock
// on readyQ is released, to prevent deadlock
} else {
// readyQ is empty and ready: it's exhausted
// release held status, allowing any subsequent
// enqueues to again put queue in ready
readyQ.clearHeld();
break;
}
}
if(curi!=null) {
// complete the requeuing begun earlier
sendToQueue(curi);
}
}
} else {
// ReadyQ key wasn't in all queues: unexpected
if (key != null) {
logger.severe("Key "+ key +
" in readyClassQueues but not allQueues");
}
}

if(shouldTerminate) {
// skip subsequent steps if already on last legs
throw new EndedException("shouldTerminate is true");
}

if(inProcessQueues.size()==0) {
// Nothing was ready or in progress or imminent to wake; ensure
// any piled-up pending-scheduled URIs are considered
this.alreadyIncluded.requestFlush();
}
}
}

//将URL加入待处理队列
public void schedule(CandidateURI caUri) {
// Canonicalization may set forceFetch flag. See
// #canonicalization(CandidateURI) javadoc for circumstance.
String canon = canonicalize(caUri);
if (caUri.forceFetch()) {
alreadyIncluded.addForce(canon, caUri);
} else {
alreadyIncluded.add(canon, caUri);
}
}

BdbFrontier:
继承了WorkQueueFrontier,是Heritrix唯一个具有实际意义的链接工厂.


Code
package org.archive.crawler.frontier;
public class BdbFrontier extends WorkQueueFrontier implements Serializable
{
/** 所有待抓取的链接*/
protected transient BdbMultipleWorkQueues pendingUris;

//初始化pendingUris,父类为抽象方法
protected void initQueue() throws IOException {
try {
this.pendingUris = createMultipleWorkQueues();
} catch(DatabaseException e) {
throw (IOException)new IOException(e.getMessage()).initCause(e);
}
}

private BdbMultipleWorkQueues createMultipleWorkQueues()
throws DatabaseException {
return new BdbMultipleWorkQueues(this.controller.getBdbEnvironment(),
this.controller.getBdbEnvironment().getClassCatalog(),
this.controller.isCheckpointRecover());
}
protected BdbMultipleWorkQueues getWorkQueues() {
return pendingUris;
}



}

BdbUriUniqFilter:
实际上是一个过滤器,它用来检查一个要进入等待队列的链接是否已经被抓取过.


Code
//添加URL
protected boolean setAdd(CharSequence uri) {
DatabaseEntry key = new DatabaseEntry();
LongBinding.longToEntry(createKey(uri), key);
long started = 0;

OperationStatus status = null;
try {
if (logger.isLoggable(Level.INFO)) {
started = System.currentTimeMillis();
}
//添加到数据库
status = alreadySeen.putNoOverwrite(null, key, ZERO_LENGTH_ENTRY);
if (logger.isLoggable(Level.INFO)) {
aggregatedLookupTime +=
(System.currentTimeMillis() - started);
}
} catch (DatabaseException e) {
logger.severe(e.getMessage());
}
if (status == OperationStatus.SUCCESS) {
count++;
if (logger.isLoggable(Level.INFO)) {
final int logAt = 10000;
if (count > 0 && ((count % logAt) == 0)) {
logger.info("Average lookup " +
(aggregatedLookupTime / logAt) + "ms.");
aggregatedLookupTime = 0;
}
}
}
//如果存在,返回false
if(status == OperationStatus.KEYEXIST) {
return false; // not added
} else {
return true;
}
}