海量数据处理的初探--亿级数据的离线计算--LevelDB简单封装

#include <iostream> #include <sstream> #include "define.h" #include "leveldb.h" static vector<CLevelDB *> g_vecLevelDB; CLevelDB::CLevelDB() { m_pLevelDB = NULL; m_bInit = false; } int CLevelDB::Init( const string &strPath ) { if ( m_bInit ) return SUCCESS; leveldb::Options options; options.create_if_missing = true; leveldb::Status status = leveldb::DB::Open( options, strPath, &m_pLevelDB ); if ( !status.ok() ) { return FAILURE; } m_bInit = true; return SUCCESS; } int CLevelDB::Set ( string &key, string &data ) { if ( m_pLevelDB == NULL ) return FAILURE; leveldb::Status status; status = m_pLevelDB->Put( leveldb::WriteOptions(), key, data ); if ( status.ok() ) { return SUCCESS; } if ( status.IsIOError() ) { return ERRNO_DB_IO; } return FAILURE; } int CLevelDB::Get ( string &key, string &data ) { if ( m_pLevelDB == NULL ) { return FAILURE; } leveldb::Status status; status = m_pLevelDB->Get( leveldb::ReadOptions(), key, &data ); if ( status.ok() ) { return SUCCESS; } if ( status.IsNotFound() ) { return ERRNO_ELEMENT_NOT_FOUND; } if ( status.IsIOError() ) { return ERRNO_DB_IO; } return FAILURE; } int CLevelDB::Del ( string &key ) { if ( m_pLevelDB == NULL ) { return FAILURE; } leveldb::Status status; status = m_pLevelDB->Delete( leveldb::WriteOptions(), key ); if ( status.ok() ) { return SUCCESS; } if ( status.IsNotFound() ) { return ERRNO_ELEMENT_NOT_FOUND; } if ( status.IsIOError() ) { return ERRNO_DB_IO; } return FAILURE; } uint32_t CLevelDB::Walk ( uint32_t counter, pcbFunc cbf ) { map<uint32_t,string> m; uint32_t todo = 0; uint32_t cnt = 0; uint32_t uin = 0; string data; leveldb::Iterator* it = m_pLevelDB->NewIterator( leveldb::ReadOptions() ); for ( it->SeekToFirst(); it->Valid(); it->Next() ) { todo++; data = it->value().ToString(); it->key().ToString().copy( (char*)&uin, sizeof(uint32_t) ); m.insert( pair<uint32_t,string>(uin,data) ); cnt++; if ( cnt == counter ) { usleep( 1000 ); // leveldb 非常消耗CPU，循环counter次让进程休眠1ms可降低CPU cbf( m ); cnt = 0; m.clear(); } } return todo; } int CLevelDBManager::Init( const string &strPathPrefix, const uint32_t dbNum ) { for ( uint32_t i = 0; i < dbNum; i++ ) { CLevelDB *db = NULL; db = new CLevelDB; if ( db == NULL ) { exit(1); } stringstream ss; ss<<strPathPrefix<<"/"<<i; string path = ss.str(); if ( db->Init( path ) == SUCCESS ) { g_vecLevelDB.push_back( db ); } } return g_vecLevelDB.size(); } CLevelDB * CLevelDBManager::GetDB ( uint32_t i ) { if ( i >= g_vecLevelDB.size() ) { return NULL; } return g_vecLevelDB[i]; } int CLevelDBManager::Fini () { int cnt = 0; vector<CLevelDB*>::iterator it; for ( it = g_vecLevelDB.begin(); it != g_vecLevelDB.end(); it++ ) { if ( *it != NULL ) { delete *it; *it = NULL; cnt++; } } return cnt; }