队列很常见,但大部分的队列是将数据放入到内存.如果数据过多,就有内存溢出危险,而且长久占据着内存,也会影响性能.比如爬虫,将要抓取的URL放到内存,而URL过多,内存肯定要爆.在读Heritrix源码中,发现Heritrix是基于Bdb实现了一个持久化队列,于是我就将这块代码独立出来,平时使用也蛮爽的,现在拿出来共享.同时数据已经持久化,相比放在内存的一次性,可以循环累加使用.

      大家也知道BDB的高性能和嵌入式.但这个持久化队列我觉得比较适合单机.如果涉及到分布式,就不大适合了.毕竟分布式要通信,负载均衡,冗余等.可以用其他的数据库等替代.

      这里大概先说下实现原理,BDB是Key-Value型数据库,而队列是FIFO.所以这个持久化队列以位置作为BDB的Key,数据作为BDB的Value.然后用两个变量,分别记录队列两头的位置,也就是头部和尾部.当有数据插入的时候,就以尾部的位置为这个数据的Key.而当要取出数据时,以头部位置作为Key,获取这个Key的数据.原理大概如此,这个类也继承AbstractQueue,这里贴上代码.以下代码需引用bdb-je,common-io,junit.请在附件中下载.

         1、  自定义的BDB环境类,可以缓存StoredClassCatalog并共享

package com.guoyun.util;import java.io.File;import com.sleepycat.bind.serial.StoredClassCatalog;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseConfig;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.Environment;
import com.sleepycat.je.EnvironmentConfig;
/*** BDB数据库环境,可以缓存StoredClassCatalog并共享* * @contributor guoyun*/
public class BdbEnvironment extends Environment {StoredClassCatalog classCatalog; Database classCatalogDB;/*** Constructor* * @param envHome 数据库环境目录* @param envConfig config options  数据库换纪念馆配置* @throws DatabaseException*/public BdbEnvironment(File envHome, EnvironmentConfig envConfig) throws DatabaseException {super(envHome, envConfig);}/*** 返回StoredClassCatalog* @return the cached class catalog*/public StoredClassCatalog getClassCatalog() {if(classCatalog == null) {DatabaseConfig dbConfig = new DatabaseConfig();dbConfig.setAllowCreate(true);try {classCatalogDB = openDatabase(null, "classCatalog", dbConfig);classCatalog = new StoredClassCatalog(classCatalogDB);} catch (DatabaseException e) {// TODO Auto-generated catch blockthrow new RuntimeException(e);}}return classCatalog;}@Overridepublic synchronized void close() throws DatabaseException {if(classCatalogDB!=null) {classCatalogDB.close();}super.close();}}

 2.  基于BDB实现的持久化队列

package com.guoyun.util;import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.util.AbstractQueue;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicLong;import org.apache.commons.io.FileUtils;import com.sleepycat.bind.EntryBinding;
import com.sleepycat.bind.serial.SerialBinding;
import com.sleepycat.bind.serial.StoredClassCatalog;
import com.sleepycat.bind.tuple.TupleBinding;
import com.sleepycat.collections.StoredMap;
import com.sleepycat.collections.StoredSortedMap;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseConfig;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.DatabaseExistsException;
import com.sleepycat.je.DatabaseNotFoundException;
import com.sleepycat.je.EnvironmentConfig;
/*** 持久化队列,基于BDB实现,也继承Queue,以及可以序列化.但不等同于Queue的时,不再使用后需要关闭* 相比一般的内存Queue,插入和获取值需要多消耗一定的时间* 这里为什么是继承AbstractQueue而不是实现Queue接口,是因为只要实现offer,peek,poll几个方法即可,* 其他如remove,addAll,AbstractQueue会基于这几个方法去实现* * @contributor guoyun* @param <E>*/
public class BdbPersistentQueue<E extends Serializable> extends AbstractQueue<E> implementsSerializable {private static final long serialVersionUID = 3427799316155220967L;private transient BdbEnvironment dbEnv;            // 数据库环境,无需序列化private transient Database queueDb;             // 数据库,用于保存值,使得支持队列持久化,无需序列化private transient StoredMap<Long,E> queueMap;   // 持久化Map,Key为指针位置,Value为值,无需序列化private transient String dbDir;                 // 数据库所在目录private transient String dbName;				// 数据库名字private AtomicLong headIndex;                   // 头部指针private AtomicLong tailIndex;                   // 尾部指针private transient E peekItem=null;              // 当前获取的值/*** 构造函数,传入BDB数据库* * @param db* @param valueClass* @param classCatalog*/public BdbPersistentQueue(Database db,Class<E> valueClass,StoredClassCatalog classCatalog){this.queueDb=db;this.dbName=db.getDatabaseName();headIndex=new AtomicLong(0);tailIndex=new AtomicLong(0);bindDatabase(queueDb,valueClass,classCatalog);}/*** 构造函数,传入BDB数据库位置和名字,自己创建数据库* * @param dbDir* @param dbName* @param valueClass*/public BdbPersistentQueue(String dbDir,String dbName,Class<E> valueClass){headIndex=new AtomicLong(0);tailIndex=new AtomicLong(0);this.dbDir=dbDir;this.dbName=dbName;createAndBindDatabase(dbDir,dbName,valueClass);}/*** 绑定数据库* * @param db* @param valueClass* @param classCatalog*/public void bindDatabase(Database db, Class<E> valueClass, StoredClassCatalog classCatalog){EntryBinding<E> valueBinding = TupleBinding.getPrimitiveBinding(valueClass);if(valueBinding == null) {valueBinding = new SerialBinding<E>(classCatalog, valueClass);   // 序列化绑定}queueDb = db;queueMap = new StoredSortedMap<Long,E>(db,                                             // dbTupleBinding.getPrimitiveBinding(Long.class),   //KeyvalueBinding,                                   // Valuetrue);                                          // allow write}/*** 创建以及绑定数据库* * @param dbDir* @param dbName* @param valueClass* @throws DatabaseNotFoundException* @throws DatabaseExistsException* @throws DatabaseException* @throws IllegalArgumentException*/private void createAndBindDatabase(String dbDir, String dbName,Class<E> valueClass) throws DatabaseNotFoundException,DatabaseExistsException,DatabaseException,IllegalArgumentException{File envFile = null;EnvironmentConfig envConfig = null;DatabaseConfig dbConfig = null;Database db=null;try {// 数据库位置envFile = new File(dbDir);// 数据库环境配置envConfig = new EnvironmentConfig();envConfig.setAllowCreate(true);envConfig.setTransactional(false);// 数据库配置dbConfig = new DatabaseConfig();dbConfig.setAllowCreate(true);dbConfig.setTransactional(false);dbConfig.setDeferredWrite(true);// 创建环境dbEnv = new BdbEnvironment(envFile, envConfig);// 打开数据库db = dbEnv.openDatabase(null, dbName, dbConfig);// 绑定数据库bindDatabase(db,valueClass,dbEnv.getClassCatalog());} catch (DatabaseNotFoundException e) {throw e;} catch (DatabaseExistsException e) {throw e;} catch (DatabaseException e) {throw e;} catch (IllegalArgumentException e) {throw e;}}/*** 值遍历器*/@Overridepublic Iterator<E> iterator() {return queueMap.values().iterator();}/*** 大小*/@Overridepublic int size() {synchronized(tailIndex){synchronized(headIndex){return (int)(tailIndex.get()-headIndex.get());}}}/*** 插入值*/@Overridepublic boolean offer(E e) {synchronized(tailIndex){queueMap.put(tailIndex.getAndIncrement(), e);   // 从尾部插入}return true;}/*** 获取值,从头部获取*/@Overridepublic E peek() {synchronized(headIndex){if(peekItem!=null){return peekItem;}E headItem=null;while(headItem==null&&headIndex.get()<tailIndex.get()){ // 没有超出范围headItem=queueMap.get(headIndex.get());if(headItem!=null){peekItem=headItem;continue;} headIndex.incrementAndGet();    // 头部指针后移}return headItem;}}/*** 移出元素,移出头部元素*/@Overridepublic E poll() {synchronized(headIndex){E headItem=peek();if(headItem!=null){queueMap.remove(headIndex.getAndIncrement());peekItem=null;return headItem;}}return null;}/*** 关闭,也就是关闭所是用的BDB数据库但不关闭数据库环境*/public void close(){try {if(queueDb!=null){queueDb.sync();queueDb.close();}} catch (DatabaseException e) {// TODO Auto-generated catch blocke.printStackTrace();} catch (UnsupportedOperationException e) {// TODO Auto-generated catch blocke.printStackTrace();}}/*** 清理,会清空数据库,并且删掉数据库所在目录,慎用.如果想保留数据,请调用close()*/@Overridepublic void clear() {try {close();if(dbEnv!=null&&queueDb!=null){dbEnv.removeDatabase(null, dbName==null?queueDb.getDatabaseName():dbName); dbEnv.close();}} catch (DatabaseNotFoundException e) {// TODO Auto-generated catch blocke.printStackTrace();} catch (DatabaseException e) {// TODO Auto-generated catch blocke.printStackTrace();} finally{try {if(this.dbDir!=null){FileUtils.deleteDirectory(new File(this.dbDir));}} catch (IOException e) {// TODO Auto-generated catch blocke.printStackTrace();}}}}

3. 测试类,测试数据准确性和性能

package com.guoyun.util;import java.io.File;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;import junit.framework.TestCase;public class BdbPersistentQueueTest extends TestCase{Queue<String> memoryQueue;Queue<String> persistentQueue;@Overrideprotected void setUp() throws Exception {super.setUp();memoryQueue=new LinkedBlockingQueue<String>();String dbDir="E:/java/test/bdbDir";File file=new File(dbDir);if(!file.exists()||!file.isDirectory()){file.mkdirs();}persistentQueue=new BdbPersistentQueue(dbDir,"pq",String.class);}@Overrideprotected void tearDown() throws Exception {super.tearDown();memoryQueue.clear();memoryQueue=null;persistentQueue.clear();persistentQueue=null;}/*** 排放值* @param queue* @return      排放的数据个数*/public int drain(Queue<String> queue){int count=0;while(true){try {queue.remove();count++;} catch (Exception e) {return count;}}}/*** * @param queue* @param size*/public void fill(Queue<String> queue,int size){for(int i=0;i<size;i++){queue.add(i+"");}}public void checkTime(int size){System.out.println("1.内存Queue插入和排空数据所耗时间");long time=0;long start=System.nanoTime();fill(memoryQueue,size);time=System.nanoTime()-start;System.out.println("\t填充 "+size+" 条数据耗时: "+(double)time/1000000+" 毫秒,单条耗时: "+((double)time/size)+" 纳秒");start=System.nanoTime();drain(memoryQueue);time=System.nanoTime()-start;System.out.println("\t排空 "+size+" 条数据耗时: "+(double)time/1000000+" 毫秒,单条耗时: "+((double)time/size)+" 纳秒");System.out.println("2.持久化Queue插入和排空数据所耗时间");start=System.nanoTime();fill(persistentQueue,size);time=System.nanoTime()-start;System.out.println("\t填充 "+size+" 条数据耗时: "+(double)time/1000000+" 毫秒,单条耗时: "+((double)time/size/1000000)+" 豪秒");start=System.nanoTime();drain(persistentQueue);time=System.nanoTime()-start;System.out.println("\t排空 "+size+" 条数据耗时: "+(double)time/1000000+" 毫秒,单条耗时: "+((double)time/size/1000)+" 豪秒");}/*** 十万数量级测试*/public void testTime_tenThousand(){System.out.println("========测试1000000(十万)条数据=================");checkTime(100000);}/*** 百万数量级测试*/public void testTime_mil(){System.out.println("========测试1000000(百万)条数据=================");checkTime(1000000);}/*** 千万数量级测试,注意要防止内存溢出*/public void testTime_tenMil(){System.out.println("========测试10000000(千万)条数据=================");checkTime(10000000);}/*** 测试队列数据准确性* @param queue* @param queueName* @param size*/public void checkDataExact(Queue<String> queue,String queueName,int size){if(queue.size()!=size){System.err.println("Error size of "+queueName);}String value=null;for(int i=0;i<size;i++){value=queue.remove();if(!((i+"").equals(value))){System.err.println("Error "+queueName+":"+i+"->"+value);}}}/*** 测试队列中数据的准确性,包括长度*/public void testExact(){int size=100;fill(memoryQueue,size);fill(persistentQueue,size);checkDataExact(memoryQueue,"MemoryQueue",100);checkDataExact(persistentQueue,"PersistentQueue",100);}}