버클 리 db 사용

napoli 코드 에서 이 버클 리 db 를 발 견 했 습 니 다.버 전 은 3.2.43 이 고 코드 는 다음 과 같 습 니 다.
인터페이스 정 의 는 다음 과 같 습 니 다.

/**
 * Project: napoli.client
 * 
 * File Created at Sep 15, 2009
 * $Id$
 * 
 * Copyright 2008 Alibaba.com Croporation Limited.
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of
 * Alibaba Company. ("Confidential Information").  You shall not
 * disclose such Confidential Information and shall use it only in
 * accordance with the terms of the license agreement you entered into
 * with Alibaba.com.
 */
package com.alibaba.napoli.client.inner.persistencestore;

import java.io.Serializable;
import java.util.List;
import java.util.Map;

/**
 *         。
 * 
 * @author ding.lid
 * @author guolin.zhuanggl
 */
public interface PersistenceStore<T extends Serializable> {
    /**
     *       。            key。
     * 
     * @param data
     */
    void write(T data);

    /**
     *       。
     * 
     * @param dataList
     */
    void batchWrite(List<T> dataList);

    /**
     *      。<br>
     *            ，    {@link #batchRead(int)}  ，      。
     * 
     * @return      。  store     ，   <code>null</code>。
     */
    Map.Entry<String, T> read();

    /**
     *      。
     * 
     * @param count        
     * @return      。  store     ，     Map( size ==0)。
     */
    Map<String, T> batchRead(int count);

    /**
     *       。
     * 
     * @param key
     */
    void delete(String key);

    /**
     *       。
     * 
     * @param keys       Key
     */
    void delete(List<String> keys);
}

 
 
버클 리 db 의 실현 은 다음 과 같다.

/**
 * Project: napoli.client
 * 
 * File Created at Sep 15, 2009
 * $Id$
 * 
 * Copyright 2008 Alibaba.com Croporation Limited.
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of
 * Alibaba Company. ("Confidential Information").  You shall not
 * disclose such Confidential Information and shall use it only in
 * accordance with the terms of the license agreement you entered into
 * with Alibaba.com.
 */
package com.alibaba.napoli.client.inner.persistencestore.impl;

import java.io.File;
import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.Map.Entry;

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

import com.alibaba.napoli.client.inner.persistencestore.PersistenceStore;
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.je.Cursor;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseConfig;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.Environment;
import com.sleepycat.je.EnvironmentConfig;
import com.sleepycat.je.EnvironmentMutableConfig;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.OperationStatus;
import com.sleepycat.je.Transaction;
import com.sleepycat.je.TransactionConfig;

/**
 * @author guolin.zhuanggl
 * @author ding.lid
 */
public class BdbPersistenceStore<T extends Serializable> implements PersistenceStore<T> {
    private static final Log    log                  = LogFactory.getLog(BdbPersistenceStore.class);

    private final static String MESSAGE_DBNAME       = "MESSAGE_DB";
    private final static String MESSAGE_CLASS_DBNAME = "MESSAGE_CLASS_DB";

    private String              bdbStorePath;
    //     10M
    private long                bdbCheckpointBytes   = 10 * 1024 * 1024;
    //     5M
    private long                bdbCacheSize         = 5 * 1024 * 1024;

    private Environment         bdbEnvironment;
    private Database            bdb;
    private StoredClassCatalog  bdbClassCatalog;

    private final Class<T>      dataClass;

    public String getBdbStorePath() {
        return bdbStorePath;
    }

    /**
     *   BDB       。         ，      。
     * 
     * @param storePath BDB       
     */
    public void setBdbStorePath(final String storePath) {
        this.bdbStorePath = storePath;
    }

    public long getBdbCheckpointBytes() {
        return bdbCheckpointBytes;
    }

    /**
     *   DBD CheckpointBytes  。    10M。
     */
    public void setBdbCheckpointBytes(final long dbdCheckpointBytes) {
        this.bdbCheckpointBytes = dbdCheckpointBytes;
    }

    public long getBdbCacheSize() {
        return bdbCacheSize;
    }

    /**
     *   DBD CacheSize  。    5M。
     */
    public void setBdbCacheSize(final long dbdCacheSize) {
        this.bdbCacheSize = dbdCacheSize;
    }

    public void init() throws DatabaseException {
        if (null == bdbStorePath) {
            throw new IllegalStateException("Member bdbStorePath is null!");
        }

        open();
    }

    private void open() throws DatabaseException {
        final File bdbDir = new File(bdbStorePath);
        if (!bdbDir.exists()) {
            if (!bdbDir.mkdirs()) {
                throw new RuntimeException("Fail to create the store directory(" + bdbStorePath
                        + ") for bdb persistence store!");
            }
        }

        final boolean readOnly = false;

        final EnvironmentConfig envConfig = new EnvironmentConfig();
        envConfig.setReadOnly(readOnly);
        envConfig.setAllowCreate(true);
        envConfig.setTransactional(true); // must setting
        // checkpoint occupied after data file increase some bytes
        envConfig.setConfigParam("je.checkpointer.bytesInterval", String
                .valueOf(bdbCheckpointBytes));

        final EnvironmentMutableConfig envMutableConfig = new EnvironmentMutableConfig();
        envMutableConfig.setCacheSize(bdbCacheSize);

        bdbEnvironment = new Environment(bdbDir, envConfig);
        bdbEnvironment.setMutableConfig(envMutableConfig);

        final DatabaseConfig dbConfig = new DatabaseConfig();
        dbConfig.setReadOnly(readOnly);
        dbConfig.setAllowCreate(!readOnly);
        dbConfig.setSortedDuplicates(false);
        dbConfig.setTransactional(true);

        bdb = bdbEnvironment.openDatabase(null, MESSAGE_DBNAME, dbConfig);

        // create class info db
        final Database classDb = bdbEnvironment.openDatabase(null, MESSAGE_CLASS_DBNAME, dbConfig);
        bdbClassCatalog = new StoredClassCatalog(classDb);
    }

    public void close() throws DatabaseException {
        // close db first, then close enviroment
        if (null != bdb) {
            bdb.close();
            bdb = null;
        }

        if (null != bdbClassCatalog) {
            bdbClassCatalog.close();
            bdbClassCatalog = null;
        }

        if (bdbEnvironment != null) {
            bdbEnvironment.cleanLog();
            bdbEnvironment.close();

            bdbEnvironment = null;
        }
    }

    /**
     *       {@link #createBdbPersistenceStore(Class)}        。
     * 
     * @param clazz         class。
     */
    public BdbPersistenceStore(final Class<T> clazz) {
        dataClass = clazz;
    }

    /**
     *   BdbPersistenceStore    。          ，      。
     * 
     * @param <DT>        。
     * @param dataType         class。
     * @return           BdbPersistenceStore     。
     */
    public static <DT extends Serializable> BdbPersistenceStore<DT> createBdbPersistenceStore(
                                                                                              final Class<DT> dataType) {
        return new BdbPersistenceStore<DT>(dataType);
    }

    /**
     *   UUID     DatabaseEntry  Key。
     */
    private static DatabaseEntry generateKeyEntry() {
        final String uuid = UUID.randomUUID().toString();
        return generateKeyEntry(uuid);
    }

    /**
     *          DatabaseEntry  Key。
     */
    private static DatabaseEntry generateKeyEntry(final String key) {
        final DatabaseEntry entry = new DatabaseEntry();
        final EntryBinding keyBinding = TupleBinding.getPrimitiveBinding(String.class);
        keyBinding.objectToEntry(key, entry);
        return entry;
    }

    /**
     *  DatabaseEntry   Key
     */
    private static String restoreKey(final DatabaseEntry entry) {
        final EntryBinding keyBinding = TupleBinding.getPrimitiveBinding(String.class);
        return (String) keyBinding.entryToObject(entry);
    }

    private DatabaseEntry generateDataEntry(final Object object) {
        final DatabaseEntry entry = new DatabaseEntry();
        final EntryBinding dataBinding = new SerialBinding(bdbClassCatalog, dataClass);
        dataBinding.objectToEntry(object, entry);

        return entry;
    }

    /**
     *  DatabaseEntry     
     */
    private T restoreData(final DatabaseEntry entry) {
        final EntryBinding dataBinding = new SerialBinding(bdbClassCatalog, dataClass);
        @SuppressWarnings("unchecked")
        final T data = (T) dataBinding.entryToObject(entry);
        return data;
    }

    /**
     *       。            key。 <br>
     *     ，    ，   ！
     */
    // FIXME     ？           ？？
    public void write(final T object) {
        if (object != null) {
            try {
                bdb.put(null, generateKeyEntry(), generateDataEntry(object));

                if (log.isTraceEnabled()) {
                    log.trace("write object to db: " + object);
                }
            } catch (final DatabaseException e) {
                log.fatal("write message failed ", e);
            }
        }
    }

    /**
     *       。 <br>
     *     ，    ，   ！
     */
    // FIXME     ？           ？？
    public void batchWrite(final List<T> objectList) {
        for (final T obj : objectList) {
            write(obj);
        }
    }

    /**
     *    。
     * 
     * @return        。   store     、     ，   null。
     */
    public Entry<String, T> read() {
        SimpleEntry<String, T> entry = null;
        Cursor cursor = null;

        try {
            cursor = bdb.openCursor(null, null);
            final DatabaseEntry foundKey = new DatabaseEntry();
            final DatabaseEntry foundData = new DatabaseEntry();

            if (cursor.getNext(foundKey, foundData, LockMode.DEFAULT) == OperationStatus.SUCCESS) {
                entry = new SimpleEntry<String, T>(restoreKey(foundKey), restoreData(foundData));
            }

            if (log.isTraceEnabled()) {
                log.trace("read object from db");
            }
        } catch (final DatabaseException e) {
            log.error("Don't warry, read message from db error, " + "late read again: "
                    + e.getMessage());
        } finally {
            if (cursor != null) {
                try {
                    cursor.close();
                } catch (final DatabaseException e) {
                    // ignore
                }
            }
        }

        return entry;
    }

    /**
     *    。
     * 
     * @return        。   store     、     ，     Map（ size 0 Map）。
     */
    public Map<String, T> batchRead(final int count) {
        final Map<String, T> map = new HashMap<String, T>(count);
        Cursor cursor = null;

        try {
            cursor = bdb.openCursor(null, null);
            final DatabaseEntry foundKey = new DatabaseEntry();
            final DatabaseEntry foundData = new DatabaseEntry();

            int readedNum = 0;
            while (cursor.getNext(foundKey, foundData, LockMode.DEFAULT) == OperationStatus.SUCCESS
                    && readedNum < count) {
                map.put(restoreKey(foundKey), restoreData(foundData));
                readedNum++;
            }

            if (log.isTraceEnabled()) {
                log.trace("read object from db, count= " + readedNum);
            }
        } catch (final DatabaseException e) {
            log.error("Don't warry, read message from db error, " + "late read again: "
                    + e.getMessage());
        } finally {
            if (cursor != null) {
                try {
                    cursor.close();
                } catch (final DatabaseException e) {
                    // ignore
                }
            }
        }

        return map;
    }

    // FIXME             。
    public void delete(final String key) {
        if (key == null) {
            return;
        }

        Transaction transaction = null;
        try {
            final TransactionConfig txnConfig = new TransactionConfig();
            txnConfig.setSync(true);

            transaction = bdbEnvironment.beginTransaction(null, txnConfig);
            if (log.isTraceEnabled()) {
                log.trace("BDB: begin to transaction");
            }

            bdb.delete(transaction, generateKeyEntry(key));
            if (log.isTraceEnabled()) {
                log.trace("BDB: delete message key=" + key);
            }

            transaction.commit();
            transaction = null;
            if (log.isTraceEnabled()) {
                log.trace("BDB: end of transaction");
            }
        } catch (final DatabaseException e) {
            log.fatal("BDB: delete failed: " + e.getMessage());
        } catch (final Throwable t) {
            log.fatal("BDB: delete failed: " + t.getMessage());
        } finally {
            if (transaction != null) {
                try {
                    transaction.abort();
                } catch (final DatabaseException e1) {
                }
            }
        }
    }

    public void delete(final List<String> keys) {
        // TODO       ，         ！
        for (final String key : keys) {
            delete(key);
        }
    }
}