Voovan是高性能异步通信、HTTP服务器和客户端通信、动态编译支持、数据库操作帮助类等工具的框架

package org.voovan.tools.collection; import org.rocksdb.*; import org.voovan.tools.TByte; import org.voovan.tools.TFile; import org.voovan.tools.TString; import org.voovan.tools.Varint; import org.voovan.tools.exception.ParseException; import org.voovan.tools.exception.RocksMapException; import org.voovan.tools.log.Logger; import org.voovan.tools.serialize.Serialize; import org.voovan.tools.serialize.TSerialize; import java.io.Closeable; import java.io.File; import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.util.*; import java.util.Comparator; import java.util.concurrent.ConcurrentHashMap; import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; /** * RocksDB 的 Map 封装 * * @author helyho * ignite-test Framework. * WebSite: https://github.com/helyho/ignite-test * Licence: Apache v2 License */ public class RocksMap<K, V> implements SortedMap<K, V>, Closeable { public enum Type { READ_ONLY, SECONDARY, TRANSACTION; } //--------------------- 公共静态变量 -------------------- static { RocksDB.loadLibrary(); } public final static String DEFAULT_DB_NAME = "Default"; //缓存 db 和他对应的 TransactionDB private static Map<String, RocksDB> ROCKSDB_MAP = new ConcurrentHashMap<String, RocksDB>(); //缓存 TransactionDB 和列族句柄的关系 private static Map<RocksDB, Map<String, ColumnFamilyHandle>> COLUMN_FAMILY_HANDLE_MAP = new ConcurrentHashMap<RocksDB, Map<String, ColumnFamilyHandle>>(); //数据文件的默认保存路径 private static String DEFAULT_DB_PATH = ".rocksdb" + File.separator; private static String DEFAULT_WAL_PATH = DEFAULT_DB_PATH + ".wal" + File.separator; private static String DEFAULT_BACKUP_PATH = DEFAULT_DB_PATH + ".backup" + File.separator; private static String DEFAULT_LOG_PATH = DEFAULT_DB_PATH + "logs" + File.separator; private static String DEFAULT_SECONDARY_DB_PATH = ".rocksdb.secondary" + File.separator; public static void setRootPath(String rootPath) { rootPath = rootPath.endsWith(File.separator) ? rootPath : rootPath + File.separator; DEFAULT_DB_PATH = rootPath; DEFAULT_WAL_PATH = rootPath + ".wal" + File.separator; DEFAULT_BACKUP_PATH = rootPath + ".backup" + File.separator; DEFAULT_LOG_PATH = rootPath + "logs" + File.separator; DEFAULT_SECONDARY_DB_PATH = TString.removeSuffix(DEFAULT_DB_PATH) + ".sec"; } /** * 获取默认数据存储路径 * @return 默认数存储据路径 */ public static String getDefaultDbPath() { return DEFAULT_DB_PATH; } /** * 设置默认数据存储据路径 * @param defaultDbPath 默认数存储据路径 */ public static void setDefaultDbPath(String defaultDbPath) { DEFAULT_DB_PATH = defaultDbPath.endsWith(File.separator) ? defaultDbPath : defaultDbPath + File.separator; DEFAULT_SECONDARY_DB_PATH = TString.removeSuffix(DEFAULT_DB_PATH) + ".sec" + File.separator;; } /** * 默认WAL数据存储据路径 * @return WAL数存储据路径 */ public static String getDefaultWalPath() { return DEFAULT_WAL_PATH; } /** * 设置WAL数据存储据路径 * @param defaultWalPath WAL数存储据路径 */ public static void setDefaultWalPath(String defaultWalPath) { DEFAULT_WAL_PATH = defaultWalPath.endsWith(File.separator) ? defaultWalPath : defaultWalPath + File.separator; } /** * 默认数据备份存储据路径 * @return WAL数存储据路径 */ public static String getDefaultBackupPath() { return DEFAULT_BACKUP_PATH; } /** * 设置数据备份存储据路径 * @param defaultBackupPath WAL数存储据路径 */ public static void setDefaultBackupPath(String defaultBackupPath) { DEFAULT_BACKUP_PATH = defaultBackupPath.endsWith(File.separator) ? defaultBackupPath : defaultBackupPath + File.separator;; } /** * 获取默认日志路径 * @return 默认数存储据路径 */ public static String getDefaultLogPath() { return DEFAULT_LOG_PATH; } /** * 设置默认日志路径 * @param defaultLogPath 默认数存储据路径 */ public static void setDefaultLogPath(String defaultLogPath) { DEFAULT_LOG_PATH = defaultLogPath.endsWith(File.separator) ? defaultLogPath : defaultLogPath + File.separator; } /** * 获取副本打开的路径 * @return 副本打开的路径 */ public static String getDefaultSecondaryDbPath() { return DEFAULT_SECONDARY_DB_PATH; } /** * 设置副本打开的路径 * @param defaultSecondaryDbPath 副本打开的路径 */ public static void setDefaultSecondaryDbPath(String defaultSecondaryDbPath) { DEFAULT_SECONDARY_DB_PATH = defaultSecondaryDbPath.endsWith(File.separator) ? defaultSecondaryDbPath : defaultSecondaryDbPath + File.separator; } /** * 根据名称获取列族 * @param rocksDB RocksDB 对象 * @param columnFamilyName 列族名称 * @return 列族句柄 */ private static ColumnFamilyHandle getColumnFamilyHandler(RocksDB rocksDB, String columnFamilyName) { return COLUMN_FAMILY_HANDLE_MAP.get(rocksDB).get(columnFamilyName); } /** * 关闭 RocksDB 极其句柄 * @param rocksDB RocksDB 对象 */ private static void closeRocksDB(RocksDB rocksDB) { for (ColumnFamilyHandle columnFamilyHandle : COLUMN_FAMILY_HANDLE_MAP.get(rocksDB).values()) { columnFamilyHandle.close(); } try { rocksDB.syncWal(); } catch (RocksDBException e) { e.printStackTrace(); } rocksDB.close(); COLUMN_FAMILY_HANDLE_MAP.remove(rocksDB); } //--------------------- 成员变量 -------------------- public transient DBOptions dbOptions; public transient ReadOptions readOptions; public transient WriteOptions writeOptions; public transient ColumnFamilyOptions columnFamilyOptions; public transient BackupEngineOptions backupEngineOptions; private transient RocksDB rocksDB; private transient ColumnFamilyHandle dataColumnFamilyHandle; private transient ThreadLocal<Transaction> threadLocalTransaction = new ThreadLocal<Transaction>(); private transient ThreadLocal<Integer> threadLocalSavePointCount = ThreadLocal.withInitial(()->new Integer(0)); private transient String dbName; private transient String dataPath; private transient String walPath; private transient String logPath; private transient String secondaryPath; private transient String backupPath; private transient String columnFamilyName; private transient Type type; private transient Boolean isDuplicate = false; private transient int transactionLockTimeout = 5000; private transient Serialize serialize; private boolean useSingleRemove = false; /** * 构造方法 */ public RocksMap() { this(null, null, null, null, null, null, null); } /** * 构造方法 * @param columnFamilyName 列族名称 */ public RocksMap(String columnFamilyName) { this(null, columnFamilyName, null, null, null, null, null); } /** * 构造方法 * @param dbName 数据库的名称, 基于数据保存目录的相对路径 * @param columnFamilyName 列族名称 */ public RocksMap(String dbName, String columnFamilyName) { this(dbName, columnFamilyName, null, null, null, null, null); } /** * 构造方法 * @param type Rocksdb 开启的模式 参考: RocksMap.Type */ public RocksMap(Type type) { t