c++线程数据库类内存混淆

我一直在从事一个项目，该项目接受来自各种主机的数据并将其注入数据库。该项目通过boost:：thread进行线程化，并使用mysql++执行实际的数据库注入。

最近，我遇到了一个严重的问题，我的线程获取了不正确的信息。例如：线程A将从服务器foo.b.bar.foobar.com接收数据，并正确确定其主机名，主机短名称为foo.bar，线程b将从服务器fi.b.ban.foobar.com收到数据，并准确确定其主机名称，主机短名为fi.ban。所有这些都会发生，没有问题。

然后它被传递到数据库模块，接下来我看到的是类似于"INSERT INTO foo_servers VALUES（'foo.bbar.foobar.com'，'fi.ban'，…"的内容，作为字符串传递到mysql。

所有服务器每五分钟发送一次数据，所以每五分钟就会创建大约1k个线程，而这总是我开始看到问题发生的时候。

这是数据库类标题：

// db.h
// MySQL database connections and DB search/parsing.
#ifndef __DBASE_H
#define __DBASE_H
// database conection details go here
//#include"serial.h"
class dbase{
public:
    dbase();
    void runQuery();
    bool runDriveQuery( const std::vector<std::string> &qvec );
    bool getQueryData( const std::string &Frstate, const std::string &Fstype, const std::string &Fsite, const std::string &Fhost, const std::string &Fatype, const std::string &Fsstate, const std::string &Fanum, const std::string &Fshanme, const std::string &Fmac, const int &vfill, const int &dnum );
private:
    void updateServer();
    void inServer();
    bool checkExists( const std::string &data, const std::string &table, const std::string &field, const std::string &f2, const std::string &d2 );
    void storeQueryData( const std::string &fsval, const int &posid ); // written
    std::vector<std::string> inval;
    std::string getLocFromSite( const std::string &site );
    void insert( const std::string &qstring );
    void update( const std::string &qquery );
};
extern dbase thread;
#endif

这是dbase.cpp:

// db.h
// MySQL database connections and DB search/parsing.
// the final step of the data flow - everything ends up in an indexable mysql database in this module.
#include<iostream>
#include<sstream>
#include<string>
#include<mysql++/mysql++.h>
#include<vector>
#include<iomanip>
#include"dbase.h"
#include"log.h"
bool dbase::runDriveQuery( const std::vector<std::string> &qvec ) {
bool ex = checkExists( qvec[0], "gd_drives", "host_shortname", "drive_label", qvec[1]);
std::stringstream qqquery;
if( ex ) {
qqquery << "UPDATE `" << dbname << "`.`gd_drives` SET spool_number=" << mysqlpp::quote_only << qvec[2];
qqquery << ", drive_status=" << mysqlpp::quote_only << qvec[3] <<", spool_status=" <<mysqlpp::quote_only;
qqquery << ", pending_sectors=" << mysqlpp::quote_only << qvec[6];
qqquery << ", reallocated_sectors=" << mysqlpp::quote_only << qvec[7];
qqquery << ";";
std::string fquery = qqquery.str();
qqquery.str("");
GDLogger.log( fquery, 1 );
update( fquery );
return true;
}
else {
qqquery << "INSERT INTO `" << dbname << "`.`gd_drives` VALUES ( " << mysqlpp::quote_only << "Hard Drive";
qqquery << ", " << "'" << qvec[0] << "/" << qvec[1] << "', " << mysqlpp::quote_only << inval[5];
qqquery << ", " << mysqlpp::quote_only << qvec[0] << ", " << mysqlpp::quote_only << qvec[1];
qqquery << ", " << mysqlpp::quote_only << "0";
qqquery << ", " << mysqlpp::quote_only << qvec[2];
qqquery << ", " << mysqlpp::quote_only << qvec[3];
qqquery << ", " << mysqlpp::quote_only << qvec[5] << ", " << mysqlpp::quote_only << qvec[4];
qqquery << ", " << mysqlpp::quote_only << qvec[6] << ", " << mysqlpp::quote_only << qvec[7];
qqquery << ", " << mysqlpp::quote_only << "0" << ", '1'";
qqquery << ");";
std::string ffquery = qqquery.str();
qqquery.str("");
GDLogger.log( ffquery, 1 );
insert( ffquery );
return true;
}
}
void dbase::runQuery() {
    GDLogger.log( "runQuery() called.", 0 );
    bool ex = checkExists( inval[1], "gd_servers", "hostname", "0", "0" );
    if ( ex == true ) {
        updateServer();
    }
    else {
        inServer();
    }
    GDLogger.log( "runQuery() completed.", 0 );
}
bool dbase::checkExists( const std::string &data, const std::string &table, const std::string &field, const std::string &f2, const std::string &d2 ) { // check if the received data matches anything already in the database
    GDLogger.log( "checkExists() called.", 0 );
    GDLogger.log ( "Checking if host already in Database...", 1 );
    mysqlpp::Connection con3( false );
    if ( con3.connect( dbname, dbhost, dbuser, dbpass ) ) {
        std::stringstream queryss;
        queryss<< "SELECT " << field << " FROM `" << dbname << "`.`" << table << "` WHERE " << field << " = " << mysqlpp::quote_only << data;
        if( f2 == "0" ) {
            queryss << ";";
        }
        else {
            queryss << "AND " << f2 << "=" << mysqlpp::quote_only << d2 << ";";
        }
        std::string qstr = queryss.str();
        GDLogger.log( qstr, 1 );
        mysqlpp::Query query = con3.query( qstr );
        mysqlpp::StoreQueryResult res = query.store();
        std::stringstream res1;
        for ( std::vector<mysqlpp::Row>::iterator it= res.begin(); it != res.end(); ++it ) {
            mysqlpp::Row row = *it;
            res1 << row[0];
        }
        std::string result = res1.str();
        GDLogger.log( result, 1 );
        if ( result == inval[1] ) {
            GDLogger.log( "Host exists in Database.", 1 );
            return true;
        }
        else {
            GDLogger.log( "Host does not exist in Database.", 1 );
            return false;
        }
    }
    GDLogger.log( "checkExists() completed.", 0 );
}
void dbase::updateServer() {
    std::stringstream uss;
    uss<< " UPDATE `" << dbname << "`.`gd_servers` SET asset_type=" << mysqlpp::quote_only << inval[0] << ", recover_status=" << mysqlpp::quote_only << inval[10] << ", gc_status=" << mysqlpp::quote_only << inval[11] << ", var_fill=" << mysqlpp::quote_only << 0 << ", open_ticket=" << mysqlpp::quote_only << "0" << ", last_report = NOW() WHERE hostname = " << mysqlpp::quote_only << inval[1] << ";";
    std::string upstr = uss.str();
    GDLogger.log( upstr, 1 );
    update( upstr );
}
void dbase::update( const std::string &qquery ) { // update the selected entry in the database
    GDLogger.log( "update() called.", 0 );
    GDLogger.log( "Updating database entry.", 1 );
    mysqlpp::Connection con2( false );
    if ( con2.connect( dbname, dbhost, dbuser, dbpass ) ) {
        mysqlpp::Query query = con2.query( qquery );
        mysqlpp::SimpleResult res = query.execute();
        if ( res ) {
            GDLogger.log( "Record updated sucessfully.", 1 );
        }
        else {
            GDLogger.log( "Error updating record.", 2 );
        }
    }
    else {
        GDLogger.log( "Error establishing connection to database.", 3 );
    }
    GDLogger.log( "update() completed.", 0 );
}
std::string dbase::getLocFromSite( const std::string &site ) { // use the site that is transmitted to determine what the region is
    GDLogger.log( "getLocFromSite() called.", 0 );
    mysqlpp::Connection conn( false );
    if ( conn.connect( dbname, dbhost, dbuser, dbpass ) ) {
        std::stringstream lfs;
        lfs<< "SELECT srg_code FROM `" << dbname << "`.`gd_sites` WHERE site_name = " << mysqlpp::quote_only << site << ";";
        std::string loc = lfs.str();
        GDLogger.log( loc, 1 );
        mysqlpp::Query query = conn.query( loc );
        if (mysqlpp::StoreQueryResult res = query.store() ) {
            std::stringstream res1;
            for ( std::vector<mysqlpp::Row>::iterator it= res.begin(); it != res.end(); ++it ) {
                mysqlpp::Row row = *it;
                res1 << row[0] << std::endl;
            }
            std::string result = res1.str();
            GDLogger.log( result, 1 );
            return result;
        }
        if ( conn.errnum() ) {
            GDLogger.log( "Error Received in fetching a row.", 2 );
            return "ERR";
        }
        else {
            GDLogger.log( "Unknown Error Occurred or Unhandled Exception.", 3 );
            return "ERR";
        }
    }
    else {
        GDLogger.log( "Unable to connect to database!", 4 );
        return "ERR";
    }
    GDLogger.log( "getLocFromSite() completed.", 0 );
}
void dbase::inServer() {
    std::stringstream iss;
    iss<< "INSERT INTO `" << dbname << "`.`gd_servers` VALUES( ";
    for( int i = 0; i < 13; i++ ) {
        iss << mysqlpp::quote_only << inval[i] << ", ";
    }
    iss << mysqlpp::quote_only << "0" << ", NOW() );";
    std::string is = iss.str();
    GDLogger.log( is, 1 );
    insert( is );
}
void dbase::insert( const std::string &qstring ) { // insert server info into the mysql database (only if not already present)
    GDLogger.log( "insert() called.", 0 );
    mysqlpp::Connection conn( false );
    if ( conn.connect( dbname, dbhost, dbuser, dbpass ) ) {
        mysqlpp::Query query = conn.query( qstring );
        mysqlpp::SimpleResult res = query.execute();
        if( res ) {
            GDLogger.log( "Data inserted successfully.", 1 );
        }
        else {
            GDLogger.log( "Failed to add data to database.", 2 );
        }
    }
    else {
        GDLogger.log( "Failed to connect to database. Please ensure mysql is running, and that your credentials are correct.", 3 );
    }
    GDLogger.log("insert() completed.", 0 );
}
bool dbase::getQueryData( const std::string &Frstate, const std::string &Fstype, const std::string &Fsite, const std::string &Fhost, const std::string &Fatype, const std::string &Fsstate, const std::string &Fanum, const std::string &Fshname, const std::string &Fmac, const int &vfill, const int &dnum ) {
    GDLogger.log( "getQueryData() called.", 0 ); // takes all the deserialized data from the serial class and sorts them into containers in preparation of db injection
    std::string Fshorthost = Fshname; // host shortname
    std::string Floc = getLocFromSite( Fsite ); // location, queried from database based on string
    std::string FOrt = "false"; // true/false value for whether there is an open RT for the server. default is false.
    std::stringstream ssfill;
    ssfill << vfill; // /ivar fill percentage
    std::string Fvfill = ssfill.str();
    std::stringstream ssdnum;
    ssdnum << dnum; // number of hard drives
    std::string Fdnum = ssdnum.str();
    std::string uid= ""; // queried from database, default is 0;
    if ( Fstype == "r" || Fstype == "h" || Fstype == "t" ) {
        std::stringstream ss;
        ss << Fstype << mysqlpp::quote << "_server";
        std::string sstype = ss.str();
        storeQueryData( sstype, 4 );
    }
    else {
        storeQueryData( Fstype, 4 );
    }
    int num = 5; // this is temporary until I write it's hook in the serial class.
    std::stringstream anum1;
    anum1 << num;
    GDLogger.log( "Retrieving values for storage in database.", 1 );
    std::string tanum = anum1.str(); // storing values in a vector for easier access
    storeQueryData( Fatype, 0 );
    storeQueryData( Fhost, 1 );
    storeQueryData( Fshorthost, 2 );
    storeQueryData( uid, 3 );
    storeQueryData( Fsite, 5 );
    storeQueryData( Floc, 6 );
    storeQueryData( Fmac, 7 );
    storeQueryData( Fdnum, 8 );
    storeQueryData( tanum , 9 );
    storeQueryData( Frstate, 10 );
    storeQueryData( Fsstate, 11 );
    storeQueryData( Fvfill, 12 );
    storeQueryData( FOrt, 13 );
    GDLogger.log( "getQueryData() completed.", 0 );
}
void dbase::storeQueryData( const std::string &fsval, const int &posid ) {
    GDLogger.log( "storeQueryData() called.", 0 );
    inval[posid] = fsval;
}
dbase::dbase() { // overrides default constructor for dbase class
    GDLogger.log( "Initializing Database Module.", 1 );
    GDLogger.log( "dbase() called.", 0 );
    // initialize the size of the query data storage vector
    inval.resize(16);
    // getQueryData(); // retrieve data to be entered into the database
}
dbase thread;

这里是线程的创建位置：

void server( boost::asio::io_service& io_service, short port ) {
    tcp::acceptor a( io_service, tcp::endpoint( tcp::v4(), port ) ); // starts
    for( ; ; ) {
        socket_ptr sock( new tcp::socket( io_service ) ); // creates a new socket for every connection and accepts
        a.accept( *sock );
        boost::thread t( boost::bind( session, sock ) ); // starts session in a new thread
     }
}

线程创建和传递到数据库模块之间发生的事情都通过了我在所述模块中构建的验证检查，但一旦它进入数据库模块，就好像它从其他线程获取数据。。。或者什么的。我已经梳理了这个代码好几个星期了，我根本不知道发生了什么。线程之间唯一的共享内存是socket_ptr，它是指向正在处理的原始帧缓冲区的指针，只被引用一次，然后被丢弃。。。

编辑：如果我试图在它的线程内实例化dbase，它会拒绝链接，说是未定义的引用。这里是它被实例化的地方：

bool handle_data( std::vector<int>& buffer_data ) { // data handling happens hhere
    GDLogger.log( "handle_data() called.", 0 );
    try {
        dbase thread; // dbase instantiates here
        boost::system::error_code ec; // declare an error object ot collect exceptions
        ser.readBits( buffer_data ); //read from buffer into saod object
        std::vector<int> dbuff;
        dbuff.resize(616);
        if( buffer_data.size() >=616 ) {
            //ser.readDVec( dbuff );
            for( int i = 0; i < 576; i++ ) {
                dbuff[i] = ser.Sdbits[i];
             }
         }
         int ddnum = ser.Sdnum;
         std::string dshname = ser.Sshname;
         ddrive dd( ddnum, dshname, dbuff ); // drive module instantiates here
         bool dsvalidated = ser.deSerialize(); // deserialize data
         if(!dsvalidated) { // self explanatory
             GDLogger.log( "Deserialized data validation returned as false.", 3 );
        }
        else { // I hate this function. Reads from the serial object to the DB object.
            thread.getQueryData( ser.Srstate, ser.Sstype, ser.Ssite, ser.Shname, ser.Satype, ser.Sservices, ser.Sanum, ser.Sshname, ser.Smac, ser.Svarfill, ser.Sdnum ); // handing all the data off to the database module.
            thread.runQuery(); // runs all the queries necessary to update or insert
            dd.readDriveData( dbuff );
        }
        if( !ec ) {
            GDLogger.log( "handle_data() call completed without error.", 0 ); // if ec is still empty, we're done!
            return true;
         }
         else {
              GDLogger.log( "handle_data() call unsuccsessful, errors occurred.", 3 ); // things broke~somewhere~, but not fatally
              return true;
          }
    }
    catch( std::exception& e ) { // catch allocation exceptions.
        std::stringstream errstr;
        errstr << "Exception in thread: " << e.what();
        std::string logentry = errstr.str();
        errstr.str("");
        GDLogger.log( logentry, 3 );
    }
}

从内部调用：

void session( socket_ptr sock ) { // the actual TCP session starts here
    bool inproc = false; // this will be set to true once input has been sucessfully processed.
    try {
        GDLogger.log( "Connection Established.", 1 );
        int icont = 0;
        for( ; ;) {
            char data[max_length];
            boost::system::error_code error;
            size_t length = sock->read_some( boost::asio::buffer( data ), error ); // reads the length of the buffer, then reads the buffer into the char* data
            char cdata[max_length];
            for( int i = 0; i < max_length; i++ ) {
                cdata[i] = data[i];
            }
            std::vector<int> idata; // our hero, the vector going into handle_data
            idata.resize( max_length );
            idata = sanData( cdata ); // sanitize
            bool dvalid = valSerialData( idata ); // validate
            if( !dvalid ) {
                GDLogger.log("Data rejected by server.", 3 );
            }
            if( dvalid ) {
                GDLogger.log("Valid Datastream accepted. Processing.", 1 );
                inproc = handle_data( idata );
            }
            if( inproc == true && error == boost::asio::error::eof ) { // if EOF reached and
                GDLogger.log( "Connection Terminated, data processed.", 1 ); // if handle_data success, close connection
                break;
            }
            else if( !inproc && error == boost::asio::error::eof && icont > 0) {
                GDLogger.log( "Connection Terminated before data was successfully processed.", 2 );
                break;
            }
            else if( error ) {
                throw boost::system::system_error( error ); //some other error
            }
            icont++; // to ensure that handle_data only gets called ONCE
        }
    }
    catch( std::exception& e ) {
        std::stringstream errstr;
        errstr << "Exception in thread: " << e.what() << "n";
        std::string logentry = errstr.str(); // catch all the lovely boost errors
        errstr.str("");
        GDLogger.log( logentry, 3 );
    }
    GDLogger.log( "Thread Exited.", 0 );
}

它抱怨的地方是另一个模块，驱动器模块，它读取串行驱动器数据：

bool ddrive::readDriveData( const vector<int> &data1 ) {
    GDLogger.log( "readDriveData() called.", 0 );
    int SequenceNumber = 0;
    vector<bool> validated;
    vector<bool> dbadded;
    vector<bool> rval;
    GDLogger.log( "Resizing vectors.", 0 );
    rval.resize(Ddnum);
    dbadded.resize( Ddnum );
    validated.resize( Ddnum );
    for( int i = 0; i < Ddnum; i++ ) {
        GDLogger.log( "Setting rval to false...", 0 );
        rval[i] = false;
    }
    for( int i = 0; i < Ddnum; i++ ) {
        for( int j = 0; j < vdlen; j++ ) {
            Ddcode[j] = data1[SequenceNumber];
            std::stringstream ff;
            ff << "SN: "<< SequenceNumber << " " << Ddcode[j];
            std::string mmm = ff.str();
            ff.str("");
            GDLogger.log( mmm, 0 );
            SequenceNumber++;
        }
        std::stringstream lstr;
        lstr << "Sequence Number: " << SequenceNumber;
        std::string logm = lstr.str();
        lstr.str("");
        GDLogger.log( logm, 0 );
        Ddpath = mapp.getKeyFromMap( Ddcode[0], Ddcode[1], 3 );
        Dspnum = readSpoolNumber( Ddcode[2], Ddcode[3] );
        Dhstatus = readDHState( Ddcode[4] );
        Dspfill = read3DigitVals( Ddcode[5], Ddcode[6], Ddcode[7] );
        Dspstate = readSpoolState( Ddcode[8] );
        Dpsec = read3DigitVals( Ddcode[9], Ddcode[10], Ddcode[11] );
        Drsec = read3DigitVals( Ddcode[12], Ddcode[13], Ddcode[14] );
        bool DDValid = true;
        if( DDValid ) {
            validated[i] = true;
            GDLogger.log( "Converting data to strings for database insertion.", 1 );
            std::stringstream qq;
            qq << Ddcode[2] << Ddcode[3];
            std::string qspnum = qq.str();
            GDLogger.log( qspnum, 0 );
            qq.str("");
            qq << Dspfill;
            std::string qspfill = qq.str();
            GDLogger.log( qspfill, 0 );
            qq.str("");
            qq << Dpsec;
            std::string qpsec = qq.str();
            GDLogger.log( qpsec, 0 );
            qq.str("");
            qq << Drsec;
            std::string qrsec = qq.str();
            GDLogger.log( qrsec, 0);
            qq.str("");
            GDLogger.log( Dhsname, 0 );
            qdata[0] = Dhsname;
            qdata[1] = Ddpath;
            qdata[2] = qspnum;
            qdata[3] = Dhstatus;
            qdata[4] = qspfill;
            qdata[5] = Dspstate;
            qdata[6] = qpsec;
            qdata[7] = qrsec;
            GDLogger.log( Dhsname, 0);
            GDLogger.log( Ddpath, 0 );
            dbadded[i] = thread.runDriveQuery( qdata ); // this is where it complains
        }
        else {
            GDLogger.log( "Drive datastream rejected by server.", 3 );
            validated[i] = false;
        }
    }
    bool valid = true;
    bool dbadd = true;
    GDLogger.log( "Validating that operation performed successfully.", 1 );
    for( int i = 0; i < Ddnum; i++ ) {
        if( !validated[i] ) {
            valid = false;
        }
        if( !dbadded[i] ) {
            dbadd = false;
        }
    }
    if( dbadd && valid ) {
        return true;
    }
}

更新：我意识到Anton在我睡觉的时候说了什么，并照他说的做了，而是改变了函数drive:：readDriveData，这样它就会把dbase的一个实例作为一个参数，比如：

bool ddrive::readDriveData( const vector<int> &data1, dbase db );

它进行编译。我看看现在会发生什么。