一个轻巧高效的多线程c++stream风格异步日志(二)

时间 2019-12-06

标签一个轻巧高效多线程 c++stream stream 风格异步日志栏目 Java 繁體版

原文原文链接

目录c++

一个轻巧高效的多线程c++stream风格异步日志(二)

一个轻巧高效的多线程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的源码，本文再也不往下写了，若是想要所有源码能够留言。