一个高效的异步日志

106 阅读 0 评论 70 点赞

我是靠谱客的博主复杂蜜粉，这篇文章主要介绍一个高效的异步日志，现在分享给大家，希望可以做个参考。

假如让你自己去写一个日志程序，我想最原始且简单的想法因该是，首先将要写入日志文件的内容转化为字符串，然后调用write系统调用将其写入文件。这种实现方法的确就是我们程序最原始的日志方法。这种做法无疑是十分低效的，那么如何让我们的日志能够高效起来呢？本篇博文就是要给大家分享一种高效的日志–异步日志

1.异步日志要实现什么?

前言中我有告诉打下那种最原始且简单的日志方法很低效，那么它低效在什么地方了呢？
主要有如下几点

复制代码

1
2
3
1.其每条日志内容都会调用write,我们都知道write为系统调用，每一条就调一次它，势必系统开销会很大
2.当我们在关键的地方调用write会不会导致关键部分的代码不能即使的执行?

上述俩个导致我们原始日志低效的主要原因的解决方案分别为:
(1)既然每条日志调用一次write会导致系统开销变大，那么我们就设计一个buffer将日志内容保存在buffer中，待buffer满之后在一次性调用write将其写入即可
(2)由于write可能会阻塞在当前位置，导致紧随其后的关键代码可能不能马上执行，那么我们就单独开个线程专门来执行write调用不就可以了么
根据我们的解决方案，我们可以总结出异步日志的基本流程因该为:

复制代码

1
2
前段线程(我们的程序)负责将日志写入buffer中，当buffer写满之后，将buffer转交给后端线程(我们的负责专门调用write来写日志的线程)，也就是一个典型的多生产者单消费者模式

2.异步日志的代码实现

接下来为大家介绍一下我用C++实现的的异步日志代码

(1)buffer类的实现

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
#ifndef FIX_BUFFER_H_
#define FIX_BUFFER_H_
#include <vector>
#include <string>
#include <assert.h>
#include <fcntl.h>
namespace netlib
{
class FixBuffer
{
public:
FixBuffer()
:buffer_(1024*1024*4)
//初始化大小为4M
{
readableIndex_ = 0;
writeableIndex_ = 0;
}
~FixBuffer()
{
}
int readableSize(void)
//可读字节数
{
return writeableIndex_ - readableIndex_;
}
int writeableSize(void) //可写字节数
{
return buffer_.size() - writeableIndex_;
}
void append(const char *data,int len) //添加数据到buffer中
{
std::copy(data,data + len,getWritePeek());
moveWriteIndex(len);
}
char *getReadPeek(void) //获得读位置的指针
{
return begin() + readableIndex_;
}
char *getWritePeek(void)
//获得写位置的指针
{
return begin() + writeableIndex_;
}
void moveWriteIndex(int len)
//移动可写下标
{
writeableIndex_ = writeableIndex_ + len;
}
void resetBuffer(void)
//重置buffer
{
readableIndex_ = 0;
writeableIndex_ = 0;
}
private:
char *begin()
{
return &*buffer_.begin();
}
std::vector<char> buffer_;
int readableIndex_;
int writeableIndex_;
};
}
#endif

buffer用std::vector来实现，主要提供的对外接口为获取buffer的读写位置指针，可读可写的大小，往buffer中添加数据等，buffer为固定大小4M

(2)用于将日志写如文件的类

logfile.h

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
#ifndef LOG_FILE_H_
#define LOG_FILE_H_
namespace netlib
{
class LogFile
{
public:
LogFile(int rollSize);
~LogFile();
//往磁盘里添加消息
void append(char *log,int len);
//滚动文件
void rollFile(int curLen);
private:
int rollSize_;
//文件滚动大小
int fd_;
int fillSize_;
//当前文件填充大小
int fileNumbers_;
//已有文件数量

};
}
#endif

logfile类的实现

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
#include "logFile.h"
#include <stdio.h>
#include <assert.h>
#include <fcntl.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <sys/stat.h>
using namespace netlib;
LogFile::LogFile(int rollSize)
:rollSize_(rollSize),
fillSize_(0),
fileNumbers_(1)
{
fd_ = open("/home/shreck/log/mylog.log",O_WRONLY | O_APPEND | O_CREAT,S_IWUSR | S_IRUSR | S_IXUSR);
assert(fd_ > 0);
}
LogFile::~LogFile()
{
close(fd_);
}
void LogFile::append(char *log,int len)
{
int ret = write(fd_,log,len);
rollFile(ret);
assert(ret == len);
}
void LogFile::rollFile(int curLen)
{
fillSize_ += curLen;
if(fillSize_ >= rollSize_)
{
printf("1G满了n");
//置0fillSize_
fillSize_ = 0;
char command[80];
snprintf(command,sizeof(command),"mv /home/shreck/log/mylog.log /home/shreck/log/mylog%d.log",fileNumbers_);
fileNumbers_++;
//改当前文件名为fileName
system(command);
//重新创建一个mylog.log文件
fd_ = open("/home/shreck/log/mylog.log",O_WRONLY | O_APPEND | O_CREAT,S_IWUSR | S_IRUSR | S_IXUSR);
assert(fd_ > 0);
}
}

(3)时间戳类的实现

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
#ifndef TIMESTAMP_H_
#define TIMESTAMP_H_
#include <time.h>
#include <string.h>
#include <sys/time.h>
#include <stdio.h>
#include <stdlib.h>
#include <string>
namespace netlib
{
class Timestamp
{
public:
Timestamp()
{
}
~Timestamp()
{
}
static struct tm now(void)
//获取当前tm格式的时间
{
struct timeval tv;
struct tm time;
gettimeofday(&tv,NULL);
//获取微妙，秒值
gmtime_r(&tv.tv_sec,&time);
//将s转换为tm格式
time.tm_year += 1900;
return time;
}
static timeval getTime(void)
{
struct timeval tv;
gettimeofday(&tv,NULL);
return tv;
}
char *toStringTime(void) //将时间转化为字符串并返回
{
struct tm time;
bzero(str_,sizeof(str_));
time = now();
snprintf(str_,sizeof(str_),"%d-%d-%d %d:%d:%d ",time.tm_year,time.tm_mon,time.tm_mday,time.tm_hour,time.tm_min,time.tm_sec);
printf("%s",str_);
return str_;
}
static int getTimeDiff(timeval v1,timeval v2)
//获得俩时间的时间差,返回值为微妙
{
int t;
t = v1.tv_sec*1000000 + v1.tv_usec - v2.tv_sec*1000000 - v2.tv_usec;
t = abs(t);
return t;
}
private:
char str_[100];
};
}
#endif

(4)异步日志类的实现

异步日志类的定义

复制代码

#ifndef ASYNLOG_H_
#define ASYNLOG_H_
#include <memory>
#include "fixBuffer.h"
#include "timestamp.h"
#include <vector>
#include <thread>
#include <string>
#include <mutex>
#include <condition_variable>
namespace netlib
{
enum LogLevel
{
OFF,
//关闭所有日志记录
FATAL,
//导致程序退出的错误
ERROE,
//发生了错误但不影响系统运行
WARN,
//会出现潜在的错误情形
INFO,
//系统发生了那些事情
DEBUG,
//调试程序有关的信息
ALL,
//输出所有日志记录
};
class AsynLog
{
public:
AsynLog(int roolSize);
~AsynLog();
void append(char *logline,int len,LogLevel level); //添加日志行
void setLevel(LogLevel level);
//设置日志的水平
void stop(void);
//关闭日志
private:
void threadFunc(void);
//线程函数
bool aboveLevel(LogLevel level);
//判断某条日志等级是否超过我们所设的level 
void writeBuffer(const char* logline,int len);
//将日志内容写入buffer中
std::string toStringForLevel(LogLevel level);
//将对应的level转化为字符串

int rollSize_;
//文件达到多大时滚动 
bool running_;
std::unique_ptr<FixBuffer> currentBuffer_;
//当前buffer
std::unique_ptr<FixBuffer> nextBuffer_;
//备用buffer
std::vector<std::unique_ptr<FixBuffer>> buffers_;
//保存buffer的vector
std::thread acceptThread_;
//后台接受数据的线程
std::mutex mutex_;
//互斥变量
std::condition_variable_any condition_;
//条件变量
Timestamp timestamp_;
//时间戳
LogLevel currentLevel_;
//当前日志等级
};
}
#endif

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
#ifndef ASYNLOG_H_
#define ASYNLOG_H_
#include <memory>
#include "fixBuffer.h"
#include "timestamp.h"
#include <vector>
#include <thread>
#include <string>
#include <mutex>
#include <condition_variable>
namespace netlib
{
enum LogLevel
{
OFF,
//关闭所有日志记录
FATAL,
//导致程序退出的错误
ERROE,
//发生了错误但不影响系统运行
WARN,
//会出现潜在的错误情形
INFO,
//系统发生了那些事情
DEBUG,
//调试程序有关的信息
ALL,
//输出所有日志记录
};
class AsynLog
{
public:
AsynLog(int roolSize);
~AsynLog();
void append(char *logline,int len,LogLevel level); //添加日志行
void setLevel(LogLevel level);
//设置日志的水平
void stop(void);
//关闭日志
private:
void threadFunc(void);
//线程函数
bool aboveLevel(LogLevel level);
//判断某条日志等级是否超过我们所设的level 
void writeBuffer(const char* logline,int len);
//将日志内容写入buffer中
std::string toStringForLevel(LogLevel level);
//将对应的level转化为字符串

int rollSize_;
//文件达到多大时滚动 
bool running_;
std::unique_ptr<FixBuffer> currentBuffer_;
//当前buffer
std::unique_ptr<FixBuffer> nextBuffer_;
//备用buffer
std::vector<std::unique_ptr<FixBuffer>> buffers_;
//保存buffer的vector
std::thread acceptThread_;
//后台接受数据的线程
std::mutex mutex_;
//互斥变量
std::condition_variable_any condition_;
//条件变量
Timestamp timestamp_;
//时间戳
LogLevel currentLevel_;
//当前日志等级
};
}
#endif

异步日志类的实现

复制代码

#include "asynLog.h"
#include "logFile.h"
#include "timestamp.h"
#include <stdio.h>
#include <string.h>
#include <memory>
#include "fixBuffer.h"
#include <mutex>
#include <condition_variable>
#include <functional>
#include <pthread.h>
using namespace netlib;
AsynLog::AsynLog(int rollSize)
:rollSize_(rollSize),
running_(true),
currentBuffer_(new FixBuffer()),
nextBuffer_(new FixBuffer()),
acceptThread_(std::bind(&AsynLog::threadFunc,this))
{
}
AsynLog::~AsynLog()
{
stop();
}
void AsynLog::setLevel(LogLevel level)
{
currentLevel_ = level;
}
bool AsynLog::aboveLevel(LogLevel level)
{
return level <= currentLevel_;
}
void AsynLog::append(char *logline,int len,LogLevel level)
{
if(aboveLevel(level))
//如果level超过所设等级
{
std::lock_guard<std::mutex> guard(mutex_);
//与时间以及线程id等字符串连接
std::string log1(logline);
//日志内容
std::string log2(timestamp_.toStringTime());//获得时间戳的字符串
char log3[30];
snprintf(log3,sizeof(log3),"threadid[%lu]: ",pthread_self());//获得线程id的字符串
std::string log4(toStringForLevel(level));
//获得等级对应的字符串
log2 = log2 + log3 + log4 + log1;
writeBuffer(log2.c_str(),log2.size());
//写入buffer
}
}
void AsynLog::writeBuffer(const char *logline,int len)
{
if(currentBuffer_->writeableSize() >= len)
//如果当前buffer空间足
{
currentBuffer_->append(logline,len);
}
else
{
buffers_.push_back(std::move(currentBuffer_)); //返回指针，自己变为空
if(nextBuffer_)
{
currentBuffer_ = std::move(nextBuffer_);
//将nextBuffer_的控制权交给currentBuffer_
}
else
{
currentBuffer_.reset(new FixBuffer());
//申请一块新的buffer
}
currentBuffer_->append(logline,len);
condition_.notify_one();
//唤醒后台线程
}
}
std::string AsynLog::toStringForLevel(LogLevel level)
{
switch(level)
{
case LogLevel::OFF:
return std::string("OFF ");break;
case LogLevel::FATAL:
return std::string("FATAL ");break;
case LogLevel::ERROE:
return std::string("ERROE ");break;
case LogLevel::WARN:
return std::string("WARN ");break;
case LogLevel::INFO:
return std::string("INFO ");break;
case LogLevel::DEBUG:
return std::string("DEBUG ");break;
case LogLevel::ALL:
return std::string("ALL ");break;
default:return std::string("UNKNOWN ");
}
return NULL;
}
void AsynLog::stop(void)
{
if(running_)
{
running_ = false;
acceptThread_.join();
//等待后台进程退出
printf("日志系统已关闭n");
}
}
void AsynLog::threadFunc(void)
{
std::unique_ptr<FixBuffer> newBuffer1(new FixBuffer);
std::unique_ptr<FixBuffer> newBuffer2(new FixBuffer);
std::vector<std::unique_ptr<FixBuffer>> buffersToWrite;
LogFile output(rollSize_);
buffersToWrite.reserve(16);
while(running_)
{
{
std::lock_guard<std::mutex> guard(mutex_);
//临界区加锁
if(buffers_.empty())
//如果buffers_为空
{
condition_.wait_for(mutex_,std::chrono::seconds(3));
}
buffers_.push_back(std::move(currentBuffer_));
currentBuffer_ = std::move(newBuffer1);
buffersToWrite.swap(buffers_);
//交换俩个容器
if(!nextBuffer_)
{
nextBuffer_ = std::move(newBuffer2);
}
}
assert(!buffersToWrite.empty());
if(buffersToWrite.size() > 25)
{
//日志异常
}
//将buffer中的内容写进文件中
for(int i = 0; i < buffersToWrite.size(); i++)
{
output.append(buffersToWrite[i]->getReadPeek(),buffersToWrite[i]->readableSize());
}
if(buffersToWrite.size() > 2)
{
buffersToWrite.resize(2);
}
if(!newBuffer1)
{
assert(!buffersToWrite.empty());
newBuffer1 = std::move(buffersToWrite[0]);
newBuffer1->resetBuffer();
//重置buffer
}
if(!newBuffer2)
{
assert(!buffersToWrite.empty());
newBuffer2 = std::move(buffersToWrite[1]);
newBuffer2->resetBuffer();
}
buffersToWrite.clear();
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
#include "asynLog.h"
#include "logFile.h"
#include "timestamp.h"
#include <stdio.h>
#include <string.h>
#include <memory>
#include "fixBuffer.h"
#include <mutex>
#include <condition_variable>
#include <functional>
#include <pthread.h>
using namespace netlib;
AsynLog::AsynLog(int rollSize)
:rollSize_(rollSize),
running_(true),
currentBuffer_(new FixBuffer()),
nextBuffer_(new FixBuffer()),
acceptThread_(std::bind(&AsynLog::threadFunc,this))
{
}
AsynLog::~AsynLog()
{
stop();
}
void AsynLog::setLevel(LogLevel level)
{
currentLevel_ = level;
}
bool AsynLog::aboveLevel(LogLevel level)
{
return level <= currentLevel_;
}
void AsynLog::append(char *logline,int len,LogLevel level)
{
if(aboveLevel(level))
//如果level超过所设等级
{
std::lock_guard<std::mutex> guard(mutex_);
//与时间以及线程id等字符串连接
std::string log1(logline);
//日志内容
std::string log2(timestamp_.toStringTime());//获得时间戳的字符串
char log3[30];
snprintf(log3,sizeof(log3),"threadid[%lu]: ",pthread_self());//获得线程id的字符串
std::string log4(toStringForLevel(level));
//获得等级对应的字符串
log2 = log2 + log3 + log4 + log1;
writeBuffer(log2.c_str(),log2.size());
//写入buffer
}
}
void AsynLog::writeBuffer(const char *logline,int len)
{
if(currentBuffer_->writeableSize() >= len)
//如果当前buffer空间足
{
currentBuffer_->append(logline,len);
}
else
{
buffers_.push_back(std::move(currentBuffer_)); //返回指针，自己变为空
if(nextBuffer_)
{
currentBuffer_ = std::move(nextBuffer_);
//将nextBuffer_的控制权交给currentBuffer_
}
else
{
currentBuffer_.reset(new FixBuffer());
//申请一块新的buffer
}
currentBuffer_->append(logline,len);
condition_.notify_one();
//唤醒后台线程
}
}
std::string AsynLog::toStringForLevel(LogLevel level)
{
switch(level)
{
case LogLevel::OFF:
return std::string("OFF ");break;
case LogLevel::FATAL:
return std::string("FATAL ");break;
case LogLevel::ERROE:
return std::string("ERROE ");break;
case LogLevel::WARN:
return std::string("WARN ");break;
case LogLevel::INFO:
return std::string("INFO ");break;
case LogLevel::DEBUG:
return std::string("DEBUG ");break;
case LogLevel::ALL:
return std::string("ALL ");break;
default:return std::string("UNKNOWN ");
}
return NULL;
}
void AsynLog::stop(void)
{
if(running_)
{
running_ = false;
acceptThread_.join();
//等待后台进程退出
printf("日志系统已关闭n");
}
}
void AsynLog::threadFunc(void)
{
std::unique_ptr<FixBuffer> newBuffer1(new FixBuffer);
std::unique_ptr<FixBuffer> newBuffer2(new FixBuffer);
std::vector<std::unique_ptr<FixBuffer>> buffersToWrite;
LogFile output(rollSize_);
buffersToWrite.reserve(16);
while(running_)
{
{
std::lock_guard<std::mutex> guard(mutex_);
//临界区加锁
if(buffers_.empty())
//如果buffers_为空
{
condition_.wait_for(mutex_,std::chrono::seconds(3));
}
buffers_.push_back(std::move(currentBuffer_));
currentBuffer_ = std::move(newBuffer1);
buffersToWrite.swap(buffers_);
//交换俩个容器
if(!nextBuffer_)
{
nextBuffer_ = std::move(newBuffer2);
}
}
assert(!buffersToWrite.empty());
if(buffersToWrite.size() > 25)
{
//日志异常
}
//将buffer中的内容写进文件中
for(int i = 0; i < buffersToWrite.size(); i++)
{
output.append(buffersToWrite[i]->getReadPeek(),buffersToWrite[i]->readableSize());
}
if(buffersToWrite.size() > 2)
{
buffersToWrite.resize(2);
}
if(!newBuffer1)
{
assert(!buffersToWrite.empty());
newBuffer1 = std::move(buffersToWrite[0]);
newBuffer1->resetBuffer();
//重置buffer
}
if(!newBuffer2)
{
assert(!buffersToWrite.empty());
newBuffer2 = std::move(buffersToWrite[1]);
newBuffer2->resetBuffer();
}
buffersToWrite.clear();
}
}

测试代码

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
#include<iostream>
#include "asynLog.h"
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <thread>
netlib::AsynLog log(1024*1024*1024);
void func(void)
{
char s[100] = "hellon";
log.setLevel(netlib::LogLevel::ALL);
while(true)
{
log.append(s,strlen(s),netlib::LogLevel::INFO);
}
}
int main(int argc,char **argv)
{
std::thread t1(func);
std::thread t2(func);
std::thread t3(func);
std::thread t4(func);
t1.join();
t2.join();
t3.join();
t4.join();
return 0;
}