双缓冲异步日志（Async Logging）

文章目录

一、日志系统简介
二、功能需求
三、性能需求
四、高效的异步日志
- 1、异步日志的概念
- 2、双缓冲异步日志解析
- 3、AsyncLogging源码
- 4、代码运行图示
五、双缓冲异步日志的相关问题

一、日志系统简介

日志通常用于故障诊断和追踪(trace)，也可用于性能分析。日志通常是分布式系统中事故调查时的唯一线索, 用来追寻蛛丝马迹, 查出原凶。

【日志需要记录的内容】：

收到每条内部消息的ID(还可以包括关键字段、长度、hash等)；
收到的每条外部消息的全文；
发出每条消息的全文, 每条消息都有全局唯一的id；
关键内部状态的变更, 等等。

【一个日志文件可分为前端(frontend)和后端(backend)两部分】：

前端提供应用程序使用的接口(API), 并生成日志消息(log message)；
后端则负责把日志消息写到目的地(destination)；
典型的多生产者-单消费者问题, 对生产者(前端)而言, 要尽量做到低延迟、低CPU开销、无阻塞；
对消费者(后端)而言, 要做到足够大的吞吐量, 并占用较少资源;

二、功能需求

【日志消息格式有几个要点】：

尽量每条日志占一行，这样很容易用awk、sed、grep等命令行工具快速联机分析日志；
时间戳精确到微妙；
始终使用GMT时区(Z)；
打印线程id，便于分析多线程程序的时序，也可以检测死锁；
打印日志级别；
打印源文件名和行号。

三、性能需求

【日志库的高效性体现在几个方面】：

每秒写上千万条日志的时候没有明显的性能损失；
能应对一个进程生产大量日志数据的场景, 例如1GB/min；
不阻塞正常的执行流程；
在多程序程序中, 不造成争用(contention)；
磁盘带宽约是110MB/S, 日志库应该能瞬时写满这个带宽(不必持续太久)；
假如每条日志消息的平均长度是110字节, 这就意味着1秒要写100万条日志。

【muduo日志库实现了几点优化措施】：

时间戳字符串中的日期和时间部分是缓存的, 一秒内的多条日志只需要重新格式化微妙部分；
日志消息的前4个字段是定长的, 因此可以避免在运行期求字符串长度(不会反复调用strlen)，因为编译器认识memcpy()函数, 对于定长的内存复制, 会在编译期把它的inline展开为高效的目标代码；
线程id是预先格式化为字符串, 在输出日志消息时只需要简单拷贝几个字节；
每行日志消息的源文件名部分采用了编译期计算来获得basename, 避免运行期strrchr()开销。

四、高效的异步日志

1、异步日志的概念

多线程程序对日志库提出了新的需求:线程安全, 即多个程序可以并发写日志, 两个线程的日志消息不会出现交织。

用一个背景线程收集日志消息, 并写入日志文件, 其他业务线程只管往这个日志线程发送日志消息, 这称为异步日志（非阻塞日志）。

2、双缓冲异步日志解析

muduo日志库是用双缓冲技术。基本思路是准备两块buffer：A和B, 前端负责往buffer A填数据(日志消息), 后端负责将buffer B的数据写入文件；当buffer A写满之后, 交换A和B, 让后端将buffer A的数据写入文件, 而前端则往buffer B填入新的日志消息, 如此往复。

使用两个buffer的好处是在新建日志消息的时候不必等待磁盘文件操作，也避免每条新日志消息都触发后端日志线程。换句话说，前端不是将一条条日志消息分别送给后端，而是将多条日志消息拼接成一个大的buffer传送给后端，相当于批处理，减少了线程唤醒的开销。

3、AsyncLogging源码

#ifndef MUDUO_BASE_ASYNCLOGGING_H
#define MUDUO_BASE_ASYNCLOGGING_H#include <muduo/base/BlockingQueue.h>
#include <muduo/base/BoundedBlockingQueue.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/base/Mutex.h>
#include <muduo/base/Thread.h>
#include <muduo/base/LogStream.h>#include <atomic>
#include <vector>namespace muduo
{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(){// 在构造函数中latch_的值为1// 线程运行之后将latch_的减为0running_ = true;thread_.start();// 必须等到latch_变为0才能从start函数中返回，这表明初始化已经完成latch_.wait();}void stop() NO_THREAD_SAFETY_ANALYSIS{running_ = false;cond_.notify();thread_.join();}private:void threadFunc();typedef muduo::detail::FixedBuffer<muduo::detail::kLargeBuffer> Buffer;// 用unique_ptr管理buffer，持有对对象的独有权，不能进行复制操作只能进行移动操作（效率更高）typedef std::vector<std::unique_ptr<Buffer>> BufferVector; typedef BufferVector::value_type BufferPtr; // 指向buffer的指针const int flushInterval_; // 定期（flushInterval_秒）将缓冲区的数据写到文件中std::atomic<bool> running_; // 是否正在运行const string basename_; // 日志名字const off_t rollSize_; // 预留的日志大小muduo::Thread thread_; // 执行该异步日志记录器的线程muduo::CountDownLatch latch_; // 倒计时计数器初始化为1，用于指示什么时候日志记录器才能开始正常工作muduo::MutexLock mutex_;muduo::Condition cond_ GUARDED_BY(mutex_);BufferPtr currentBuffer_ GUARDED_BY(mutex_); // 当前的缓冲区BufferPtr nextBuffer_ GUARDED_BY(mutex_); // 下一个缓冲区BufferVector buffers_ GUARDED_BY(mutex_); // 缓冲区队列
};}  // namespace muduo#endif  // MUDUO_BASE_ASYNCLOGGING_H

#include <muduo/base/AsyncLogging.h>
#include <muduo/base/LogFile.h>
#include <muduo/base/Timestamp.h>#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"), // thread绑定threadFunc回调函数latch_(1),mutex_(),cond_(mutex_),currentBuffer_(new Buffer),nextBuffer_(new Buffer),buffers_()
{currentBuffer_->bzero(); // 缓冲区清零nextBuffer_->bzero();buffers_.reserve(16); // vector预定大小，避免自动增长（效率更高）
}/********************************************************************
Description :
前端在生成一条日志消息时，会调用AsyncLogging::append()。
如果currentBuffer_够用，就把日志内容写入到currentBuffer_中，
如果不够用(就认为其满了)，就把currentBuffer_放到已满buffer数组中，
等待消费者线程（即后台线程）来取。则将预备好的另一块缓冲
（nextBuffer_）移用为当前缓冲区（currentBuffer_）。
*********************************************************************/
void AsyncLogging::append(const char* logline, int len)
{muduo::MutexLockGuard lock(mutex_);// 如果当前buffer的长度大于要添加的日志记录的长度，即当前buffer还有空间，就添加到当前日志。if (currentBuffer_->avail() > len){currentBuffer_->append(logline, len);}// 当前buffer已满。else {// 把当前buffer添加到buffer数组中。buffers_.push_back(std::move(currentBuffer_));// 如果另一块缓冲区不为空，则将预备好的另一块缓冲区移用为当前缓冲区。if (nextBuffer_){currentBuffer_ = std::move(nextBuffer_);}// 如果前端写入速度太快了，一下子把两块缓冲都用完了，那么只好分配一块新的buffer，作当前缓冲区。else{currentBuffer_.reset(new Buffer);}// 添加日志记录。currentBuffer_->append(logline, len);// 通知后端开始写入日志数据。cond_.notify();}
}/********************************************************************
Description :
如果buffers_为空，使用条件变量等待条件满足（即前端线程把一个已经满了
的buffer放到了buffers_中或者超时）。将当前缓冲区放到buffers_数组中。
更新当前缓冲区（currentBuffer_）和另一个缓冲区（nextBuffer_）。
将bufferToWrite和buffers_进行swap。这就完成了将写了日志记录的buffer
从前端线程到后端线程的转变。
*********************************************************************/
void AsyncLogging::threadFunc()
{assert(running_ == true);latch_.countDown();LogFile output(basename_, rollSize_, false);BufferPtr newBuffer1(new Buffer);BufferPtr newBuffer2(new Buffer);newBuffer1->bzero();newBuffer2->bzero();BufferVector buffersToWrite; // 写入日志记录文件的BufferVector。buffersToWrite.reserve(16);while (running_){assert(newBuffer1 && newBuffer1->length() == 0);assert(newBuffer2 && newBuffer2->length() == 0);assert(buffersToWrite.empty());{muduo::MutexLockGuard lock(mutex_);// 如果buffers_为空，那么表示没有数据需要写入文件，那么就等待指定的时间。if (buffers_.empty()){cond_.waitForSeconds(flushInterval_);}// 无论cond是因何（一是超时，二是当前缓冲区写满了）而醒来，都要将currentBuffer_放到buffers_中。  // 如果是因为时间到（3秒）而醒，那么currentBuffer_还没满，此时也要将之写入LogFile中。  // 如果已经有一个前端buffer满了，那么在前端线程中就已经把一个前端buffer放到buffers_中  // 了。此时，还是需要把currentBuffer_放到buffers_中（注意，前后放置是不同的buffer，  // 因为在前端线程中，currentBuffer_已经被换成nextBuffer_指向的buffer了）。buffers_.push_back(std::move(currentBuffer_));// 将新的buffer（newBuffer1）移用为当前缓冲区（currentBuffer_）。currentBuffer_ = std::move(newBuffer1);// buffers_和buffersToWrite交换数据，此时buffers_所有的数据存放在buffersToWrite，而buffers_变为空。buffersToWrite.swap(buffers_);// 如果nextBuffer_为空，将新的buffer（newBuffer2）移用为另一个缓冲区（nextBuffer_）。if (!nextBuffer_){nextBuffer_ = std::move(newBuffer2);}}assert(!buffersToWrite.empty());// 如果将要写入文件的buffer列表中buffer的个数大于25，那么将多余数据删除。// 前端陷入死循环，拼命发送日志消息，超过后端的处理能力，这是典型的生产速度超过消费速度，// 会造成数据在内存中的堆积，严重时引发性能问题(可用内存不足)或程序崩溃(分配内存失败)。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());}// 将buffersToWrite的数据写入到日志文件中for (const auto& buffer : buffersToWrite){output.append(buffer->data(), buffer->length());}// 重新调整buffersToWrite的大小if (buffersToWrite.size() > 2){buffersToWrite.resize(2);}// 从buffersToWrite中弹出一个作为newBuffer1 if (!newBuffer1){assert(!buffersToWrite.empty());newBuffer1 = std::move(buffersToWrite.back());buffersToWrite.pop_back();newBuffer1->reset();}// 从buffersToWrite中弹出一个作为newBuffer2if (!newBuffer2){assert(!buffersToWrite.empty());newBuffer2 = std::move(buffersToWrite.back());buffersToWrite.pop_back();newBuffer2->reset();}// 清空buffersToWritebuffersToWrite.clear();output.flush();}output.flush();
}

4、代码运行图示

【第一种情况】：

【第二种情况】：

【第三种情况】：

【第四种情况】：

五、双缓冲异步日志的相关问题

什么时候切换写到另一个日志文件？前一个buffer已经写满了，则交换两个buffer（写满的buffer置空）。
日志串写入过多，日志线程来不及消费，怎么办？直接丢掉多余的日志buffer，腾出内存，防止引起程序故障。
什么时候唤醒日志线程从Buffer中取数据？其一是超时，其二是前端写满了一个或者多个buffer。