mongodb连接池c++ 封装

linux平台下mongodb c++连接池封装，线程安全

//函数返回0:成功 >0 出错
class cmongo{
    public:
        //默认构造函数，默认连接数为1
        cmongo();
        //传入连接数到构造函数，默认连接数为size
        cmongo(int size);
        //析构函数
        ~cmongo();
    public:
        //设置tcp读写超时时间
        int set_wr_timeout(double t);
        //连接
        int conn(string mhost="127.0.0.1",int mport=27017);
        //设置db collection
        int setdb(string mdb,string mcollection);
 
        int setindex(string key);
        //查询
        int get(map<string,string>& out,vector<string> in,string key,string key_val);
        //投递一批要查询的字段，fields为要查询哪些字段
        int gets(map< string,map<string,string> >& rout,vector<string> fields,vector<string> in,string key);
        //dump key-value dumpkey对应一个value
        int dumpkey(map< string,string >& rout,string key,string val);
        //dump key->map<key,value> dumpkey对应一组value
        int dumpvals(map< string,map<string,string> >& rout,vector<string> in,string key);
        //写入
        int set(map<string,string> in,string key,string key_val);
        //批量写入
        //更新接口，批量更新key="id"
        //  "123456":<key,value>,<key,value>
        //  "123457":<key,value>,<key,value>
        int sets(map< string,map<string,string> > in,string key);
        //删除
        int remove(string key,string key_val);
    private:
        string doc;
        //tcp读写超时时间
        double wr_timeout;
        pthread_mutex_t _jobmux;
        sem_t _jobsem;
        map<DBClientConnection*,bool> _joblst;
        pthread_mutex_t _dbmux;
 
};
 
 
 
cmongo::cmongo(int size){
    //doc
    doc=string(DB_DB)+"."+string(DB_COLLECTION);
    wr_timeout=3;
    //最大连接0-200
    if(size<0){
        size=1;
    }
    if(size>200){
        size=200;
    }
    if(_joblst.size()>0){
        return;
    }
    bool auto_conn=true;
    pthread_mutex_init(&_jobmux,NULL);
    if((sem_init(&_jobsem,0,0))<0){
        return;
    }
    pthread_mutex_lock(&_jobmux);
    for(int i=0;i<size;++i){
        DBClientConnection* pconn = new DBClientConnection(auto_conn,0,wr_timeout);
        if(pconn != NULL){
            _joblst[pconn]=false;
        }
    }
    pthread_mutex_unlock(&_jobmux);
 
 
}
cmongo::~cmongo(){
    doc="";
    pthread_mutex_lock(&_jobmux);
    map<DBClientConnection*,bool>::iterator it=_joblst.begin();
    while(it != _joblst.end()){
        delete it->first;
        it++;
    }
    pthread_mutex_unlock(&_jobmux);
}
int cmongo::set_wr_timeout(double t){
    wr_timeout=t;
    return RET_OK;
}
int cmongo::conn(string mhost,int mport){
    pthread_mutex_lock(&_jobmux);
    map<DBClientConnection*,bool>::iterator it=_joblst.begin();
    while(it!=_joblst.end()){
        string errmsg="";
        HostAndPort hp(mhost,mport);
        if(!(it->first->connect(hp,errmsg))){
            cerr<<"connect mhost:"<<mhost<<" mport:"<<mport<<" msg:"<<errmsg<<endl;
            it->second=true;
        }
        sem_post(&_jobsem);
        it++;
    }
    pthread_mutex_unlock(&_jobmux);
    return RET_OK;
}
int cmongo::setdb(string mdb,string mcollection){
    if(mdb.empty() || mcollection.empty()){
        return RET_PARERR;
    }
    doc=mdb+"."+mcollection;
    return RET_OK;
}
int cmongo::setindex(string key){
    if(key.empty()){
        return RET_PARERR;
    }  
    sem_wait(&_jobsem);
    pthread_mutex_lock(&_jobmux);
    map<DBClientConnection*,bool>::iterator it=_joblst.begin();
    while(it!=_joblst.end()){
        if(it->second == false){
            it->second=true;
            break;
        }
        it++;
    }
    pthread_mutex_unlock(&_jobmux);
    string bindex="{"+key+":1}";
    it->first->ensureIndex(doc,fromjson(bindex));
 
    pthread_mutex_lock(&_jobmux);
    it->second=false;
    pthread_mutex_unlock(&_jobmux);
    sem_post(&_jobsem);
 
    return RET_OK;
}
//out为检索出来的key-value数据对应，in 为要检索的字段,key,key_value为要检索的条件,暂不支持多条件检索
//单列查询
int cmongo::get(map<string,string>& out,vector<string> in,string key,string key_val){
    //key key_val 要检索字段
    if(key.empty() || key_val.empty() || in.size()<=0){
        return RET_PARERR;
    }
    BSONObjBuilder b;
    for(vector<string>::iterator iter=in.begin();iter!=in.end();++iter){
        b.append(*iter,1);
    }
 
    sem_wait(&_jobsem);
    pthread_mutex_lock(&_jobmux);
    map<DBClientConnection*,bool>::iterator it=_joblst.begin();
    while(it!=_joblst.end()){
        if(it->second == false){
            it->second=true;
            break;
        }
        it++;
    }
    pthread_mutex_unlock(&_jobmux);
    BSONObj ob=b.obj();
 
    BSONObj p=it->first->findOne(doc,QUERY(key<<key_val),&ob);
 
    map<string,string> temp;
    for(vector<string>::iterator iter=in.begin();iter!=in.end();++iter){
        string mkey=*iter;
        temp[*iter]=p.getStringField(mkey.c_str());
    }
    out=temp;
 
    pthread_mutex_lock(&_jobmux);
    it->second=false;
    pthread_mutex_unlock(&_jobmux);
    sem_post(&_jobsem);
 
    return RET_OK;
}
//查询key为key的一批数据的 某些字段
//fields为要查询的字段集
//key="id" 值为in 一批key
//返回key->map<key,value>
int cmongo::gets(map< string,map<string,string> >& rout,vector<string> fields,vector<string> in,string key){
    if(key.empty()){
        return RET_PARERR;
    }
    sem_wait(&_jobsem);
    pthread_mutex_lock(&_jobmux);
    map<DBClientConnection*,bool>::iterator it=_joblst.begin();
    while(it!=_joblst.end()){
        if(it->second == false){
            it->second=true;
            break;
        }
        it++;
    }
    pthread_mutex_unlock(&_jobmux);
 
    BSONObjBuilder b;
    b.append(key,1);
    for(vector<string>::iterator iter=fields.begin();iter!=fields.end();++iter){
        b.append(*iter,1);
    }
 
    BSONObj p=b.obj();
    for(vector<string>::iterator iter2=in.begin();iter2!=in.end();++iter2){
        BSONObj ob=it->first->findOne(doc,QUERY(key<<*iter2),&p);
        map<string,string> temp;
        for(vector<string>::iterator iter=fields.begin();iter!=fields.end();++iter){
            string mkey=*iter;
            temp[*iter]=ob.getStringField(mkey.c_str());   
        }
        rout[ob.getStringField(key.c_str())]=temp;
    }
 
    pthread_mutex_lock(&_jobmux);
    it->second=false;
    pthread_mutex_unlock(&_jobmux);
    sem_post(&_jobsem);
 
    return RET_OK;
}
//dumpkey key-value 返回 key对应的val值
//key val
int cmongo::dumpkey(map< string,string >& rout,string key,string val){
    if(key.empty()){
        return RET_PARERR;
    }
    sem_wait(&_jobsem);
    pthread_mutex_lock(&_jobmux);
    map<DBClientConnection*,bool>::iterator it=_joblst.begin();
    while(it!=_joblst.end()){
        if(it->second == false){
            it->second=true;
            break;
        }
        it++;
    }
    pthread_mutex_unlock(&_jobmux);
 
    BSONObjBuilder b;
    b.append(key,1);
    if(!val.empty()){
        b.append(val,1);
    }
 
    BSONObj p=b.obj();
 
    pthread_mutex_lock(&_dbmux);
    auto_ptr<DBClientCursor> cursor = it->first->query(doc,Query(),0,0,&p);
    while(cursor->more()){
        BSONObj ob=cursor->next();
        rout[ob.getStringField(key.c_str())]=ob.getStringField(val.c_str());
    }
    pthread_mutex_unlock(&_dbmux);
 
    pthread_mutex_lock(&_jobmux);
    it->second=false;
    pthread_mutex_unlock(&_jobmux);
    sem_post(&_jobsem);
 
    return RET_OK;
}
//dumpkey key对应多个value
//key->map<key,value>.
//其实dumpvals接口完全可以包含dumpkey，为了方便运用独立出来
//out 返回的key 对应的map<key,value>
//in 每个key需要对应的返回哪些字段
//key="id"
int cmongo::dumpvals(map< string,map<string,string> >& rout,vector<string> in,string key){
    if(key.empty()){
        return RET_PARERR;
    }
    sem_wait(&_jobsem);
    pthread_mutex_lock(&_jobmux);
    map<DBClientConnection*,bool>::iterator it=_joblst.begin();
    while(it!=_joblst.end()){
        if(it->second == false){
            it->second=true;
            break;
        }
        it++;
    }
    pthread_mutex_unlock(&_jobmux);
 
    BSONObjBuilder b;
    b.append(key,1);
    for(vector<string>::iterator iter=in.begin();iter!=in.end();++iter){
        b.append(*iter,1);
    }
 
    BSONObj p=b.obj();
 
    pthread_mutex_lock(&_dbmux);
    auto_ptr<DBClientCursor> cursor = it->first->query(doc,Query(),0,0,&p);
    while(cursor->more()){
        BSONObj ob=cursor->next();
        map<string,string> temp;
        for(vector<string>::iterator iter=in.begin();iter!=in.end();++iter){
            string val=*iter;
            temp[val]=ob.getStringField(val.c_str());
        }
        rout[ob.getStringField(key.c_str())]=temp;
        temp.clear();
    }
    pthread_mutex_unlock(&_dbmux);
 
    pthread_mutex_lock(&_jobmux);
    it->second=false;
    pthread_mutex_unlock(&_jobmux);
    sem_post(&_jobsem);
 
    return RET_OK;
}
//更新接口，暂不支持key对应多条记录的更新
int cmongo::set(map<string,string> in,string key,string key_val){
    //如果map没有数据，返回参数错误
    if(in.size()<=0 || key.empty() || key_val.empty()){
        return RET_PARERR;
    }
    BSONObjBuilder b;
    map<string,string>::iterator iter;
    for(iter=in.begin();iter!=in.end();++iter){
        b.append(iter->first,iter->second);
    }
     
    sem_wait(&_jobsem);
    pthread_mutex_lock(&_jobmux);
    map<DBClientConnection*,bool>::iterator it=_joblst.begin();
    while(it!=_joblst.end()){
        if(it->second == false){
            it->second=true;
            break;
        }
        it++;
    }
    pthread_mutex_unlock(&_jobmux);
    BSONObj ob=b.obj();
    it->first->update(doc,QUERY(key<<key_val),BSON("$set"<<ob),true);
 
    int ret=RET_OK;
    string errmsg=it->first->getLastError();
    if(!errmsg.empty()){
        ret=RET_ERR;
    }
 
    pthread_mutex_lock(&_jobmux);
    it->second=false;
    pthread_mutex_unlock(&_jobmux);
    sem_post(&_jobsem);
 
    return ret;
}
//更新接口，批量更新key="id"
//  "123456":<key,value>,<key,value>
//  "123457":<key,value>,<key,value>
int cmongo::sets(map< string,map<string,string> > in,string key){
    //如果map没有数据，返回参数错误
    if(in.size()<=0 || key.empty() ){
        return RET_PARERR;
    }
 
    sem_wait(&_jobsem);
    pthread_mutex_lock(&_jobmux);
    map<DBClientConnection*,bool>::iterator it=_joblst.begin();
    while(it!=_joblst.end()){
        if(it->second == false){
            it->second=true;
            break;
        }
        it++;
    }
    pthread_mutex_unlock(&_jobmux);
 
    int ret=RET_OK;
    map< string,map<string,string> >::iterator iter;
    for(iter=in.begin();iter!=in.end();++iter){
        BSONObjBuilder b;
        for(map<string,string>::iterator iter2=iter->second.begin();iter2!=iter->second.end();++iter2){
            b.append(iter2->first,iter2->second);
        }
        BSONObj ob=b.obj();
        it->first->update(doc,QUERY(key<<iter->first),BSON("$set"<<ob),true);
        string errmsg=it->first->getLastError();
        if(!errmsg.empty()){
            ret=RET_ERR;
        }
    }
     
 
    pthread_mutex_lock(&_jobmux);
    it->second=false;
    pthread_mutex_unlock(&_jobmux);
    sem_post(&_jobsem);
 
    return ret;
}
//删除接口，删除记录 key=id key_val=587.即删除id="587"的记录
int cmongo::remove(string key,string key_val){
 
    if(key.empty() || key_val.empty()){
        return RET_PARERR;
    }
    sem_wait(&_jobsem);
    pthread_mutex_lock(&_jobmux);
    map<DBClientConnection*,bool>::iterator it=_joblst.begin();
    while(it!=_joblst.end()){
        if(it->second == false){
            it->second=true;
            break;
        }
        it++;
    }
    pthread_mutex_unlock(&_jobmux);
 
    it->first->remove(doc,BSON(key << key_val));
 
    pthread_mutex_lock(&_jobmux);
    it->second=false;
    pthread_mutex_unlock(&_jobmux);
    sem_post(&_jobsem);
 
    return RET_OK;
}
 

标签： linux 安全

版权申明：本站文章部分自网络，如有侵权，请联系：west999com@outlook.com
特别注意：本站所有转载文章言论不代表本站观点！
本站所提供的图片等素材，版权归原作者所有，如需使用，请与原作者联系。