muduo異步日志實現(xiàn)

陳碩老師的muduo網(wǎng)絡(luò)庫的異步日志的實現(xiàn)，今晚有點(diǎn)晚了，我明晚再把這個異步日志抽出來，作為一個獨(dú)立的日志庫。

所在文件

• AsyncLogging.cc

• AsyncLogging.h

• LogFile.h

• LogFile.cc

• CountDownLatch.h

• CountDownLatch.cc

  這個CountDownLatch有點(diǎn)像信號量，但是又只有down操作，上網(wǎng)查了以下類似的，作用有點(diǎn)像屏障

class AsyncLogging : noncopyable
{public:AsyncLogging(const string& basename,off_t rollSize,int flushInterval = 3);~AsyncLogging(){if (running_){stop();}}void append(const char* logline, int len);void start(){running_ = true;thread_.start();	// 啟動線程latch_.wait();		// 這里的wait調(diào)用其實并不會阻塞// 主線程}void stop() NO_THREAD_SAFETY_ANALYSIS{running_ = false;cond_.notify();thread_.join();}private:// 默認(rèn)的守護(hù)進(jìn)程執(zhí)行的函數(shù)void threadFunc();/*Buffer的數(shù)據(jù)結(jié)構(gòu)private:std::vector<char> buffer_;size_t readerIndex_;size_t writerIndex_;static const char kCRLF[];*/typedef muduo::detail::FixedBuffer<muduo::detail::kLargeBuffer> Buffer;typedef std::vector<std::unique_ptr<Buffer>> BufferVector;typedef BufferVector::value_type BufferPtr;	// 表示容器元素的類型const int flushInterval_;						// 刷盤的時間間隔std::atomic<bool> running_;					// 是否運(yùn)行const string basename_;						// 文件名const off_t rollSize_;						// 日志回滾的大小muduo::Thread thread_;						// 日志類自帶一個守護(hù)線程來寫muduo::CountDownLatch latch_;					muduo::MutexLock mutex_;muduo::Condition cond_ GUARDED_BY(mutex_);BufferPtr currentBuffer_ GUARDED_BY(mutex_);	// 當(dāng)前緩沖BufferPtr nextBuffer_ GUARDED_BY(mutex_);		// 預(yù)備緩沖BufferVector buffers_ GUARDED_BY(mutex_);		// 已經(jīng)寫滿并等待落盤的緩沖
};

#include <stdio.h>using namespace muduo;AsyncLogging::AsyncLogging(const string& basename,off_t rollSize,int flushInterval): flushInterval_(flushInterval),running_(false),basename_(basename),rollSize_(rollSize),thread_(std::bind(&AsyncLogging::threadFunc, this), "Logging"),latch_(1),mutex_(),cond_(mutex_),currentBuffer_(new Buffer),nextBuffer_(new Buffer),buffers_()
{currentBuffer_->bzero();nextBuffer_->bzero();buffers_.reserve(16);
}void AsyncLogging::append(const char* logline, int len)
{muduo::MutexLockGuard lock(mutex_);if (currentBuffer_->avail() > len){currentBuffer_->append(logline, len);}else{buffers_.push_back(std::move(currentBuffer_));if (nextBuffer_){currentBuffer_ = std::move(nextBuffer_);}else{// 四個緩沖區(qū)都寫滿了currentBuffer_.reset(new Buffer); // Rarely happens}currentBuffer_->append(logline, len);// 這里的notify不一定什么時候都有效// 如果此時守護(hù)線程正在工作// 那么這個信后就會丟失，但是沒有造成影響// 但是有可能守護(hù)線程正在條件變量上睡眠cond_.notify();}
}void AsyncLogging::threadFunc()
{assert(running_ == true);latch_.countDown();		// 計數(shù)器減一，latch_ == 0，并喚醒主線程LogFile output(basename_, rollSize_, false);	// 初始化一個輸出流BufferPtr newBuffer1(new Buffer);BufferPtr newBuffer2(new Buffer);newBuffer1->bzero();newBuffer2->bzero();BufferVector buffersToWrite;	// 相當(dāng)于一個前后端交互的單元// 將寫滿的buffer裝到vector中// 在傳輸?shù)胶蠖薭uffersToWrite.reserve(16);while (running_){assert(newBuffer1 && newBuffer1->length() == 0);assert(newBuffer2 && newBuffer2->length() == 0);assert(buffersToWrite.empty());{muduo::MutexLockGuard lock(mutex_);// 使用條件變量完成定時任務(wù)if (buffers_.empty())  // unusual usage!{cond_.waitForSeconds(flushInterval_);}buffers_.push_back(std::move(currentBuffer_));currentBuffer_ = std::move(newBuffer1);buffersToWrite.swap(buffers_);	// buffers_變成一個空的buffers_if (!nextBuffer_){nextBuffer_ = std::move(newBuffer2);}}assert(!buffersToWrite.empty());// 如果日志太多了if (buffersToWrite.size() > 25){char buf[256];snprintf(buf, sizeof buf, "Dropped log messages at %s, %zd larger buffers\n",Timestamp::now().toFormattedString().c_str(),buffersToWrite.size()-2);fputs(buf, stderr);// 先輸出一個報警的日志output.append(buf, static_cast<int>(strlen(buf)));// 清除多余的日志buffersToWrite.erase(buffersToWrite.begin()+2, buffersToWrite.end());}for (const auto& buffer : buffersToWrite){// FIXME: use unbuffered stdio FILE ? or use ::writev ?output.append(buffer->data(), buffer->length());}if (buffersToWrite.size() > 2){// drop non-bzero-ed buffers, avoid trashing// vector底層是智能指針不用擔(dān)心內(nèi)存泄露buffersToWrite.resize(2);}if (!newBuffer1){assert(!buffersToWrite.empty());newBuffer1 = std::move(buffersToWrite.back());buffersToWrite.pop_back();newBuffer1->reset();}if (!newBuffer2){assert(!buffersToWrite.empty());newBuffer2 = std::move(buffersToWrite.back());buffersToWrite.pop_back();newBuffer2->reset();}buffersToWrite.clear();output.flush();}output.flush();
}