一、设计

1.怎么落地

将日志落地到数据库，首先肯定是要建表的，而日志落地其实就是向对应表当中插入数据

那么怎么落地呢？
文件落地方式：virtual void log_fs(const char *data, size_t len) = 0;
因为文件是面向字节流的，所以这么落地日志是完全OK的

而数据库落地是向表当中插入数据，因此他的落地应该是这样的：

virtual void log_db(const LogMessage &message) = 0;

将一条LogMessage插入表当中

因此我们的LogSink日志落地基类无法满足数据库的落地需求，要不然就给他增加一个log_db函数
让子类选择性的对其中一个进行重写，另一个进行空实现，这是一种方式

不过不便于扩展，为了遵循高内聚，低耦合的程序设计原则，我们将LogSink分为两个日志基类：
FSLogSink和DBLogSink

新增加的MySQLSink和SqliteSink都继承于DBLogSink

2.落地的具体设计

我们的日志落地方式是允许用户灵活的进行格式控制的，这是为了让用户能够选择性的只记录自己想要的数据

而将日志落地到数据库，是非字符串形式的，因此格式化的意义不大

所以我们的日志落地到数据库当中是无需进行格式化的，这也印证了log_db的参数为何只需要一个LogMessage即可

virtual void log_db(const LogMessage &message) = 0;

所以我们再将日志落地到数据库时，是直接将所有字段全部都进行记录的

3.表的设计

1.MySQL

2.SQLite3

同样的，SQLite3当中LigData的创建也是如此，只不过具体细节要求不同：

二、数据库访问Helper的实现

1.需要事务，但是无需回滚，如何理解？

1.需要事务

2.无需回滚

不过我们贴心的给了大家savepoint和rollback的使用，大家可以使用

2.SqliteHelper

1.SQLite3常用接口介绍

SQLite3 官方文档

2.实现

class SqliteHelper
{
public:using SqliteCallback = int (*)(void *, int, char **, char **);SqliteHelper(const std::string &dbfile): _dbfile(dbfile), _handler(nullptr) {}~SqliteHelper(){close();}bool open(){if (sqlite3_open_v2(_dbfile.c_str(), &_handler, SQLITE_OPEN_CREATE | SQLITE_OPEN_READWRITE, nullptr) != SQLITE_OK){std::cout << "打开数据库失败 " << errmsg() << " _dbfile：" << _dbfile << "\n";return false;}return true;}void close(){if (_handler != nullptr){sqlite3_close_v2(_handler);_handler = nullptr;}}bool begin(){if (sqlite3_exec(_handler, "begin;", nullptr, nullptr, nullptr) != SQLITE_OK){std::cout << "SQLite3开启事务失败，" << errmsg() << "\n";return false;}return true;}bool exec(const std::string &sql, SqliteCallback cb, void *arg){if (sqlite3_exec(_handler, sql.c_str(), cb, arg, nullptr) != SQLITE_OK){std::cout << "执行sql语句：" << sql << " 失败，" << errmsg() << std::endl;return false;}return true;}bool commit(){if (sqlite3_exec(_handler, "commit;", nullptr, nullptr, nullptr) != SQLITE_OK){std::cout << "SQLite3提交事务失败，" << errmsg() << "\n";return false;}return true;}bool savePoint(const std::string &point){std::string savepoint_sql = "savepoint " + point + ";";if (sqlite3_exec(_handler, savepoint_sql.c_str(), nullptr, nullptr, nullptr) != SQLITE_OK){std::cout << "SQLite3 设置事务保存点失败：" << errmsg() << "\n";return false;}return true;}bool rollback(const std::string &point = ""){std::string rollback_sql = "rollback";if (!point.empty()){rollback_sql += " to " + point + ";";}if (sqlite3_exec(_handler, rollback_sql.c_str(), nullptr, nullptr, nullptr) != SQLITE_OK){std::cout << "SQLite3 事务回滚失败：" << errmsg() << "\n";return false;}return true;}std::string errmsg(){if (_handler != nullptr)return sqlite3_errmsg(_handler);elsereturn "sqlite3句柄为空";}private:std::string _dbfile;sqlite3 *_handler;
};

3.MySQLHelper

1.MySQL常用接口介绍

2.实现

我们要求用户将MySQL连接所需字段放到配置文件当中，将配置文件传递过来
我们内部进行解析

格式要求：一个字段占一行
以key:value的形式进行传递host:你MySQL服务器所在机器的IP地址
user:你的用户名
passwd:你的密码
db:你提前创建好并赋予权限了的数据库
port:你MySQL服务器的端口号

class MySQLHelper
{
public:MySQLHelper(const std::string &conf_file) : _handler(mysql_init(nullptr)), _conf_file(conf_file){}bool begin(){if (mysql_query(_handler, "begin;") != 0){std::cout << "mysql开启事务失败，" << errmsg() << "\n";return false;}return true;}bool commit(){if (mysql_query(_handler, "commit;") != 0){std::cout << "mysql提交事务失败，" << errmsg() << "\n";return false;}return true;}bool savePoint(const std::string &point){std::string savepoint_sql = "savepoint " + point + ";";if (mysql_query(_handler, savepoint_sql.c_str()) != 0){std::cout << "MySQL 设置事务保存点失败：" << errmsg() << "\n";return false;}return true;}bool rollback(const std::string &point = ""){std::string rollback_sql = "rollback";if (!point.empty()){rollback_sql += " to " + point + ";";}if (mysql_query(_handler, rollback_sql.c_str()) != 0){std::cout << "MySQL 事务回滚失败：" << errmsg() << "\n";return false;}return true;}bool open(){// 1. 读取配置文件，拿到host、user、passwd、db、portif (!load()){mysql_close(_handler);_handler = nullptr;return false;}// 2. 连接MySQL数据库_handler = mysql_real_connect(_handler, _conf_map["host"].c_str(), _conf_map["user"].c_str(),_conf_map["passwd"].c_str(), _conf_map["db"].c_str(), std::stoi(_conf_map["port"]), nullptr, 0);if (_handler == nullptr){std::cout << "MySQL连接失败, 原因: " << mysql_error(_handler) << "\n";mysql_close(_handler);_handler = nullptr;return false;}return true;}~MySQLHelper(){close();}void close(){if (_handler != nullptr){mysql_close(_handler);_handler = nullptr;}}bool exec(const std::string &sql){if (0 != mysql_query(_handler, sql.c_str())){std::cout << "sql语句执行失败: " << sql << " ,原因: " << errmsg() << std::endl;return false;}return true;}std::string errmsg(){if (_handler != nullptr)return mysql_error(_handler);elsereturn "mysql句柄为空";}private:bool load(){std::ifstream ifs(_conf_file);std::string line;while (std::getline(ifs, line)){size_t pos = line.find(':');if (pos == std::string::npos){std::cout << "MySQL配置文件解析失败,某一行不符合规范: " << line << "\n";return false;}_conf_map[line.substr(0, pos)] = line.substr(pos + 1);}return true;}MYSQL *_handler;std::string _conf_file;std::unordered_map<std::string, std::string> _conf_map;
};

4.FileHelper补充

FileHelper里面又加了一个createFile接口，用来创建文件

static bool createFile(const std::string &filename)
{// 1. 先看该文件是否存在if (exists(filename))return true;// 2. 写方式打开（即创建）// 这里以追加写打开，防止多线程重入该函数导致意想不到的bug，提高代码健壮性std::ofstream ofs(filename, std::ios::app);if (!ofs.is_open()){std::cout << "创建文件失败，filename: " << filename << "\n";return false;}ofs.close();return true;
}

三、数据库系列LogSink实现

1.DBLogSink实现

class DBLogSink
{
public:using ptr = std::shared_ptr<DBLogSink>;virtual ~DBLogSink() {}virtual bool log(const LogMessage &message) = 0;// 开启事务 ->  多次insert -> 关闭事务virtual void log(const std::vector<LogMessage> &message_vec, size_t sz) = 0;
};

2.SqliteSink实现

不要忘了先创建数据库目录和文件，再用_helper打开该文件

class SqliteSink : public DBLogSink
{
public:SqliteSink(const std::string &filename): _helper(filename){// 1. 创建文件所在目录if (!FileHelper::createDir(FileHelper::getPath(filename))){std::cout << "SQLite3数据库文件所在目录创建失败\n";abort();}// 2. 创建文件if (!FileHelper::createFile(filename)){std::cout << "SQLite3数据库文件创建失败\n";abort();}// 3. 打开数据库文件if (!_helper.open()){std::cout << "SQLite3数据库文件打开失败\n";abort();}// 4. 建表if (!createTable()){std::cout << "SQLite3数据库的建表失败\n";abort();}}virtual void log(const LogMessage &message){std::ostringstream insert_sql;insert_sql << "insert into LogData(level,file,line,logger_name,thread_id,body) values(";insert_sql << "'" << LogLevel::LogLevel_Name(message._level) << "',";insert_sql << "'" << message._file << "',";insert_sql << message._line << ",";insert_sql << "'" << message._logger_name << "',";insert_sql << message._thread_id << ",";insert_sql << "'" << message._body << "');";if (!_helper.exec(insert_sql.str(), nullptr, nullptr)){std::cout << "插入数据失败，表名：LogData\n";}}// 开启事务 ->  多次insert -> 关闭事务virtual void log(const std::vector<LogMessage> &message_vec, size_t sz){if (!_helper.begin()){std::cout << "SQLlite3：开启事务失败\n";return;}for (int i = 0; i < sz; i++){log(message_vec[i]);}if (!_helper.commit()){std::cout << "SQLlite3：提交事务失败\n";return;}}private:bool createTable(){// 指明autoincrement必须要用integer，不能用int// SQLite3的current_timestamp是格林威治时间，需要加上8小时改为北京时间static std::string create_sql = R"(create table if not exists LogData(id integer primary key autoincrement,date text default(datetime(current_timestamp,'+8 hours')),level varchar(10),file varchar(32),line int,logger_name varchar(32) not null,thread_id int,body text);)";if (!_helper.exec(create_sql, nullptr, nullptr)){std::cout << "创建SQLite3数据库表失败，表名：LogData\n";return false;}return true;}SqliteHelper _helper;
};

3.MySQLSink实现

注意：这里是要解析配置文件

class MySQLSink : public DBLogSink
{
public:MySQLSink(const std::string &conf_file): _helper(conf_file){// 打开数据库文件if (!_helper.open()){std::cout << "MySQL数据库文件打开失败\n";abort();}if (!createTable()){std::cout << "MySQL数据库的建表失败\n";abort();}}virtual void log(const LogMessage &message){std::ostringstream insert_sql;insert_sql << "insert into LogData(level,file,line,logger_name,thread_id,body) values(";insert_sql << "'" << LogLevel::LogLevel_Name(message._level) << "',";insert_sql << "'" << message._file << "',";insert_sql << message._line << ",";insert_sql << "'" << message._logger_name << "',";insert_sql << message._thread_id << ",";insert_sql << "'" << message._body << "');";if (!_helper.exec(insert_sql.str())){std::cout << "插入数据失败，表名：LogData\n";}}// 开启事务 ->  多次insert -> 关闭事务virtual void log(const std::vector<LogMessage> &message_vec, size_t sz){if (!_helper.begin()){std::cout << "MySQL：开启事务失败\n";return;}for (int i = 0; i < sz; i++){log(message_vec[i]);}if (!_helper.commit()){std::cout << "MySQL：提交事务失败\n";return;}}private:bool createTable(){std::string create_sql = R"(create table if not exists LogData(id int primary key auto_increment,date datetime default current_timestamp,level varchar(10),file varchar(32),line int,logger_name varchar(32) not null,thread_id bigint,body text);)";if (!_helper.exec(create_sql)){std::cout << "创建MySQL数据库表失败，表名：LogData\n";return false;}return true;}MySQLHelper _helper;
};

4.LogSinkFactory完善

增加一个create_db即可

class LogSinkFactory
{
public:// 函数模板template <class SinkType, class... Args>static FSLogSink::ptr create_fs(Args &&...args){return std::make_shared<SinkType>(std::forward<Args>(args)...);}template <class SinkType, class... Args>static DBLogSink::ptr create_db(Args &&...args){return std::make_shared<SinkType>(std::forward<Args>(args)...);}
};

四、同步日志器的修改

1.Logger抽象类的修改

Logger需要加一个成员：std::vector<DBLogSink::ptr> _db_sinks;

同时增加一个用来进行数据库日志落地的接口：

virtual void log_db(const LogMessage &message) = 0;

并且在construct当中继续复用该LogMessage：

void construct(LogLevel::value level, const std::string &file, size_t line, char *body)
{// 1. 构造LogMessageLogMessage message(level, file, line, _logger_name, body);// 2. free掉body,否则会内存泄漏free(body);body = nullptr;// 3. 将LogMessage进行格式化std::string real_message = _formatter->format(message);// 4. 复用log进行实际的日志落地log_fs(real_message.c_str(), real_message.size());log_db(message);
}

2.SyncLogger的修改

同步日志器需要重写父类的log_db函数

virtual void log_db(const LogMessage &message)
{// 这里必须要加锁，因为存在多线程同时调用同一个日志器对象的log函数的情况std::unique_lock<std::mutex> ulock(_mutex);for (auto &sp : _db_sinks){sp->log(message);}
}

直接加锁并且复用即可

3.LoggerBuilder的修改

LoggerBuilder需要加一个 能够往_db_sinks当中添加成员的函数
并且在reset的时候还要清理那里

template <class SinkType, class... Args>
void buildDBLogSink(Args &&...args)
{_db_sinks.push_back(LogSinkFactory::create_db<SinkType>(std::forward<Args>(args)...));
}
// 允许一个建造者在调用build接口构造完对象之后将建造者当中保存的原数据进行重置
void reset()
{_logger_name.clear();_logger_type = LoggerType::SYNC;_limit_level = LogLevel::value::DEBUG;_fs_sinks.clear();_db_sinks.clear();_formatter.reset();
}

4.LocalLoggerBuilder和GlobalLoggerBuilder的修改

1. 修改一下同步和异步日志器的构造函数，将_db_sinks进行传入
2. 只有当对应日志器既不向文件当中落地，也不向数据库当中落地时，才会默认加上标准输出方向的落地

virtual Logger::ptr build()
{if (_logger_name.empty()){std::cout << "日志器名称为空！！\n";abort();}if (_formatter.get() == nullptr){_formatter = std::make_shared<Formatter>();}if (_fs_sinks.empty() && _db_sinks.empty()){_fs_sinks.push_back(LogSinkFactory::create_fs<StdoutSink>());}Logger::ptr ret;if (_logger_type == LoggerType::SYNC){ret = std::make_shared<SyncLogger>(_logger_name, _limit_level, _formatter, _fs_sinks, _db_sinks);}else{// 后面实现完异步日志器之后完善ret = std::make_shared<AsyncLogger>(_logger_name, _limit_level, _formatter, _fs_sinks, _db_sinks, _async_type);}// 重置this->reset();return ret;
}

五、数据库日志缓冲区的实现

1.底层容器的选择

文件日志缓冲区当中存的是vector<char>，这是可行的，因为文件是面向字节流的，日志输出时以字节为单位没毛病

而对于数据库而言，日志的落地是需要一条一条进行落地的，因此vector行不通，除非进行LogMessage的序列化和反序列化

如果序列化+反序列化，那不还是得一条一条插入吗，直接传结构体他不香吗？
折腾来折腾去，毫无意义的好吗

因此我们搞成：vector<LogMessage>

2.原初的vector?

class DBBuffer0
{
public:void push(const LogMessage &message){_buffer.push_back(message);}bool empty(){return _buffer.empty();}void swap(DBBuffer2 &buffer){_buffer.swap(buffer._buffer);}void clear(){_buffer.clear();}const std::vector<LogMessage> &getBuffer(){return _buffer;}private:std::vector<LogMessage> _buffer;
};

没啥毛病，清晰易懂，可是我们知道：vector一般的size变化是：
插入1024*1024条数据，这种扩容方式在面对海量数据时，扩容代价还是太大

因此，我们可以用一下reserve
单论扩容而言，reserve是比resize快的，因为resize还会初始化元素，
而LogMessage的默认构造没用，因此我们不用resize，而是选择reserve

3.具体代码

class DBBuffer
{
public:DBBuffer(size_t default_size = default_buff_size) {_buffer.reserve(default_buff_size);}void push(const LogMessage &message){if (_buffer.size() == _buffer.capacity()){_buffer.reserve(2 * _buffer.capacity());}_buffer.push_back(message);}bool empty(){return _buffer.empty();}bool full(){return _buffer.size() == _buffer.capacity();}void swap(DBBuffer &buffer){_buffer.swap(buffer._buffer);}void clear(){_buffer.clear();}size_t readableSize(){return _buffer.size();}const std::vector<LogMessage> &getBuffer(){return _buffer;}private:std::vector<LogMessage> _buffer;
};

这个bool full()是给异步安全模式使用的

六、AsyncLooper的完善

跟文件日志缓冲区类似，就是多加一份资源的事
没什么特别的，跟文件日志缓冲区那里一样的

using AsyncFSCallback = std::function<void(FSBuffer &)>;
using AsyncDBCallback = std::function<void(DBBuffer &)>;class AsyncLooper
{struct Resource{std::mutex _mutex;std::condition_variable _cond_produce;std::condition_variable _cond_consume;};
public:using ptr = std::shared_ptr<AsyncLooper>;AsyncLooper(AsyncFSCallback fs_callback,AsyncDBCallback db_callback, AsyncType async_type = AsyncType::ASYNC_SAFE): _fs_callback(fs_callback),_db_callback(db_callback), _async_type(async_type), _isrunning(true),_fs_worker(&AsyncLooper::fs_thread_routine, this),_db_worker(&AsyncLooper::db_thread_routine,this) {}~AsyncLooper(){if (_isrunning){_isrunning = false;if (_fs_worker.joinable()){_fs_resource._cond_produce.notify_all();_fs_resource._cond_consume.notify_all();_fs_worker.join();}if(_db_worker.joinable()){_db_resource._cond_produce.notify_all();_db_resource._cond_consume.notify_all();_db_worker.join();}}}void push(const char *data, size_t len){{std::unique_lock<std::mutex> ulock(_fs_resource._mutex);if (_async_type == AsyncType::ASYNC_SAFE && _fs_buffer_produce.writeableSize() < len){_fs_resource._cond_produce.wait(ulock, [this, len]() -> bool{ return !_isrunning || _fs_buffer_produce.writeableSize() >= len; });}_fs_buffer_produce.push(data, len);}// 唤醒消费者_fs_resource._cond_consume.notify_all();}void push(const LogMessage& message){{std::unique_lock<std::mutex> ulock(_db_resource._mutex);if (_async_type == AsyncType::ASYNC_SAFE && _db_buffer_produce.full()){_db_resource._cond_produce.wait(ulock, [this]() -> bool{ return !_isrunning || !_db_buffer_produce.full(); });}_db_buffer_produce.push(message);}_db_resource._cond_consume.notify_all();}private:void fs_thread_routine(){while (true){{// 当停止运行，且生产缓冲区没有数据了，才能退出if (!_isrunning && _fs_buffer_produce.empty())break;std::unique_lock<std::mutex> ulock(_fs_resource._mutex);_fs_resource._cond_consume.wait(ulock, [this]() -> bool{ return !_isrunning || !_fs_buffer_produce.empty(); });// 交换生产和消费缓冲区_fs_buffer_consume.swap(_fs_buffer_produce);}if (_async_type == AsyncType::ASYNC_SAFE){// 唤醒生产者_fs_resource._cond_produce.notify_all();}// 调用日志落地回调函数_fs_callback(_fs_buffer_consume);// 把_buffer_consume重置_fs_buffer_consume.reset();}}void db_thread_routine(){while (true){{std::unique_lock<std::mutex> ulock(_db_resource._mutex);// 当停止运行，且生产缓冲区没有数据了，才能退出if (!_isrunning && _db_buffer_produce.empty())break;_db_resource._cond_consume.wait(ulock, [this]() -> bool{ return !_isrunning || !_db_buffer_produce.empty(); });// 醒了之后即使发现当停止运行，且生产缓冲区没有数据了,也不能退, 因为停止时业务线程有可能还需要放数据// 因为二者想要醒来需要竞争锁, 因此不敢说当前生产缓冲区无数据我就退出,不行,需要等到下次循环时的判断// 交换生产和消费缓冲区_db_buffer_consume.swap(_db_buffer_produce);}if (_async_type == AsyncType::ASYNC_SAFE){// 唤醒生产者_db_resource._cond_produce.notify_all();}// 调用日志落地回调函数_db_callback(_db_buffer_consume);// 把_buffer_consume重置_db_buffer_consume.clear();}}std::atomic<bool> _isrunning;Resource _fs_resource;Resource _db_resource;AsyncFSCallback _fs_callback;AsyncDBCallback _db_callback;AsyncType _async_type;FSBuffer _fs_buffer_produce;FSBuffer _fs_buffer_consume;DBBuffer _db_buffer_produce;DBBuffer _db_buffer_consume;std::thread _fs_worker;std::thread _db_worker;
};

七、异步日志器的完善

重写父类的log_db和realDBLog即可

virtual void log_db(const LogMessage &message)
{_looper->push(message);
}void realDBLog(DBBuffer &buffer)
{// 无需加锁, 因为_looper内部本来就加锁了size_t readableSize = buffer.readableSize();for (auto &sink : _db_sinks){sink->log(buffer.getBuffer(), readableSize);}
}

八、测试

1.前置工作

const std::string sync_logger_db = "./logdb/sync.db";
const std::string async_safe_logger_db = "./logdb/async_safe.db";
const std::string async_unsafe_logger_db = "./logdb/async_unsafe.db";const std::string mysql_conf = "./mysql.conf";void init_sqlite()
{// 创建全局日志器LoggerBuilder::ptr builder = std::make_shared<GlobalLoggerBuilder>();// 1. 同步日志器builder->buildLoggerName(sync_logger_name);builder->buildLoggerType(LoggerType::SYNC);builder->buildDBLogSink<SqliteSink>(sync_logger_db);builder->build();// 2. 异步安全日志器builder->buildLoggerName(async_safe_logger_name);builder->buildLoggerType(LoggerType::ASYNC);builder->buildDBLogSink<SqliteSink>(async_safe_logger_db);builder->buildAsyncType(AsyncType::ASYNC_SAFE);builder->build();// 3. 异步不安全日志器builder->buildLoggerName(async_unsafe_logger_name);builder->buildLoggerType(LoggerType::ASYNC);builder->buildDBLogSink<SqliteSink>(async_unsafe_logger_db);builder->buildAsyncType(AsyncType::ASYNC_UNSAFE);builder->build();
}void init_mysql()
{// 创建全局日志器LoggerBuilder::ptr builder = std::make_shared<GlobalLoggerBuilder>();// 1. 同步日志器builder->buildLoggerName(sync_logger_name);builder->buildLoggerType(LoggerType::SYNC);builder->buildDBLogSink<MySQLSink>(mysql_conf);builder->build();// 2. 异步安全日志器builder->buildLoggerName(async_safe_logger_name);builder->buildLoggerType(LoggerType::ASYNC);builder->buildDBLogSink<MySQLSink>(mysql_conf);builder->buildAsyncType(AsyncType::ASYNC_SAFE);builder->build();// 3. 异步不安全日志器builder->buildLoggerName(async_unsafe_logger_name);builder->buildLoggerType(LoggerType::ASYNC);builder->buildDBLogSink<MySQLSink>(mysql_conf);builder->buildAsyncType(AsyncType::ASYNC_UNSAFE);builder->build();
}

依然还是：
const int log_num = 1024*1024; const int per_size = 100;

2.SQLite3测试

因为同步是一条一条插入的，所以磁盘IO非常频繁，因此
同步我们就给2万条就行了，实际当中不推荐用同步

1.单线程

1.同步

2.异步安全

3.异步不安全

2.多线程

以三个线程为例，先把数据库文件删掉

1.同步

2.异步安全

3.异步不安全

3.MySQL测试

1.单线程

1.同步

这速度，算了吧，打印2万条就行啊

2.异步安全

异步的话最多耽误业务线程10s，可是要耽误工作线程几分钟。。。
这里数据量大的时候才能看出异步的好处


然后把表删了

3.异步不安全

2.多线程

依然是以3个线程为例

1.同步

2万条

2.异步安全

3.异步不安全

以上就是日志系统扩展一：日志落地数据库：MySQL、SQLite3的全部内容哦~