增加数据库连接池。

znone

2017-02-15 76949c0bf4e34f15a5f9b1c2b870477c7dbeb6b8

增加数据库连接池。

 include/qtl_database_pool.hpp

New file
@@ -0,0 +1,174 @@
#ifndef _QTL_DATABASE_POOL_H_
#define _QTL_DATABASE_POOL_H_

#include <memory>
#include <vector>
#include <atomic>
#include <thread>
#include <mutex>
#include <chrono>
#include <algorithm>

namespace qtl
{

template<typename Database>
class database_pool
{
public:
    typedef Database value_type;
    typedef std::shared_ptr<Database> pointer;

    database_pool()
        : m_trying_connection(false), m_stop_thread(false)
    {
    }

    virtual ~database_pool()
    {
        if(m_background_thread.joinable())
        {
            m_stop_thread=true;
            m_background_thread.join();
        }
        clear();
    }

    pointer get()
    {
        Database* db=popup();
        if(db==NULL && m_trying_connection==false)
            db=create_database();
        return pointer(db, [this](Database* db) {
            recovery(db);
        });
    }

    bool test_alive()
    {
        if(m_databases.empty()) 
            return false;
        std::unique_lock<std::mutex> lock(m_pool_mutex);
        auto it=m_databases.begin();
        while(it!=m_databases.end())
        {
            Database* db=*it;
            if(!db->is_alive())
            {
                delete db;
                it=m_databases.erase(it);
            }
            else
            {
                ++it;
            }
        }
        if(m_databases.empty())
        {
            lock.unlock();
            try_connect();
            return false;
        }
        else return true;
    }

private:
    std::vector<Database*> m_databases;
    std::mutex m_pool_mutex;
    std::atomic<bool> m_trying_connection;
    std::thread m_background_thread;
    bool m_stop_thread;

    virtual bool open_database(Database& db)=0;
    void recovery(Database* db)
    {
        if(db==NULL) return;
        if(db->is_alive())
        {
            std::lock_guard<std::mutex> lock(m_pool_mutex);
            m_databases.push_back(db);
        }
        else
        {
            delete db;
            {
                std::lock_guard<std::mutex> lock(m_pool_mutex);
                clear();
            }
            try_connect();
        }
    }

    Database* create_database()
    {
        Database* db=new Database;
        if(open_database(*db))
            return db;

        {
            std::lock_guard<std::mutex> lock(m_pool_mutex);
            clear();
        }
        try_connect();
        return NULL;
    }

    Database* popup()
    {
        Database* db=NULL;
        std::lock_guard<std::mutex> lock(m_pool_mutex);
        if(!m_databases.empty())
        {
            db=m_databases.back();
            m_databases.pop_back();
        }
        return db;
    }

    void try_connect()
    {
        if(m_trying_connection)
            return;

        m_trying_connection=true;
        try
        {
            m_background_thread=std::thread(&database_pool<Database>::background_connect, this);
        }
        catch (std::system_error&)
        {
            m_trying_connection=false;
        }
    }

    void background_connect()
    {
        Database* db=NULL;
        int interval=1;
        while(db==NULL && m_stop_thread==false)
        {
            db=create_database();
            if(db==NULL)
            {
                std::this_thread::sleep_for(std::chrono::seconds(interval));
                if(interval<60) interval<<=1;
            }
        }
        if(db)
        {
            recovery(db);
        }
        m_background_thread.detach();
        m_trying_connection=false;
    }

    void clear()
    {
        std::for_each(m_databases.begin(), m_databases.end(), std::default_delete<Database>());
        m_databases.clear();
    }
};

}

#endif //_QTL_DATABASE_POOL_H_

 include/qtl_mysql_pool.hpp

New file
@@ -0,0 +1,36 @@
#ifndef _QTL_MYSQL_POOL_H_
#define _QTL_MYSQL_POOL_H_

#include "qtl_database_pool.hpp"
#include "qtl_mysql.hpp"

namespace qtl
{

namespace mysql
{

class database_pool : public qtl::database_pool<database>
{
public:
    database_pool() : m_port(0) { }
    virtual ~database_pool() { }
    virtual bool open_database(database& db) override
    {
        db.charset_name("utf8");
        return db.open(m_host.data(), m_user.data(), m_password.data(), m_database.data(), 0, m_port);
    }

protected:
    std::string m_host;
    unsigned short m_port;
    std::string m_database;
    std::string m_user;
    std::string m_password;
};

}

}

#endif //_QTL_MYSQL_POOL_H_

 include/qtl_sqlite.hpp

@@ -423,6 +423,19 @@
    {
        simple_execute("ROLLBACK TRANSACTION");
    }

    bool is_alive()
    {
#ifdef _WIN32
        return true;
#else
        int has_moved=0;
        int result=sqlite3_file_control(m_db, NULL, SQLITE_FCNTL_HAS_MOVED, &has_moved);
        if(result!=SQLITE_OK)
            throw sqlite::error(result);
        return has_moved==0;
#endif //_WIN32
    }
    const char* errmsg() const { return sqlite3_errmsg(m_db); }
    int error() const { return sqlite3_errcode(m_db); }
    uint64_t insert_id() { return sqlite3_last_insert_rowid(m_db); }

 include/qtl_sqlite_pool.hpp

New file
@@ -0,0 +1,40 @@
#ifndef _QTL_SQLITE_POOL_H_
#define _QTL_SQLITE_POOL_H_

#include "qtl_sqlite.hpp"
#include "qtl_database_pool.hpp"

namespace qtl
{

namespace sqlite
{

class database_pool : public qtl::database_pool<database>
{
public:
    database_pool() : m_flags(SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE) { }
    virtual bool open_database(database& db) override
    {
        try
        {
            db.open(m_filename.data(), m_flags);
            return true;
        }
        catch (error& e)
        {
            return false;
        }
    }

protected:
    std::string m_filename;
    int m_flags;
};

}

}

#endif //_QTL_SQLITE_POOL_H_


 test/test_sqlite.mak

@@ -5,7 +5,7 @@
OBJ=TestSqlite.o sqlite3.o md5.o
CFLAGS=-g -D_DEBUG -O2 -I. -I/usr/local/include 
CXXFLAGS=-I../include -std=c++11
LDFLAGS= -L/usr/local/lib -ldl -lcpptest -lpthread
LDFLAGS= -L/usr/local/lib -pthread -ldl -lcpptest

all : $(TARGET)