结合生产消费者模式实现异步日志功能

软件程序开发过程中,日志是诊断bug必不可少的功能,日志功能通常是将每条日志信息按照一定的格式写入指定的文件,但是,实时将日志信息写入文件,必定耗费时间,对于性能要求比较高的机器来说,可能是无法接受的,并且由于时间差问题可能会带来无法预料的问题。

基于上面的原因,解决方案是将日志信息临时存储内存,然后启动线程来将内存中的日志写入文件,因此,本文将结合生产消费者模式来实现异步写入日志的功能。

生产者消费者模式,顾名思义,就是生产者生成数据,消费者处理数据。首先,将通过例子来说明生产者消费者的模式,然后再介绍异步写入日志的功能,其功能代码虽然简单,但是对于日志功能要求场景不多的人来说,却是相当实用的。

一、生产者消费者模式

1、实现简单的生产消费者管理类,生产者即函数Product, 它首先加锁,然后将数据写入队列,最后通过条件变量来唤醒消费者来处理数据;消费者即函数Consume,  它首先加锁,调用条件变量的wait等待接受信号,如果接受到信号,那么从队列中取出数据然后处理,这里需要注意的是取出数据之后,可以提前解锁,以便生产者能够尽快处理数据,另外wait函数添加的匿名函数,它判断队列是否为空,如果不为空,才继续往下处理数据,如果为空,那么继续等待,   这样做的原因是wait返回有可能不是因为接受到生产者发送的信号。

#include <iostream>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <queue>

class JDataManager
{
public:
    //生产者
    void Product()
    {
        for(int i = 0; i < 5; i++)
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            m_queue.push(string("hello"));
            m_condition.notify_one();
            std::cout << "Product  i =  " << i << std::endl;
        }
    }
    
    //消费者
    void Consume()
    {
        int i = 0;
        do
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            m_condition.wait(lock, [&]{ return !m_queue.empty(); } );
            std::string str = m_queue.front();
            m_queue.pop();
            std::cout << "Consume i = " << i << std::endl;
            i++;
        }while(true);
    }
private:
    std::mutex m_mutex;
    std::queue<std::string> m_queue;
    std::condition_variable m_condition;
};

2、上面实现了生产消费者管理类,接下来测试下其运行效果,首先启动线程用于执行消费者函数, 休眠两秒,再启动第二个线程用于执行生产者函数

JDataManager data_manager;

std::thread consumer(&JDataManager::Consume, &data_manager);

std::this_thread::sleep_for(std::chrono::seconds(2));

std::thread productor(&JDataManager::Product, &data_manager);


if(productor.joinable())
{
    productor.join();
}
if(consumer.joinable())
{
    consumer.join();
}

3、最后运行打印的信息如下图所示,生产者生成的数据,消费者对应的提取出来。这先启动消费者,再启动生产者后,运行的效果是合理的。

4、上面是先启动消费者,再启动生产者,如果反过来呢,即先启动生产者,再启动消费者

std::thread productor(&JDataManager::Product, &data_manager);

std::this_thread::sleep_for(std::chrono::seconds(2));

std::thread consumer(&JDataManager::Consume, &data_manager);

5、最后运行打印的信息如下图所示,先启动生产者再启动消费者,消费者也能够正常处理生产者生成的数据。

二、异步日志功能

1、日志是程序中每个模块都会使用到的功能,所以,考虑采用单例模式来实现日志的基本框架。

/// 类定义
class JMyLog
{
public:
    ~JMyLog();
    static JMyLog* Instance(void);

private:
    JMyLog();
);

private:
    static JMyLog* m_pMyLog;
};


/// 类实现
JMyLog* JMyLog::m_pMyLog = nullptr;

JMyLog::JMyLog()
{
}


JMyLog::~JMyLog()
{
    if (m_pMyLog)
    {
        delete m_pMyLog;
        m_pMyLog = nullptr;
    }
}

JMyLog* JMyLog::Instance(void)
{
    if (m_pMyLog == nullptr)
    {
        m_pMyLog = new JMyLog();
    }
    return m_pMyLog;
}

2、实现将每条日志信息写入队列的函数接口,这个相当于生产者, 它负责将写入的每条日志写入队列,再通过条件变量通知消费者处理数据。

/// 类定义
class JMyLog
{
public:
    ~JMyLog();
    static JMyLog* Instance(void);
    // 每条日志信息写入队列
    void WriteLog(int iLogLevel, const std::string &strFileName, int iLineNum
                  , const std::string &strFunName, const char *pFmt, ...);
private:
    JMyLog();

    std::string GetSysTimeToMs();
    std::string GetFirstLog();
    std::string GetLevelInfo(int iLevel);
    std::string GetThreadId();

private:
    static JMyLog* m_pMyLog;
    std::mutex  m_mutex;
    std::deque<std::string> m_deque;
    int m_iLogLevel;
    std::condition_variable m_condVariable;
};


/// 函数实现
static const std::string LOG_DEBUG = "DEBUG";

void JMyLog::WriteLog(int iLogLevel, const std::string &strFileName, int iLineNum
              , const std::string &strFunName, const char *pFmt, ...)
{

    std::unique_lock<std::mutex> lock(m_mutex);
    va_list vaa;
    va_start(vaa, pFmt);
    char ac_logbuf[1024];
    std::memset(ac_logbuf, 0x00, sizeof(ac_logbuf));

    snprintf(ac_logbuf, sizeof (ac_logbuf) - 2, "[%s][%s:%d:%s][%s][%s]"
             , GetSysTimeToMs().c_str()
             , strFileName.c_str()
             , iLineNum
             , strFunName.c_str()
             , GetThreadId().c_str()
             , GetLevelInfo(iLogLevel).c_str());

    size_t ilog_len = strlen(ac_logbuf);
    vsnprintf(ac_logbuf + ilog_len, sizeof(ac_logbuf) - ilog_len -2, pFmt, vaa);
    ilog_len = strlen(ac_logbuf);
    ac_logbuf[ilog_len] = '\n';
    m_deque.push_back(ac_logbuf);
    va_end(vaa);
    m_condVariable.notify_one();
    lock.unlock();
}

std::string JMyLog::GetSysTimeToMs()
{
    time_t timep;
    struct timeb tb;

    time (&timep);
    char tmp[128] ={0};
    strftime(tmp, sizeof(tmp), "%Y-%m-%d %H:%M:%S",localtime(&timep) );

    char tmp2[128] ={0};
    ftime(&tb);
    snprintf(tmp2,sizeof(tmp2),"%d",tb.millitm);

    std::ostringstream buffer;
    buffer << tmp << "." << tmp2 ;

    return buffer.str();
}

std::string JMyLog::GetLevelInfo(int iLevel)
{
    std::string str_level_info;
    switch (iLevel)
    {
        case E_LOG_DEBUG:
        {
            str_level_info = LOG_DEBUG;
            break;
        }
        default:
        {
            str_level_info = LOG_DEBUG;
            break;
        }
    }
    return str_level_info;
}

std::string JMyLog::GetThreadId()
{
    std::ostringstream thread_id;
    thread_id << std::hex << std::this_thread::get_id();
    return thread_id.str();
}

std::string JMyLog::GetFirstLog()
{
    std::string str = m_deque.front();
    m_deque.pop_front();
    return str;
}

3、上面实现了生产者的日志生成功能后,接下来就是实现消费者的日志处理功能,由于考虑的是异步的模式,所以,消费者需要在线程中运行。下面实现的消费者是从队列中取出数据,然后将日志信息打印到终端,后面将添加日志写入文件的功能。

/// 类定义
class JMyLog
{
public:
    ~JMyLog();
    static JMyLog* Instance(void);
    // 每条日志信息写入队列
    void WriteLog(int iLogLevel, const std::string &strFileName, int iLineNum
                  , const std::string &strFunName, const char *pFmt, ...);
private:
    JMyLog();

    std::string GetSysTimeToMs();
    std::string GetFirstLog();
    std::string GetLevelInfo(int iLevel);
    std::string GetThreadId();

    // 启动线程
    void StartThread();
    // 线程执行函数
    void ThreadExce();

private:
    static JMyLog* m_pMyLog;
    std::mutex  m_mutex;
    std::deque<std::string> m_deque;
    int m_iLogLevel;
    std::condition_variable m_condVariable;
};

/// 函数实现
void JMyLog::StartThread()
{
    std::thread thread_obj(&JMyLog::ThreadExce, this);
    thread_obj.detach();
}

void JMyLog::ThreadExce()
{
    while(true)
    {
        std::unique_lock<std::mutex> lock_log(m_mutex);
        m_condVariable.wait(lock_log, [&] { return !m_deque.empty();});
        std::string str = GetFirstLog();
        lock_log.unlock();
        if (!str.empty())
        {
            std::cout << str;
        }
        str.clear();
    }
}

4、为了用户更加方便的调用,我们定义了如下所示的宏

enum E_LOG_LEVEL
{
    E_LOG_DEBUG = 1,
};

#define MyLogD(pFmt, ...) \
    JMyLog::Instance()->WriteLog(E_LOG_DEBUG, __FILE__, __LINE__, __func__, pFmt, ##__VA_ARGS__);

5、测试代码如下所示,调用者按照类似printf的格式使用定义好的宏MyLogD

MyLogD("%s", "this is mylog1.");
MyLogD("%s", "this is mylog2.");

6、日志输出的格式如下图所示

[2019-11-24 16:27:28.950][../JQtTestStudy/debbugtest/jdebugcppattr.cpp:259:TestMyLog][0x7fffa4c16380][DEBUG]this is mylog1.
[2019-11-24 16:27:28.951][../JQtTestStudy/debbugtest/jdebugcppattr.cpp:260:TestMyLog][0x7fffa4c16380][DEBUG]this is mylog2.

7、 为了将日志信息写入文件,封装日志文件写入类,该写入类主要实现打开文件、关闭文件,日志写入文件以及刷新文件的四个函数。

#include <string>
#include <fstream>

/// 类定义
class JLogFileHandler
{
public:
    JLogFileHandler(const std::string &strFilePath);
    ~JLogFileHandler();

    void Open();
    void Close();
    void Write(const std::string &strInfo);
    void Flush();

private:
    void InitFilePath();

private:
    std::string m_strFilePath;
    std::ofstream m_outfstream;
};

/// 类实现
JLogFileHandler::JLogFileHandler(const std::string &strFilePath)
    : m_strFilePath(strFilePath)
{
    InitFilePath();
}


JLogFileHandler::~JLogFileHandler()
{

}

void JLogFileHandler::InitFilePath()
{

}

void JLogFileHandler::Open()
{
    if (!m_outfstream.is_open())
    {
        m_outfstream.open(m_strFilePath);
    }
}

void JLogFileHandler::Close()
{
    if (m_outfstream.is_open())
    {
        m_outfstream.close();
    }
}

void JLogFileHandler::Write(const std::string &strInfo)
{
    m_outfstream << strInfo;
}

void JLogFileHandler::Flush()
{
    m_outfstream.flush();
}

8、JMyLog类添加文件处理者对象,然后线程执行函数中将日志信息写入文件,并且执行刷新功能

/// 类定义
class JMyLog
{
public:
    ~JMyLog();
    static JMyLog* Instance(void);
    // 每条日志信息写入队列
    void WriteLog(int iLogLevel, const std::string &strFileName, int iLineNum
                  , const std::string &strFunName, const char *pFmt, ...);
private:
    JMyLog();

    std::string GetSysTimeToMs();
    std::string GetFirstLog();
    std::string GetLevelInfo(int iLevel);
    std::string GetThreadId();

    // 启动线程
    void StartThread();
    // 线程执行函数
    void ThreadExce();

    // 初始化文件
    void InitFile();

private:
    static JMyLog* m_pMyLog;
    std::mutex  m_mutex;
    std::deque<std::string> m_deque;
    int m_iLogLevel;
    std::condition_variable m_condVariable;
    std::shared_ptr<JLogFileHandler> m_FileHandler; // 文件处理者

};

/// 函数实现
JMyLog::JMyLog()
{
    InitFile();
    StartThread();
}

void JMyLog::ThreadExce()
{
    while(true)
    {
        std::unique_lock<std::mutex> lock_log(m_mutex);
        m_condVariable.wait(lock_log, [&] { return !m_deque.empty();});
        std::string str = GetFirstLog();
        lock_log.unlock();
        if (!str.empty())
        {
            std::cout << str;

            m_FileHandler->Write(str);
            m_FileHandler->Flush();
        }
        str.clear();
    }
}

void JMyLog::InitFile()
{
    std::string str_file = "../../../log/test.log";
    m_FileHandler = std::make_shared<JLogFileHandler>(str_file);
    m_FileHandler->Open();
}

9、再次运行测试代码,查看日志目录下生成了日志文件test.log,并且日志信息也成功写入到文件

三、总结

最后再来总结异步日志功能的实现步骤,首先采用单例模式实现日志的基本框架,接着实现日志生产者,即提供写入日志信息接口,然后再实现日志消费者,日志消费者运行在线程中,并且收到信号才开始处理数据,最后实现文件处理者,将日志信息写入文件。至此,结合生产者消费者模式的异步日志功能完成了。