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 WorkQueueimplements 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; } } 相关资源：敏捷开发V1.0.pptx