bdbcache.java

高效海量访问系统

package com.cache.bdb;

import java.util.HashMap;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.cache.behavior.ICache;
import com.sleepycat.bind.EntryBinding;
import com.sleepycat.bind.serial.ClassCatalog;
import com.sleepycat.bind.serial.SerialBinding;
import com.sleepycat.collections.IteratorIndex;
import com.sleepycat.collections.StoredIterator;
import com.sleepycat.collections.StoredMap;
import com.sleepycat.je.Database;

/**
 * @author Administrator
 *
 * TODO To change the template for this generated type comment go to
 * Window - Preferences - Java - Code Style - Code Templates
 */
public class BDBCache implements ICache {
	private static final Log log = LogFactory.getLog(BDBCache.class);

	private StoredMap cache;
	private Database database;
	private ClassCatalog catalog;
    private Class keyClass;
    private Class valueClass;

    private Object mutex = new Object(); //同步信号量，用于保证多线程顺序读取记录
    private int index; //current record index


	// for IoC
	BDBCache () {
	}

	BDBCache ( Database db , ClassCatalog cc , Class keyClass, Class valueClass) {
		this.database = db;
		this.catalog = cc;
        this.keyClass = keyClass;
        this.valueClass = valueClass;
        this.validateCache();
	}

	public Object get( Object key ) {
        this.validateCache();
		Object ret = null;
		if ( cache!=null ) {
			ret = cache.get(key);
		}
		return ret;
	}

	public void put( Object key , Object value ) {
        this.validateCache();
        if (cache != null) {
          cache.put(key, value);
        }
	}

    /**
     * 取指定数量的记录
     * 读取策略：首先保证多线程顺序读取cache，保证不同线程不会读取相同的记录。如果已经读取
     * 完所有记录，则将读取指针回溯，从首记录重新开始读取。该策略主要是为了保证被读取后未被
     * 审的贴子能够在一段时间后能够再次得到审帖的机会。
     * @param count int 记录数量
     * @return Map <key, object>map序列
     */
    public Map next(int size) {
      this.validateCache();
      Map recs = new HashMap();
      //检查iterator是否存在，不存在则创建
      StoredIterator keyIterator = (StoredIterator) cache.keySet().iterator();
      //取记录
      if (cache != null) {
        //同步对记录的访问,不同线程必须顺序读取记录,避免重复读相同记录
        synchronized (mutex) {
          Object key;
          Object value;
          int i = 0;
          //将指针移动到指定当前记录位置
          IteratorIndex.moveToIndex(keyIterator, index);
          //读取记录
          while (keyIterator.hasNext() && (i < size)) {
            key = keyIterator.next();
            value = cache.get(key);
            if (value != null) {
              recs.put(key, cache.get(key));
              i++;
            }
            index++; //记录位置
          }

          //检查iterator指针是否指向列表尾部，如果是则复位。
          if (i < size) {
            //关闭原Iterator
            StoredIterator.close(keyIterator);
            //复位
            index = 0;
            //重新生成新的Iterator,记录指针指向首记录
            keyIterator = (StoredIterator) cache.keySet().iterator();
            log.debug("reset iterator, records maybe read repeatable...");
            //继续取记录(新取到的记录可能是以前取出过但未被审核的)
            while (keyIterator.hasNext() && (i < size)) {
              key = keyIterator.next();
              value = cache.get(key);
              if (value != null) {
                recs.put(key, cache.get(key));
                i++;
              }
              index++; //记录位置
              log.debug("key: " + key);
            }
          }
        }
      }
      //关闭Iterator， StoredIterator需要显式关闭
      StoredIterator.close(keyIterator);

      return recs;
    }

	public void remove( Object key ) {
        this.validateCache();
		if ( cache!=null ) {
			cache.remove(key);
		}
	}

	public void removeAll() {
        this.validateCache();
		cache.clear();
	}

	public boolean containsKey( Object key ) {
        this.validateCache();
		boolean ret = false;
		if ( cache!=null ) {
			ret = cache.containsKey(key);
		}
		return ret;
	}

	public boolean containsValue( Object value ) {
        this.validateCache();
		boolean ret = false;
		if ( cache!=null ) {
			ret = cache.containsValue(value);
		}
		return ret;
	}

    /**
     * 如果不存在则建立cache
     */
    private void validateCache() {
      if (cache == null) {
        synchronized (this) {
          if (cache == null) {
            EntryBinding keyBinding = new SerialBinding(catalog, keyClass);
            EntryBinding valueBinding = new SerialBinding(catalog, valueClass);
            StoredMap newCache = new StoredMap(database, keyBinding,
                                               valueBinding, true);
            cache = newCache;
          }
        }
      }
    }


	public void setDatabase( Database db ) {
		this.database = db;
	}

  public void setKeyClass(Class keyClass) {
    this.keyClass = keyClass;
  }

  public void setValueClass(Class valueClass) {
    this.valueClass = valueClass;
  }

  public Database getDatabase() {
		return database;
	}

  public Class getKeyClass() {
    return keyClass;
  }

  public Class getValueClass() {
    return valueClass;
  }

  public void setClassCatalog( ClassCatalog cc ) {
		this.catalog = cc;
	}

	public ClassCatalog getClassCatalog() {
		return catalog;
	}

	
}