/*** Return the work queue for the given classKey, or null* if no such queue exists.* * @param classKey key to look for* @return the found WorkQueue*/protected WorkQueue getQueueFor(final String classKey) {      WorkQueue wq = allQueues.getOrUse(classKey,new Supplier<WorkQueue>() {public BdbWorkQueue get() {String qKey = new String(classKey); // ensure private minimal keyBdbWorkQueue q = new BdbWorkQueue(qKey, BdbFrontier.this);q.setTotalBudget(getQueueTotalBudget()); //-1
                        System.out.println(getQueuePrecedencePolicy().getClass().getName());getQueuePrecedencePolicy().queueCreated(q);return q;}});return wq;}

/*** Create a virtual queue inside the given BdbMultipleWorkQueues * * @param classKey*/public BdbWorkQueue(String classKey, BdbFrontier frontier) {super(classKey);this.origin = BdbMultipleWorkQueues.calculateOriginKey(classKey);if (LOGGER.isLoggable(Level.FINE)) {LOGGER.fine(getPrefixClassKey(this.origin) + " " + classKey);}// add the queue-front 'cap' entry; see...// http://sourceforge.net/tracker/index.php?func=detail&aid=1262665&group_id=73833&atid=539102
        frontier.getWorkQueues().addCap(origin);}

protected void insertItem(final WorkQueueFrontier frontier,final CrawlURI curi, boolean overwriteIfPresent) throws IOException {try {final BdbMultipleWorkQueues queues = ((BdbFrontier) frontier).getWorkQueues();queues.put(curi, overwriteIfPresent);if (LOGGER.isLoggable(Level.FINE)) {LOGGER.fine("Inserted into " + getPrefixClassKey(this.origin) +" (count " + Long.toString(getCount())+ "): " +curi.toString());}} catch (DatabaseException e) {throw new IOException(e);}}

protected CrawlURI peekItem(final WorkQueueFrontier frontier)throws IOException {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);}// ensure CrawlURI, if any,  came from acceptable range: if(!ArchiveUtils.startsWith(key.getData(),origin)) {LOGGER.severe("inconsistency: "+classKey+"("+getPrefixClassKey(origin)+") with " + getCount() + " items gave "+ curi +"("+getPrefixClassKey(key.getData()));// clear curi to allow retrycuri = null; // reset key to original origin for retry
                key.setData(origin);}if (curi!=null) {// successbreak;}if (tries>3) {LOGGER.severe("no item where expected in queue "+classKey);break;}tries++;LOGGER.severe("Trying get #" + Integer.toString(tries)+ " in queue " + classKey + " with " + getCount()+ " items using key "+ getPrefixClassKey(key.getData()));}return curi;}

protected void deleteItem(final WorkQueueFrontier frontier,final CrawlURI peekItem) throws IOException {try {final BdbMultipleWorkQueues queues = ((BdbFrontier) frontier).getWorkQueues();queues.delete(peekItem);} catch (DatabaseException e) {throw new IOException(e);}}

/*** Create the single object (within which is one BDB database)* inside which all the other queues live. * * @return the created BdbMultipleWorkQueues* @throws DatabaseException*/protected BdbMultipleWorkQueues createMultipleWorkQueues()throws DatabaseException {Database db;boolean recycle = (recoveryCheckpoint != null);BdbModule.BdbConfig dbConfig = new BdbModule.BdbConfig();dbConfig.setAllowCreate(!recycle);// Make database deferred write: URLs that are added then removed // before a page-out is required need never cause disk IO.db = bdb.openDatabase("pending", dbConfig, recycle);return new BdbMultipleWorkQueues(db, bdb.getClassCatalog());}

/*** Create the multi queue in the given environment. * * @param env bdb environment to use* @param classCatalog Class catalog to use.* @param recycle True if we are to reuse db content if any.* @throws DatabaseException*/public BdbMultipleWorkQueues(Database db,StoredClassCatalog classCatalog)throws DatabaseException {this.pendingUrisDB = db;crawlUriBinding =new KryoBinding<CrawlURI>(CrawlURI.class);
//            new RecyclingSerialBinding<CrawlURI>(classCatalog, CrawlURI.class);
//            new BenchmarkingBinding<CrawlURI>(new EntryBinding[] {
//                new KryoBinding<CrawlURI>(CrawlURI.class,true),
//                new KryoBinding<CrawlURI>(CrawlURI.class,false),
//                new RecyclingSerialBinding<CrawlURI>(classCatalog, CrawlURI.class),
//            });
            }

/*** Put the given CrawlURI in at the appropriate place. * * @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.log(Level.SEVERE,"URI enqueueing failed; "+status+ " "+curi, new RuntimeException());}}

/*** Calculate the insertKey that places a CrawlURI in the* desired spot. First bytes are always classKey (usu. host)* based -- ensuring grouping by host -- terminated by a zero* byte. Then 8 bytes of data ensuring desired ordering * within that 'queue' are used. The first byte of these 8 is* priority -- allowing 'immediate' and 'soon' items to * sort above regular. Next 1 byte is 'precedence'. Last 6 bytes * are ordinal serial number, ensuring earlier-discovered * URIs sort before later. * * NOTE: Dangers here are:* (1) priorities or precedences over 2^7 (signed byte comparison)* (2) ordinals over 2^48* * Package access & static for testing purposes. * * @param curi* @return a DatabaseEntry key for the CrawlURI*/static DatabaseEntry calculateInsertKey(CrawlURI curi) {byte[] classKeyBytes = null;int len = 0;classKeyBytes = curi.getClassKey().getBytes(Charsets.UTF_8);len = classKeyBytes.length;byte[] keyData = new byte[len+9];System.arraycopy(classKeyBytes,0,keyData,0,len);keyData[len]=0;long ordinalPlus = curi.getOrdinal() & 0x0000FFFFFFFFFFFFL;ordinalPlus = ((long)curi.getSchedulingDirective() << 56) | ordinalPlus;long precedence = Math.min(curi.getPrecedence(), 127);ordinalPlus = (((precedence) & 0xFFL) << 48) | ordinalPlus;ArchiveUtils.longIntoByteArray(ordinalPlus, keyData, len+1);return new DatabaseEntry(keyData);}

/*** Get the next nearest item after the given key. Relies on * external discipline -- we'll look at the queues count of how many* items it has -- to avoid asking for something from a* range where there are no associated items --* otherwise could get first item of next 'queue' by mistake. * * <p>TODO: hold within a queue's range* * @param headKey Key prefix that demarks the beginning of the range* in <code>pendingUrisDB</code> we're interested in.* @return CrawlURI.* @throws DatabaseException*/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. "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 (ClassCastException cce) {Object obj = crawlUriBinding.entryToObject(result);LOGGER.log(Level.SEVERE,"see [#HER-1283]: deserialized " + obj.getClass() + " has ClassLoader " + obj.getClass().getClassLoader().getClass(),cce);return null; } 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) {LOGGER.severe("bdb queue cap missing: " + status.toString() + " "  + new String(headKey.getData()));return status;}if (result.getData().length > 0) {LOGGER.severe("bdb queue has nonzero size: " + result.getData().length);return OperationStatus.KEYEXIST;}// get next item (real first item of queue)status = cursor.getNext(headKey,result,null);} finally { if(cursor!=null) {cursor.close();}}return status;}

/*** Delete the given CrawlURI from persistent store. Requires* the key under which it was stored be available. * * @param item* @throws DatabaseException*/public void delete(CrawlURI item) throws DatabaseException {OperationStatus status;DatabaseEntry de = (DatabaseEntry)item.getHolderKey();status = pendingUrisDB.delete(null, de);if (status != OperationStatus.SUCCESS) {LOGGER.severe("expected item not present: "+ item+ "("+ (new BigInteger(((DatabaseEntry) item.getHolderKey()).getData())).toString(16) + ")");}}

/*** 遍历记录* Utility method to perform action for all pending CrawlURI instances.* @param c Closure action to perform* @throws DatabaseException*/protected void forAllPendingDo(Closure c) throws DatabaseException {DatabaseEntry key = new DatabaseEntry();DatabaseEntry value = new DatabaseEntry();Cursor cursor = pendingUrisDB.openCursor(null, null);while (cursor.getNext(key, value, null) == OperationStatus.SUCCESS) {if (value.getData().length == 0) {continue;}CrawlURI item = (CrawlURI) crawlUriBinding.entryToObject(value);c.execute(item);}cursor.close(); }

/*** @param m marker or null to start with first entry* @param maxMatches* @return list of matches starting from marker position* @throws DatabaseException*/public CompositeData getFrom(String m, int maxMatches, Pattern pattern, boolean verbose) throws DatabaseException {int matches = 0;int tries = 0;ArrayList<String> results = new ArrayList<String>(maxMatches);DatabaseEntry key;if (m == null) {key = getFirstKey();} else {byte[] marker = m.getBytes(); // = FrontierJMXTypes.fromString(m);key = new DatabaseEntry(marker);}DatabaseEntry value = new DatabaseEntry();Cursor cursor = null;OperationStatus result = null;try {cursor = pendingUrisDB.openCursor(null,null);result = cursor.getSearchKey(key, value, null);while(matches < maxMatches && result == OperationStatus.SUCCESS) {if(value.getData().length>0) {CrawlURI curi = (CrawlURI) crawlUriBinding.entryToObject(value);if(pattern.matcher(curi.toString()).matches()) {if (verbose) {results.add("[" + curi.getClassKey() + "] " + curi.shortReportLine());} else {results.add(curi.toString());}matches++;}tries++;}result = cursor.getNext(key,value,null);}} finally {if (cursor !=null) {cursor.close();}}if(result != OperationStatus.SUCCESS) {// end of scanm = null;} else {m = new String(key.getData()); // = FrontierJMXTypes.toString(key.getData());
        }String[] arr = results.toArray(new String[results.size()]);CompositeData cd;try {cd = new CompositeDataSupport(/*FrontierJMXTypes.URI_LIST_DATA*/ null,new String[] { "list", "marker" },new Object[] { arr, m });} catch (OpenDataException e) {throw new IllegalStateException(e);}return cd;}

/*** Binding for use with BerkeleyDB-JE that uses Kryo serialization rather* than BDB's (custom version of) Java serialization.* * @contributor gojomo*/
public class KryoBinding<K> implements EntryBinding<K> {protected Class<K> baseClass;protected AutoKryo kryo = new AutoKryo(); protected ThreadLocal<WeakReference<ObjectBuffer>> threadBuffer = new ThreadLocal<WeakReference<ObjectBuffer>>() {@Overrideprotected WeakReference<ObjectBuffer> initialValue() {return new WeakReference<ObjectBuffer>(new ObjectBuffer(kryo,16*1024,Integer.MAX_VALUE));}};/*** Constructor. Save parameters locally, as superclass * fields are private. * * @param classCatalog is the catalog to hold shared class information** @param baseClass is the base class for serialized objects stored using* this binding*/@SuppressWarnings("unchecked")public KryoBinding(Class baseClass) {this.baseClass = baseClass;kryo.autoregister(baseClass);// TODO: reevaluate if explicit registration should be requiredkryo.setRegistrationOptional(true);}public Kryo getKryo() {return kryo;}private ObjectBuffer getBuffer() {WeakReference<ObjectBuffer> ref = threadBuffer.get();ObjectBuffer ob = ref.get();if (ob == null) {ob = new ObjectBuffer(kryo,16*1024,Integer.MAX_VALUE);threadBuffer.set(new WeakReference<ObjectBuffer>(ob));}return ob;        }/*** Copies superclass simply to allow different source for FastOoutputStream.* * @see com.sleepycat.bind.serial.SerialBinding#entryToObject*/public void objectToEntry(K object, DatabaseEntry entry) {entry.setData(getBuffer().writeObjectData(object));}@Overridepublic K entryToObject(DatabaseEntry entry) {return getBuffer().readObjectData(entry.getData(), baseClass);}
}