一個輕巧高效的多線程c++stream風格異步日誌(二)

時間  2019-12-06
標籤 一個 輕巧 高效 多線程 c++stream stream 風格 異步 日誌 欄目 Java 简体版
原文   原文鏈接

目錄c++

一個輕巧高效的多線程c++stream風格異步日誌(二)

前言

本文緊接上一篇文章: 介紹上文中的一條條日誌是如何異步導入本地文件的.
首先會簡單介紹下LogFile類，以後會具體講解下AsyncLogging中的雙緩衝機制.
整個日誌模塊的結構圖,
git

LogFile類

LogFile日誌文件類 完成日誌文件的管理工做.
rollFile() ：滾動文件 當日志超過m_rollSize大小時會滾動一個新的日誌文件出來.
getLogFileName() ：用與滾動日誌時，給日誌文件取名，以滾動時間做爲後綴.
m_mutex ：用於append()數據時,給文件上鎖.
append() ：黏入日誌.
flush() ：沖刷緩衝.github

LogFile 有一個AppendFIle類，它是最終用於操做本地文件的類.
append() : 裏面會調用系統函數fwrite()寫入本地文件.
flush() ： 沖刷緩衝.
writtenBytes() ： 獲取已寫字節數.緩存

AsyncLogging類

AsyncLogging異步日誌類， 完成日誌的異步寫入工做.
介紹它的接口前,先描述下它的工做邏輯.多線程

AsyncLogging 有如下述幾類緩存.
m_currentBuffer : 指向當前接收其餘線程append過來的日誌的緩存.
m_buffers : 用於存放當前已寫滿或過了沖刷週期的日誌緩存的指針容器.
m_nextBuffer : 指向當m_currentBuffer滿後用於替代m_currentBuffer的緩存.app

backupBuffer1 : 備用緩存.
backupBuffer2 : 備用緩存.
buffersToWrite : 和m_buffers經過交換swap()後append()到LogFile的指針容器.異步

AsyncLogging 使用的雙緩衝機制 有兩個緩存容器 : m_buffers 、buffersToWrite 交替使用 . 一下咱們簡稱爲 A 和 B .
A 用於接收 其餘線程 append() 進來的日誌.
B 用於將目前已接受的緩存 寫入 日誌文件. 當B寫完時 , clean() B , 交換A，B，如此往復.async

優勢 ： 新建的日誌沒必要等待磁盤操做，也避免了每條新日誌都觸發日誌線程，而是將多條日誌拼程一個大的buffer 傳送給日誌線程寫入文件. 至關於批處理， 減小線程喚醒頻率 ，下降開銷。
另外 ，爲了及時將 日誌消息寫入文件， 便是 buffer A 中尚未push進來日誌 也會每三秒 執行一次上述的寫入操做.函數

AsyncLogging使用一個更大的LogBuffer來保存一條條Logger傳送過來的日誌.
Mutex ：用來控制多線程的寫入.
Condition : 用來等待緩衝區中的數據.
Thread : 使用一個線程處理緩存的交換,以及日誌的寫入.
優化

AsyncLogging實現

下面會給出AsyncLogging的簡單實現.
實際上還有幾個備用緩存,這裏沒有加上去,以便於理解程序; 備用緩存主要是爲了減小反覆new 操做帶來的系統開銷，

#ifndef _ASYNC_LOGGING_HH
#define _ASYNC_LOGGING_HH
#include "MutexLock.hh"
#include "Thread.hh"
#include "LogStream.hh"
#include "ptr_vector.hh"
#include "Condition.hh"

#include <string>

class AsyncLogging
{
public:
    AsyncLogging(const std::string filePath, off_t rollSize, int flushInterval = 3);
    ~AsyncLogging();

    void start(){
        m_isRunning = true;
        m_thread.start();
    }

    void stop(){
        m_isRunning = false;
        m_cond.notify();
    }

    void append(const char *logline, int len);

private:
    AsyncLogging(const AsyncLogging&);
    AsyncLogging& operator=(const AsyncLogging&);

    void threadRoutine();

    typedef LogBuffer<kLargeBuffer> Buffer;
    typedef oneself::ptr_vector<Buffer> BufferVector;
    typedef oneself::auto_ptr<Buffer> BufferPtr;

    const int m_flushInterval;
    bool m_isRunning;
    off_t m_rollSize;
    std::string m_filePath;
    Thread m_thread;
    MutexLock m_mutex;
    Condition m_cond;

    BufferPtr m_currentBuffer;
    BufferVector m_buffers;
};

#endif



//AsyncLogging.cpp
#include "AsyncLogging.hh"
#include "LogFile.hh"
#include <assert.h>
#include <stdio.h>

AsyncLogging::AsyncLogging(const std::string filePath, off_t rollSize, int flushInterval)
    :m_filePath(filePath),
     m_rollSize(2048),
     m_flushInterval(flushInterval),
     m_isRunning(false),
     m_thread(std::bind(&AsyncLogging::threadRoutine, this)),
     m_mutex(),
     m_cond(m_mutex),
     m_currentBuffer(new Buffer),
     m_buffers()
{
}

AsyncLogging::~AsyncLogging(){
    if(m_isRunning) stop();
}

void AsyncLogging::append(const char* logline, int len){
    MutexLockGuard lock(m_mutex);
    if(m_currentBuffer->avail() > len){
        m_currentBuffer->append(logline, len);
    }
    else{
        m_buffers.push_back(m_currentBuffer.release());
        
        m_currentBuffer.reset(new Buffer);

        m_currentBuffer->append(logline, len);
        m_cond.notify();
    }
}

void AsyncLogging::threadRoutine(){
    assert(m_isRunning == true);
    LogFile output(m_filePath, m_rollSize, false);
    BufferVector buffersToWrite;
    buffersToWrite.reserve(8);

    while(m_isRunning){
        assert(buffersToWrite.empty());
        {
            MutexLockGuard lock(m_mutex);
            if(m_buffers.empty()){
                m_cond.waitForSeconds(m_flushInterval);
            }
            m_buffers.push_back(m_currentBuffer.release());
            m_currentBuffer.reset(new Buffer);
            m_buffers.swap(buffersToWrite);
        }

        assert(!buffersToWrite.empty());

        for(size_t i = 0; i < buffersToWrite.size(); ++i){
            output.append(buffersToWrite[i]->data(), buffersToWrite[i]->length());
        }

        buffersToWrite.clear();
        output.flush();
    }

    output.flush();
}

增長備用緩存

增長備用緩存優化上面程序，上面程序一共在兩個地方執行了new操做.
1.m_currentBuffer 填滿時，須要把它填進容器的時候.
2.到時間了須要把m_currentBuffer裏面的內容寫入本地文件時,會把它當前的內容移出來，這時候須要new一個新緩存來給m_currentBuffer.

因而咱們準備一個m_nextBuffer來作m_currentBuffer的備用緩存.同時在線程中增長兩個backupBuffer 給m_nextBuffer 當備用緩存；當日志量大到不夠用的時候, 再考慮用new 操做來動態添加緩存。

#ifndef _ASYNC_LOGGING_HH
#define _ASYNC_LOGGING_HH
#include "MutexLock.hh"
#include "Thread.hh"
#include "LogStream.hh"
#include "ptr_vector.hh"


#include "Condition.hh"
#include <memory>
#include <string>

class AsyncLogging
{
public:
    AsyncLogging(const std::string filePath, off_t rollSize, int flushInterval = 3);
    ~AsyncLogging();

    void start(){
        m_isRunning = true;
        m_thread.start();
    }

    void stop(){
        m_isRunning = false;
        m_cond.notify();
    }

    void append(const char *logline, int len);

private:
    AsyncLogging(const AsyncLogging&);
    AsyncLogging& operator=(const AsyncLogging&);

    void threadRoutine();

    typedef LogBuffer<kLargeBuffer> Buffer;
    typedef myself::ptr_vector<Buffer> BufferVector;
    typedef std::unique_ptr<Buffer> BufferPtr;

    const int m_flushInterval;
    bool m_isRunning;
    off_t m_rollSize;
    std::string m_filePath;
    Thread m_thread;
    MutexLock m_mutex;
    Condition m_cond;

    BufferPtr m_currentBuffer;
    BufferPtr m_nextBuffer;
    BufferVector m_buffers;
};

#endif


//AsynvLogging.cpp
#include "AsyncLogging.hh"
#include "LogFile.hh"
#include <assert.h>
#include <stdio.h>

AsyncLogging::AsyncLogging(const std::string filePath, off_t rollSize, int flushInterval)
    :m_filePath(filePath),
     m_rollSize(rollSize),
     m_flushInterval(flushInterval),
     m_isRunning(false),
     m_thread(std::bind(&AsyncLogging::threadRoutine, this)),
     m_mutex(),
     m_cond(m_mutex),
     m_currentBuffer(new Buffer),
     m_nextBuffer(new Buffer),
     m_buffers()
{
}

AsyncLogging::~AsyncLogging(){
    if(m_isRunning) stop();
}

void AsyncLogging::append(const char* logline, int len){
    MutexLockGuard lock(m_mutex);
    if(m_currentBuffer->avail() > len){
        m_currentBuffer->append(logline, len);
    }
    else{
        m_buffers.push_back(m_currentBuffer.release());
        
        if(m_nextBuffer){
            m_currentBuffer = std::move(m_nextBuffer);
        }
        else{
            m_currentBuffer.reset(new Buffer);
        }

        m_currentBuffer->append(logline, len);
        m_cond.notify();
    }
}

void AsyncLogging::threadRoutine(){
    assert(m_isRunning == true);
    LogFile output(m_filePath, m_rollSize, false);
    BufferPtr backupBuffer1(new Buffer);
    BufferPtr backupBuffer2(new Buffer);
    BufferVector buffersToWrite;
    buffersToWrite.reserve(8);

    while(m_isRunning){
        assert(buffersToWrite.empty());
        {
            MutexLockGuard lock(m_mutex);
            if(m_buffers.empty()){
                m_cond.waitForSeconds(m_flushInterval);
            }
            m_buffers.push_back(m_currentBuffer.release());
            m_currentBuffer = std::move(backupBuffer1);
            m_buffers.swap(buffersToWrite);
            if(!m_nextBuffer)
                m_nextBuffer = std::move(backupBuffer2);
        }

        assert(!buffersToWrite.empty());

        for(size_t i = 0; i < buffersToWrite.size(); ++i){
            output.append(buffersToWrite[i]->data(), buffersToWrite[i]->length());
        }

        if(buffersToWrite.size() > 2)
        {
            // drop non-bzero-ed buffers, avoid trashing
            buffersToWrite.resize(2);
        }

        if(!backupBuffer1)
        {
            assert(!buffersToWrite.empty());
            backupBuffer1 = std::move(buffersToWrite.pop_back());
            backupBuffer1->reset();
        }

        if(!backupBuffer2)
        {
            assert(!buffersToWrite.empty());
            backupBuffer2 = std::move(buffersToWrite.pop_back());
            backupBuffer2->reset();
        }

        buffersToWrite.clear();
        output.flush();
    }

    output.flush();
}

結語

本文主要介紹了muduo中AsyncLogging類的實現,其中的雙緩存機制.
LogFile類及AppendFIle類 分別是日誌文件管理類和本地文件的基本操做類. 不難理解，感興趣的話能夠看看muduo的源碼，本文再也不往下寫了，若是想要所有源碼能夠留言。

最新源碼：
https://github.com/BethlyRoseDaisley/SimpleMuduo/tree/master/AsyncLogging

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程