从0开始搭建编程框架——插件

Posted 2022-07-29 breaksoftware

        我将插件设计为两种类型。一种是框架自身携带的插件，用于增强其基础能力。一种是用户自定义插件，用于完成业务。本文将分别介绍在peleus框架下，这两种插件该怎么写。（转载请指明出于breaksoftware的csdn博客）

        本文谈下框架自身携带的插件怎么写。我们以操作mysql为例。

        第一步把mysqlclient开发环境给准备好。

Wget https://downloads.mysql.com/archives/get/file/mysql-connector-c-6.1.11-src.tar.gz .
tar -zxvf mysql-connector-c-6.1.11-src.tar.gz
cd mysql-connector-c-6.1.11-src
cmake -G "Unix Makefiles"
cmake -G "Unix Makefiles" -DCMAKE_INSTALL_PREFIX=/usr
make
sudo make install

        由于实际安装到系统中的头文件不全，我们还要从https://github.com/mysql/mysql-server/blob/5.6/include/hash.h把hash.h文件放到/usr/include下。

        因为mysql-connector-c是C语言编码，为了统一peleus下语言栈，我们将其封装成一个类

#include "mysql_connection.h"
#include <butil/logging.h>
#include <errmsg.h>
#include "mysql_error.h"

namespace peleus 
namespace plugins 
namespace mysql_conns 

using peleus::plugins::configure::mysql_connection;

MysqlConnection::MysqlConnection() 
    _conn = NULL;


MysqlConnection::~MysqlConnection() 
    _disconnect();


void MysqlConnection::init(const mysql_connection& conf) 
    _conf.CopyFrom(conf);


int MysqlConnection::mysql_real_query(const std::string& sql) 
    int ret = 0;
    if (!_conn) 
        ret = _reconnct();
        if (ret) 
            return ret;
        
    

    LOG(INFO) << "query sql:" << sql.c_str();
    const int max_retry = 2;
    for (auto i = 0; i < max_retry; i++) 
        ret = ::mysql_real_query(_conn, sql.c_str(), sql.length());
        if (0 == ret) 
            break;
        
        else if (i < max_retry - 1)
            LOG(INFO) << "mysql connection is lost.retry " << i;
            int ret_re = _reconnct();
            if (ret_re) 
                LOG(WARNING) << "mysql connection is lost.retry error" << ret_re;
                return ret_re;
            
        
        else if (ret != CR_SERVER_LOST && ret != CR_SERVER_GONE_ERROR) 
            LOG(ERROR) << "query error:" << ret << " sql:" << sql.c_str();
            return ret;
        
    

    return ret;


MYSQL_RES* MysqlConnection::mysql_store_result() 
    if (_conn) 
        return ::mysql_store_result(_conn);
    
    else 
        return NULL;
    


int MysqlConnection::_reconnct() 
    if (_conn) 
        _disconnect();
    
    return _connect();


int MysqlConnection::_connect() 
    _conn = mysql_init(NULL);
    if (!_conn) 
        return CR_OUT_OF_MEMORY;
    
    unsigned long time_out = _conf.read_timeout();
    unsigned long conn_out = _conf.conn_timeout();
    mysql_options(_conn, MYSQL_OPT_READ_TIMEOUT, (char*)&time_out);
    mysql_options(_conn, MYSQL_OPT_CONNECT_TIMEOUT, (char*)&conn_out);

    if (!mysql_real_connect(_conn,
            _conf.host().c_str(),
            _conf.user().c_str(),
            _conf.password().c_str(),
            _conf.dbname().c_str(),
            _conf.port(),
            NULL, 0))
    
        _disconnect();
        return CR_SERVER_GONE_ERROR;
    

    int ret = mysql_set_character_set(_conn, _conf.character_set().c_str());
    if (ret) 
        _disconnect();
    

    return ret;


void MysqlConnection::_disconnect() 
    if (_conn) 
        mysql_close(_conn);
        _conn = NULL;
    

        MysqlConnection类主要对外提供连接和查询数据库的功能，并支持断开重连功能。这是针对一个连接的，我们需要连接池来提高连接的使用率。

MysqlConnectionPool::MysqlConnectionPool() 


MysqlConnectionPool::~MysqlConnectionPool() 
    for (auto it = _unused.begin(); it != _unused.end(); it++) 
        if (*it) 
            delete *it;
            *it = NULL;
        
    


void MysqlConnectionPool::init(const peleus::plugins::
        configure::mysql_connection_pool& conf) 
    _conf.CopyFrom(conf);
    int size = _conf.size();
    if (size < 0) 
        _use_pool = false;
        return;
    

    _use_pool = true;
    pthread_mutex_init(&_mutex, NULL);
    pthread_mutex_lock(&_mutex);
    for (int i = 0; i< size; i++) 
        MysqlConnection* conn = new MysqlConnection();
        conn->init(_conf.conn_info());
        _unused.push_back(conn);
    
    pthread_mutex_unlock(&_mutex);


int MysqlConnectionPool::query(const std::string& sql, MysqlSession& session) 
    if (_use_pool) 
        return query_from_pool(sql, session);
    
    return query_once(sql, session);


void MysqlConnectionPool::recovery(MysqlConnection* conn) 
    if (_use_pool) 
        pthread_mutex_lock(&_mutex);
        _unused.push_back(conn);
        pthread_mutex_unlock(&_mutex);
    


int MysqlConnectionPool::query_once(const std::string& sql, MysqlSession& session) 
    MysqlConnection conn;
    conn.init(_conf.conn_info());
    int ret = conn.mysql_real_query(sql);
    session._res = conn.mysql_store_result();
    return ret;


int MysqlConnectionPool::query_from_pool(const std::string& sql, MysqlSession& session) 
    MysqlConnection* conn = NULL;
    pthread_mutex_lock(&_mutex);
    if (!_unused.empty()) 
        conn = _unused.back();
        _unused.pop_back();
    
    pthread_mutex_unlock(&_mutex);
    if (!conn) 
        LOG(ERROR) << "need more connection";
        return LACK_CONNECTION;
    

    int ret = conn->mysql_real_query(sql);
    session._res = conn->mysql_store_result();
    session._conn = conn;
    session._conn_pool = this;
    return ret;

        MysqlConnectionPool将空闲连接放在_unused中。当需要做查询操作时，便从_unused中取出一个连接使用。但是连接池里的连接不能交由用户管理，于是需要暴露query_from_pool方法，将连接隐藏起来。

        由于业务可能需要连接多个数据库，所以需要做一个针对多个数据库的连接管理器

MysqlConns::MysqlConns() 


MysqlConns::~MysqlConns() 
    for (auto it = _conns.begin(); it != _conns.end(); it++) 
        if (it->second) 
            delete it->second;
            it->second = NULL;
        
    


void MysqlConns::init(const char* conf_floder_path) 
    TraversalCallback traversal_floder = [this](const char* path) this->init(path);;
    TraversalCallback traversal_file = [this](const char* path) this->init_from_file(path);;
    TraversalFloder traversal;
    traversal.set_callback(traversal_floder, traversal_file, __FILE__);
    traversal.init(conf_floder_path);


void MysqlConns::init_from_file(const char* conf_path) 
    mysql_connection_pool config;
    bool suc = peleus::utils::file2conf(conf_path, config);
    if (!suc) 
        LOG(ERROR) << "init " << conf_path << " error";
        return;
    
    const std::string& conn_name = config.name();
    MysqlConnectionPool* conn_pool = init_conn_pool(config);
    if (!conn_pool) 
        LOG(ERROR) << "init mysql connection pool error:" << conn_name;
        return;
    
    _conns[conn_name] = conn_pool;


MysqlConnectionPool* MysqlConns::init_conn_pool(const mysql_connection_pool& conf) 
    MysqlConnectionPool* conn_pool = new MysqlConnectionPool();
    conn_pool->init(conf);
    return conn_pool;



int MysqlConns::query(const std::string& conn_name,
        const std::string& sql,
        MysqlSession& session)

    auto it = _conns.find(conn_name);
    if (it == _conns.end()) 
        LOG(ERROR) << "cant find mysql connection:" << conn_name.c_str();
        return NO_CONNECTION;
    
    if (!it->second) 
        LOG(ERROR) << "mysql connection pool is null";
        return NULL_CONNECTION;
    
    return it->second->query(sql, session);

        至此，我们将mysql最基础的操作部分给封装完毕。现在开始写插件

class MysqlVisitorService :
    public peleus::plugins::MysqlVisitorService,
    public peleus::entrance::Entrance

public:
    explicit MysqlVisitorService(const char* name);

public:
    virtual void query(::google::protobuf::RpcController* controller,
       const ::peleus::plugins::MysqlVisitorServiceRequest* request,
       ::peleus::plugins::MysqlVisitorServiceResponse* response,
       ::google::protobuf::Closure* done) override;
public:
    virtual void on_init(const char* conf_path) final;
    virtual void reset() final;
private:

    void parse_res(
            MYSQL_RES* res,
            ::peleus::plugins::MysqlVisitorServiceResponse* response);

    void parse_session(
            peleus::plugins::mysql_conns::MysqlSession& session,
            ::peleus::plugins::MysqlVisitorServiceResponse* response);
;

        MysqlVisitorService将作为内部服务的一个业务逻辑存在，同时它也是一个入口，所以继承于Entrance接口。MysqlVisitorService是protobuf产生的接口类

syntax="proto2";
option cc_generic_services = true;

package peleus.plugins;

message MysqlVisitorServiceRequest 
    required string conn_name = 1;
    required string command = 2;
;

message MysqlVisitorRow 
    repeated string column = 1;
;

message MysqlVisitorServiceResponse 
    required bool suc = 1;
    repeated MysqlVisitorRow rows = 2;
;

service MysqlVisitorService 
    rpc query(MysqlVisitorServiceRequest) returns (MysqlVisitorServiceResponse);
;

        MysqlVisitorService的业务代码就是query，其会调用MysqlConns类的方法去查询结果，然后解析和封装返回

MysqlVisitorService::MysqlVisitorService(const char* name) :
    peleus::entrance::Entrance(name) 


void MysqlVisitorService::query(::google::protobuf::RpcController* controller,
    const ::peleus::plugins::MysqlVisitorServiceRequest* request,
    ::peleus::plugins::MysqlVisitorServiceResponse* response,
    ::google::protobuf::Closure* done)

    ClosureGuard done_guard(done);
    const std::string& conn_name = request->conn_name();
    const std::string& command = request->command();

    MysqlSession session;
    int ret = MysqlConns::get_mutable_instance().query(conn_name, command, session);
    parse_session(session, response);
    response->set_suc(0 == ret ? true : false);


void MysqlVisitorService::parse_session(
        MysqlSession& session,
        ::peleus::plugins::MysqlVisitorServiceResponse* response)

    MYSQL_RES* res = session._res;
    if (!res) 
        return;
    

    int row_size = mysql_num_rows(res);
    for (int i = 0; i < row_size; i++) 
        parse_res(res, response);
    


void MysqlVisitorService::parse_res(
        MYSQL_RES* res,
        ::peleus::plugins::MysqlVisitorServiceResponse* response)

    MysqlVisitorRow* row = response->add_rows();
    MYSQL_ROW res_row = mysql_fetch_row(res);
    if (!row) 
        LOG(ERROR) << "add_rows return null";
        return;
    
    int field_size = mysql_num_fields(res);
    for (int i = 0; i < field_size; i++) 
        row->add_column(std::string(res_row[i]));
    

        最后在插件启动代码中加入注册方法就行了

extern "C" 
    bool plugins_main(const char*);
;

#define USE(x, y)   \\
    if (item_conf.name() == x)     \\
        y::get_mutable_instance().init(item_conf.conf_path().c_str());    \\
    

#define REG(x, y, z)    \\
    if (item_conf.name() == x)    \\
        return register_class<y>(z);    \\
    

bool load_moudle(const plugin_conf& item_conf) 
    if (!item_conf.on()) 
        return true;
    

    REG("mysql_visitor_service", MysqlVisitorService, "MysqlVisitorService");

    return true;


bool plugins_main(const char* conf_path) 
    plugins_conf config;
    bool suc = file2conf(conf_path, config);
    if (!suc) 
        LOG(ERROR) << "conv config error:" << conf_path;
        return false;
    

    size_t size = config.conf_size();

    for (size_t i = 0; i < size; i++) 
        const plugin_conf& item_conf = config.conf(i);
        if (!load_moudle(item_conf)) 
            break;
        
    

    return true;