bklydatabase.java

短信开发用于文件交换处理转发的类模块

package com.pub.berkeleydb;

import java.util.Iterator;
import java.util.List;

import org.apache.log4j.Logger;
import org.apache.log4j.Priority;

import com.sleepycat.je.BtreeStats;
import com.sleepycat.je.Cursor;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseConfig;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.DeadlockException;
import com.sleepycat.je.Environment;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.OperationStatus;
import com.sleepycat.je.SecondaryDatabase;
import com.sleepycat.je.Transaction;

public class BklyDatabase {
	private static Logger log = Logger.getLogger(BklyDatabase.class);

	//建立berkeleydb的环境
	private BklyEnv _env = null;

	//建立berkeleydb数据库
	private Database _db = null;

	//是否可以复写
	private boolean _overWrite = true;

	//private List _idxs = new Vector();
	//操作的实例对象
	protected Class _recordClass = null;

	//操作的纪录对象
	protected IRecordCreator _rc = null;

	private int _retryTimes = 20;

	public int retryTimes() {
		return _retryTimes;
	}

	public void retryTimes(int t) {
		_retryTimes = t;
	}

	public BklyDatabase() {
	}

	//构造函数将操作的对象初始化
	public BklyDatabase(Class type, IRecordCreator rc) {
		_recordClass = type;
		_rc = rc;
	}

	//打开数据库环境
	public int open(String name, BklyEnv env) {
		return open(name, env, false, false, false, true, false);
	}

	public int open(String name, BklyEnv env, boolean mustExist,
			boolean readonly, boolean dup, boolean trans, boolean ow) {
		if (_recordClass == null || _rc == null) {
			return -1;
		}
		if (env == null) {
			return -2;
		}
		_env = env;

		DatabaseConfig dbCfg = new DatabaseConfig();
		dbCfg.setReadOnly(readonly);
		dbCfg.setAllowCreate(!mustExist);
		dbCfg.setTransactional(trans);
		dbCfg.setSortedDuplicates(dup);
		// if(cmptor != null) dbCfg.setBtreeComparator(cmptor);

		try {
			_db = env.getOriginEnv().openDatabase(null, name, dbCfg);
		} catch (Throwable err) {
			_db = null;
			log.error(null, err);
			return -3;
		}
		//将自身保存到对列中
		_env.addDatabase(this);

		_overWrite = ow;

		return 0;
	}

	public synchronized BklyEnv getBklyEnv() {
		return _env;
	}

	public synchronized Database getOriginDB() {
		return _db;
	}

	public synchronized Environment getOriginEnv() {
        try
        {
            _env.getOriginEnv().evictMemory();
        }
        catch(Exception ex)
        {
            ex.printStackTrace();
        }
		return _env.getOriginEnv();
	}
	
//	public synchronized Cursor getCursor(Transaction txn,CursorConfig cursorConfig) throws DatabaseException{
//		synchronized(_db){
//			return _db.openCursor(txn, null);
//		}
//	}

	public long count() {
		try {
			return _db != null ? ((BtreeStats) _db.getStats(null))
					.getLeafNodeCount() : 0;
		} catch (Throwable err) {
		}
		return -1;
	}

	public void close() {
		if (_db == null) {
			return;
		}

		List l = null;
		try {
			l = _db.getSecondaryDatabases();
		} catch (Throwable err) {
			log.error(null, err);
		}
		if (l != null) {
			Iterator it = l.iterator();
			while (it.hasNext()) {
				try {
					((SecondaryDatabase) it.next()).close();
				} catch (Throwable err) {
					log.error(null, err);
				}
			}
		}

		try {
			_db.close();
			_db = null;
		} catch (Throwable err) {
			log.error(null, err);
			_db = null;
		}
		_env.delDatabase(this);
	}

	// DatabaseEntry for key
	private static ThreadLocal tlPutKDE = new ThreadLocal() {
		protected Object initialValue() {
			return new DatabaseEntry();
		}
	};

	public static DatabaseEntry tmpPutKDE() {
		return (DatabaseEntry) tlPutKDE.get();
	}

	// DatabaseEntry for value
	private static ThreadLocal tlPutVDE = new ThreadLocal() {
		protected Object initialValue() {
			return new DatabaseEntry();
		}
	};

	public static DatabaseEntry tmpPutVDE() {
		return (DatabaseEntry) tlPutVDE.get();
	}

	/////////////////////////////////////////

	// DatabaseEntry for key
	private static ThreadLocal tlPopKDE = new ThreadLocal() {
		protected Object initialValue() {
			return new DatabaseEntry();
		}
	};

	public static DatabaseEntry tmpPopKDE() {
		return (DatabaseEntry) tlPopKDE.get();
	}

	// DatabaseEntry for value
	private static ThreadLocal tlPopVDE = new ThreadLocal() {
		protected Object initialValue() {
			return new DatabaseEntry();
		}
	};

	public static DatabaseEntry tmpPopVDE() {
		return (DatabaseEntry) tlPopVDE.get();
	}

	/////////////////////////////////////////

	// DatabaseEntry for key
	private static ThreadLocal tlIndexKDE = new ThreadLocal() {
		protected Object initialValue() {
			return new DatabaseEntry();
		}
	};

	public static DatabaseEntry tmpIndexKDE() {
		return (DatabaseEntry) tlIndexKDE.get();
	}

	// DatabaseEntry for value
	private static ThreadLocal tlIndexVDE = new ThreadLocal() {
		protected Object initialValue() {
			return new DatabaseEntry();
		}
	};

	public static DatabaseEntry tmpIndexVDE() {
		return (DatabaseEntry) tlIndexVDE.get();
	}

	/////////////////////////////////////////

	//DatabaseEntry  for indexKey
	private static ThreadLocal tlIKDE = new ThreadLocal() {
		protected Object initialValue() {
			return new DatabaseEntry();
		}
	};

	public static DatabaseEntry tmpIKDE() {
		return (DatabaseEntry) tlIKDE.get();
	}

	/**
	 * @param r IRecord
	 * @return int
	 * 1--duplicate
	 * 0--success
	 * -1--record class error
	 * -2--put fail
	 * -3--exception
	 */
	public int put(Object r) {

		if (!_recordClass.isInstance(r)) {
			return -1;
		}

		int retVal = 0;
		int retryTimes = 0;
		Transaction txn = null;
		DatabaseEntry k = tmpPutKDE();/*new DatabaseEntry();*/
		DatabaseEntry v = tmpPutVDE();/*new DatabaseEntry();*/
		OperationStatus ret = OperationStatus.NOTFOUND;

		synchronized (_rc) {
			try {
				_rc.toKey(r, k);
				_rc.toValue(r, v);
			} catch (Exception ex) {
				log.error(null, ex);
				retVal = -4;
			} catch (Throwable err) {
				log.error(null, err);
				retVal = -3;
			}
		}
		while (true) { //swan 2006-9-18

			try {
				txn = getOriginEnv().beginTransaction(null, null /*txnCfg*/);

				//判断是否要进行相同主键出现时的覆盖
				
				if (_overWrite) {
					ret = getOriginDB().put(txn, k, v);
				} else {
					ret = getOriginDB().putNoOverwrite(txn, k, v);
				}
				retVal = (ret == OperationStatus.SUCCESS) ? 0 : -2;
				if (retVal == 0) {
					txn.commit();
				} else {
					retVal = ret == OperationStatus.KEYEXIST ? -100 : retVal;
					txn.abort();
				}
				txn = null;
			} catch (DeadlockException ex) {
				retVal = -500;
			} //swan 2006-6-22
			catch (Throwable err) {
				log.error(null, err);
				retVal = -3;
			} finally {
				if (txn != null) {
					try {
						txn.abort();
						txn = null;
					} catch (Throwable err) {
					}
				}
			}

			retryTimes++;
			if (retVal != -500 || retryTimes >= _retryTimes) {
				break;
			}

			retVal = 0;
			txn = null;
			ret = OperationStatus.NOTFOUND;
		}

		k = null;
		v = null;

		//    if (retryTimes > 1 && log.isDebugEnabled()) {
		//      log.debug("put retryTimes=" + retryTimes + "\tresult=" + retVal);
		//    }

		return retVal;
	}

	/**
	 * @param r Object
	 * @param search boolean
	 * @param del boolean
	 * @return int
	 * 0--success
	 * -1--record class error
	 * -2--record to key error
	 * -3--no data
	 * -4--exception
	 */
	public int get(Object r, boolean search, boolean del) {

		if (!_recordClass.isInstance(r)) {
			return -1;
		}

		int retVal = 0;
		int retryTimes = 0;
		Cursor cursor = null;
		Transaction txn = null;
		DatabaseEntry key = tmpPopKDE();/*new DatabaseEntry();*/
		DatabaseEntry data = tmpPopVDE();/*new DatabaseEntry();*/
		OperationStatus ret = OperationStatus.NOTFOUND;

		try {
			synchronized (_rc) {
				if (search && _rc.toKey(r, key) != 0) {
					return -2;
				}
			}
		} catch (Throwable ex) {
			log.error("", ex);
			return -2;
		}

		while (true) { //swan 2006-9-18

			try {
				if (del) {
					txn = getOriginEnv().beginTransaction(null, null);
				}
				cursor = getOriginDB().openCursor(txn, null);
//				cursor = getCursor(txn, null);
				if (search) {
					ret = cursor.getSearchKey(key, data, LockMode.DEFAULT);
				} else {
					ret = cursor.getFirst(key, data, LockMode.DEFAULT);
				}

				if (ret == OperationStatus.SUCCESS && del) {
					cursor.delete();
				}

				cursor.close();
				cursor = null;

				if (ret == OperationStatus.SUCCESS) {
					if (txn != null) {
						txn.commit();
					}
					synchronized (_rc) {
						if (!search) {
							_rc.fromKey(r, key);
						}
						_rc.fromValue(r, data);
					}
				} else {
					if (txn != null) {
						txn.abort();
					}
					retVal = -3;
				}

				txn = null;
			} catch (DeadlockException ex) {
				retVal = -500;
			} //swan 2006-6-22
			catch (Throwable err) {
				log.error(null, err);
				retVal = -4;
			} finally {
				if (cursor != null) {
					try {
						cursor.close();
					} catch (Throwable err) {
					}
				}
				if (txn != null) {
					try {
						txn.abort();
					} catch (Throwable err) {
					}
				}
			}

			retryTimes++;
			if (retVal != -500 || retryTimes >= _retryTimes) {
				break;
			}

			retVal = 0;
			cursor = null;
			txn = null;
			ret = OperationStatus.NOTFOUND;
		}

		key = null;
		data = null;

		//    if (retryTimes > 1 && log.isDebugEnabled()) {
		//      log.debug("get retryTimes=" + retryTimes + "\tresult=" + retVal);
		//    }

		return retVal;
	}

	/**
	 * @param r Object
	 * @param idx BklyIndex
	 * @param del boolean
	 * @return int
	 * 0--success
	 * -1--record class error
	 * -2-- not find
	 * -3--delete fail
	 * -4--exception
	 */
	public int get(Object r, BklyIndex idx, boolean del) {
		if (!_recordClass.isInstance(r)) {
			return -1;
		}

		int retVal = 0;
		int retryTimes = 0;
		//建立业务
		Transaction txn = null;
		DatabaseEntry idxKey = tmpIKDE();/*new DatabaseEntry();*/
		DatabaseEntry key = tmpIndexKDE();/*new DatabaseEntry();*/
		DatabaseEntry data = tmpIndexVDE();/*new DatabaseEntry();*/
		OperationStatus ret = OperationStatus.NOTFOUND;

		idx.getIndexCreator().toIndexKey(r, idxKey);

		while (true) {

			try {
				//Cursor cur = idx.getOriginIndex().openSecondaryCursor(null, null);
				if (del) {
					txn = getOriginEnv().beginTransaction(null, null);
				}
				ret = idx.getOriginIndex().get(txn, idxKey, key, data,
						LockMode.DEFAULT);
				if (ret == OperationStatus.SUCCESS) {
					if (del) {
						ret = getOriginDB().delete(txn, key);
						retVal = (ret == OperationStatus.SUCCESS) ? 0 : -3; //delete error
					}
				} else {
					retVal = -2; //not find
				}

				if (txn != null) {
					if (retVal == 0) {
						txn.commit();
					} else {
						txn.abort();
					}
					txn = null;
				}
				synchronized (_rc) {
					if (retVal == 0) {
						_rc.fromKey(r, key);
						_rc.fromValue(r, data);
					}
				}
			} catch (DeadlockException ex) {
				retVal = -500;
			} //swan 2006-6-22
			catch (Throwable err) {
				if (log.isEnabledFor(Priority.ERROR)) {
					log.error(null, err);
				}
				retVal = -4; //exception
			} finally {
				if (txn != null) {
					try {
						txn.abort();
					} catch (Throwable err) {
					}
				}
			}

			retryTimes++;
			if (retVal != -500 || retryTimes >= _retryTimes) {
				break;
			}

			retVal = 0;
			txn = null;
			ret = OperationStatus.NOTFOUND;

			//key = null;
			//key = new DatabaseEntry();

			//data = null;
			//data = new DatabaseEntry();
		}
		idxKey = null;
		key = null;
		data = null;

		//    if (retryTimes > 1 && log.isDebugEnabled()) {
		//      log.debug("get[Index] retryTimes=" + retryTimes + "\tresult=" + retVal);
		//    }

		return retVal;
	}

	public IRecordCreator getRecordCeator() {
		return _rc;
	}

	public Class getRecordClass() {
		return _recordClass;
	}

	protected void finalize() {
		if (_db != null) {
			close();
		}
	}

}