扩充Log4j来实现性能优化的异步日志收集器
日志收集在互联网企业尤其是大数据时代是一件非常重要的事情，日志记录着用户行为和系统行为，是一种重要的数据来源。Log4j是Java应用程序使用的最多的一种日志收集工作。目前大量的Java应用程序都使用着Lo4j 1.2.X版本，Log4j 1.X版本饱受诟病的原因是使用了大量的锁，在高并发的情况下影响了系统的性能。这篇简单提供一种思路，说说如何扩展一下Log4j，提升一下Log4j的性能。
网上有很多分析Log4j源码的文章，这里不重复说了，只简单分析一下最重要的几个组件。
先看一下Log4j的配置文件log4j.xml
<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
<appender name="rootAppender" class="org.apache.log4j.DailyRollingFileAppender">
<param name="File" value="test.log" />
<param name="DatePattern" value="'.'yyyy-MM-dd" />
<layout class="org.apache.log4j.PatternLayout">
<param name="ConversionPattern" value="%d{yyyy-MM-dd HH:mm:ss.SSS} %5p %c{1}:%L - %m%n" />
</layout>
</appender>
<logger name="com.test" additivity="false">
<level value="debug" />
<appender-ref ref="rootAppender" />
</logger>
<root>
<level value="debug" />
<appender-ref ref="rootAppender" />
</root>
</log4j:configuration>
从log4j.xml里面我们就可以看到Log4j最主要的几个组件
1. Logger，表示日志收集器，包含了各种Level下收集日志的方法，比如debug, info, error等。Logger的一个重要属性是additivity，表示是否附加到父Logger。Logger被设置成单根的树形结构，根就是Root，收集日志时，可以从叶子Logger一直往上直到Root。
Logger还包含了level表示级别，包含了一个appender列表，一个Logger可以对应多个Appender
2. Appender，表示如何处理日志，可以写本地文件，可以写远程文件，也可以输出到消息队列，等等。最常见的场景就是写本地文件。写文件的抽象是WriterAppender，封装了一个过滤器流QuietWriter。既然是过滤器流，那么可以包装节点流，默认的就是FileOutputStream流。Log4j也支持Buffer流，在WriterAppender的子类FileAppender里面有bufferedIO属性，如果采用Buffer流，就会在FileOutputStream外包一层BufferedWriter，最后在包到QuietWriter中。
public
synchronized
void setFile(String fileName, boolean append, boolean bufferedIO, int bufferSize)
throws IOException {
LogLog.debug("setFile called: "+fileName+", "+append);
if(bufferedIO) {
setImmediateFlush(false);
}
reset();
FileOutputStream ostream = null;
try {
ostream = new FileOutputStream(fileName, append);
} catch(FileNotFoundException ex) {
String parentName = new File(fileName).getParent();
if (parentName != null) {
File parentDir = new File(parentName);
if(!parentDir.exists() && parentDir.mkdirs()) {
ostream = new FileOutputStream(fileName, append);
} else {
throw ex;
}
} else {
throw ex;
}
}
Writer fw = createWriter(ostream);
if(bufferedIO) {
fw = new BufferedWriter(fw, bufferSize);
}
this.setQWForFiles(fw);
this.fileName = fileName;
this.fileAppend = append;
this.bufferedIO = bufferedIO;
this.bufferSize = bufferSize;
writeHeader();
LogLog.debug("setFile ended");
}
关于Appender，尤其是基于文件的FileAppender，一定要理解，一个FileAppender实例表示的是一个打开文件对象，在计算机底层知识拾遗（四）理解文件系统 中我们说了从Linux的角度来说，open调用就创建了一个打开文件对象，并返回了一个文件描述符，其他系统调用都用文件描述符来引用这个打开文件对象。打开文件对象包含了当前的读写位置，文件长度等等。 而在Java中并没有提供open操作，在Java中，new一个FileInputStream/FileOutputStream实例表示创建了一个打开文件对象。FileInputStream/FileOutputStream维护着当前的读写位置，还维护了文件描述符FileDescriptor对象和FileChannel对象。随便提一下，日志收集操作基本上都是对文件的Append，FileOutputStream支持文件的append打开方式。
可以看到， new一个FileOutputStream会创建一个FileDescriptor对象，并支持open和openAppend方式。这和Linux的open系统调用完成的功能是一样的。
public
class FileOutputStream extends OutputStream
{
/**
* The system dependent file descriptor. The value is
* 1 more than actual file descriptor. This means that
* the default value 0 indicates that the file is not open.
*/
private FileDescriptor fd;
private FileChannel channel= null;
private boolean append = false;
public FileOutputStream(File file, boolean append)
throws FileNotFoundException
{
String name = (file != null ? file.getPath() : null);
SecurityManager security = System.getSecurityManager();
if (security != null) {
security.checkWrite(name);
}
if (name == null) {
throw new NullPointerException();
}
fd = new FileDescriptor();
fd.incrementAndGetUseCount();
this.append = append;
if (append) {
openAppend(name);
} else {
open(name);
}
}
}
理解了一个FileOutputStream对象实例表示了一个打开文件之后，我们再来看一下操作系统内部为进程维护运行时文件数据结构的图。可以看到文件描述符指向了打开文件对象，打开文件对象又指向了inode对象，inode对象代表了一个文件在操作系统中的实例。
对于Java来说，JVM采用了单进程多线程的方式来支持并发，进程打开的所有文件对象对所有线程都是共享的。既然是共享，就涉及到多线程并发安全问题。在这里涉及到到3个地方存在并发安全问题。
1. 一个new FileOutputStream对象表示一个打开文件对象。那么当多个线程使用同一个FileOutputStream对象写文件时，就需要进行同步操作。
2. 当多个线程new了多个FileOutputStream对象对同一个文件进行写操作时，会出现并发问题，写入的顺序和位置都是不确定的。这种情况比较难同步，因为多个FileOutputStream对象是分散的，难以加锁。尽量避免这样写文件。当然也有解决方案，比如采用集中的地方来对外提供FileOutputStream对象，使用文件锁对inode对象加锁等等。
3. 当有另外的JVM进程对同一个文件进行操作时。这种情况只能通过加文件锁的方式来解决
Java提供了文件锁的支持，FileChannel提供了lock方法，并返回一个FileLock对象。
对于Log4j的使用场景，肯定是多线程并发的情况。第3种情况不用考虑。第二种情况比较复杂，所以只有第一种情况是适合的。Log4j也是采用了这一种方式。
可以把一个FileAppender对象理解成维护了一个打开文件对象，当Logger在多线程情况下把日志写入文件时，需要对Appender进行同步，也就是说对Logger加锁，保证使用同一个Logger对象时，一次只有一个线程使用这个FileAppender来写文件，避免了多线程情况下写文件错误。
就是下面这两段代码进行的同步操作，但是这也是Log4j被诟病最多的地方
Hierarchy.java
// 多个线程使用同一个Logger时，加锁
public
void callAppenders(LoggingEvent event) {
int writes = 0;
for(Category c = this; c != null; c=c.parent) {
// Protected against simultaneous call to addAppender, removeAppender,...
synchronized(c) {
if(c.aai != null) {
writes += c.aai.appendLoopOnAppenders(event);
}
if(!c.additive) {
break;
}
}
}
if(writes == 0) {
repository.emitNoAppenderWarning(this);
}
}
AppenderSkeleton.java
// 多个线程使用同一个Appender时，对Appender加锁
public
synchronized
void doAppend(LoggingEvent event) {
if(closed) {
LogLog.error("Attempted to append to closed appender named ["+name+"].");
return;
}
if(!isAsSevereAsThreshold(event.getLevel())) {
return;
}
Filter f = this.headFilter;
FILTER_LOOP:
while(f != null) {
switch(f.decide(event)) {
case Filter.DENY: return;
case Filter.ACCEPT: break FILTER_LOOP;
case Filter.NEUTRAL: f = f.getNext();
}
}
this.append(event);
}
这几个锁在高并发的情况下对系统的性能影响很大，会阻塞掉业务进程。而日志收集工作应该是作为一个辅助功能，不能影响主业务功能。
所以可以考虑从几个方面对Log4j进行扩展
1. 同步操作改异步操作，将对业务线程的影响减到最小。比如使用一个缓冲队列，Logger.debug操作只将LoggingEvent放入缓冲队列就结束，然后用单独的线程做消费者去消费缓冲队列
2. 使用阻塞队列来做缓冲，可以用基于数组的有界队列的，内存可控，但是在高并发下可能会阻塞生产者线程，也可以用基于链表的无界队列，用内存空间换性能
3. 消费这个缓冲队列时，可以持续消费，也可以采用定时批量消费的方式。持续消费就是使用一个无限循环的线程，只要缓冲队列不空，就从队列中取LoggingEvent来写文件。由于FileAppender维护了一个打开的FileOutputStream对象实例，不会重复执行打开文件操作，也就是说只打开一次，然后一直写，性能也可以接受。
定时批量消费就是采用一个定时的线程，一个时间段消费一次，一次消费一批数据。
Log4j本身就提供一个AsyncAppender类来异步收集日志。
1. 它采用阻塞式的有界缓冲队列，但是没有采用基于管城的BlockingQueue，而是使用了普通的ArrayList，并结合条件队列操作wait, notifty来实现阻塞队列。默认容量是128个LoggingEvent
2. 包含了一个AppenderAttachableImpl，这是一个Appender的列表结构，可以包含多个Appender
3. 使用了Dispatcher线程来做消费者
4. 消费者采用持续消费的方式，只要缓冲队列不为空，就唤醒消费者，先将buffer填充到一个数组中，然后单独对这个数组消费，buffer可以继续给生产者使用。
public class AsyncAppender extends AppenderSkeleton
implements AppenderAttachable {
public static final int DEFAULT_BUFFER_SIZE = 128；
private final List buffer = new ArrayList();
private final Map discardMap = new HashMap();
private int bufferSize = DEFAULT_BUFFER_SIZE;
/** Nested appenders. */
AppenderAttachableImpl aai;
private final AppenderAttachableImpl appenders;
public AsyncAppender() {
appenders = new AppenderAttachableImpl();
aai = appenders;
dispatcher =
new Thread(new Dispatcher(this, buffer, discardMap, appenders));
dispatcher.setDaemon(true);
dispatcher.setName("AsyncAppender-Dispatcher-" + dispatcher.getName());
dispatcher.start();
}
private static class Dispatcher implements Runnable {
private final AsyncAppender parent;
private final List buffer;
private final Map discardMap;
private final AppenderAttachableImpl appenders;
}
AsyncAppender的代码比较老，可以使用ArrayBlockingQueue对其进行改写，减少条件队列的操作。也可以进一步优化，采用无界的LinkedBlockingQueue，采用空间换性能的方式，这样不会阻塞生产者线程，也就是不阻塞业务线程。只有当LinedBlockingQueue为空时才会阻塞消费线程。这是可以接受的，因为消费线程不是业务线程，也防止了消费线程的空转。
还可以使用定时批量处理的生产者消费者模式来异步处理线程，下面是一个简单的基于定时线程池的定时批量异步日志收集器实现
1. 使用ScheduledExecutorService 定时线程池
2. 使用LinkedBlockingQueue无界阻塞队列做缓冲，不会阻塞生产者线程
3. 定时执行Dispatcher线程，每次把缓冲队列填充到一个局部的列表，只对局部列表中的数据处理，缓冲队列可以继续被生产者使用
4. appenders上的锁是防止添加删除appender时出并发问题，实际上Logger一旦创建，很少会在运行时取修改appenders，所以这个锁基本没有影响
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Appender;
import org.apache.log4j.AppenderSkeleton;
import org.apache.log4j.helpers.AppenderAttachableImpl;
import org.apache.log4j.spi.AppenderAttachable;
import org.apache.log4j.spi.LoggingEvent;
public class ScheduledAsyncAppender extends AppenderSkeleton implements
AppenderAttachable {
private final AppenderAttachableImpl appenders;
protected final ScheduledExecutorService executorService;
protected final LinkedBlockingQueue<LoggingEvent> queue;
protected long initialDelay;
protected long period;
protected TimeUnit unit;
public ScheduledAsyncAppender(long initialDelay, long period, TimeUnit unit) {
this.initialDelay = initialDelay;
this.period = period;
this.unit = unit;
appenders = new AppenderAttachableImpl();
// 初始化线程池
executorService = Executors.newScheduledThreadPool(1);
queue = new LinkedBlockingQueue<LoggingEvent>();
executorService.scheduleWithFixedDelay(new Dispatcher(), initialDelay,
period, unit);
}
/**
* 生产者线程只把LoggingEvent加入阻塞队列就返回，由于是无界队列，所以生产者线程不会阻塞，不影响业务
* */
@Override
protected void append(LoggingEvent event) {
try {
queue.put(event);
} catch (InterruptedException e) {
// if producer interrupted, do nothing
}
}
@Override
public void close() {
executorService.shutdown();
synchronized (appenders) {
Enumeration iter = appenders.getAllAppenders();
if (iter != null) {
while (iter.hasMoreElements()) {
Object next = iter.nextElement();
if (next instanceof Appender) {
((Appender) next).close();
}
}
}
}
}
private class Dispatcher implements Runnable {
@Override
public void run() {
List<LoggingEvent> logCache = new ArrayList<LoggingEvent>();
queue.drainTo(logCache);
for (LoggingEvent event : logCache) {
synchronized (appenders) {
appenders.appendLoopOnAppenders(event);
}
}
}
}
@Override
public void addAppender(final Appender newAppender) {
synchronized (appenders) {
appenders.addAppender(newAppender);
}
}
@Override
public Enumeration getAllAppenders() {
synchronized (appenders) {
return appenders.getAllAppenders();
}
}
@Override
public Appender getAppender(final String name) {
synchronized (appenders) {
return appenders.getAppender(name);
}
}
@Override
public boolean isAttached(final Appender appender) {
synchronized (appenders) {
return appenders.isAttached(appender);
}
}
@Override
public void removeAllAppenders() {
synchronized (appenders) {
appenders.removeAllAppenders();
}
}
@Override
public void removeAppender(final Appender appender) {
synchronized (appenders) {
appenders.removeAppender(appender);
}
}
@Override
public void removeAppender(final String name) {
synchronized (appenders) {
appenders.removeAppender(name);
}
}
@Override
public boolean requiresLayout() {
return false;
}
}
具体采用持续消费，还是定时批量消费，可以根据实际情况进行选择。定时批量消费时还可以优化，比如先把批量的日志日志变成一个大日志，一次只写一次文件文件，减少写文件的次数。
最后看一下如何运行时创建Logger，而不是使用log4j.xml配置文件的方式
static{
Layout patternLayout = new PatternLayout();
Appender appender = null;
String fileName = Configuration.getInstance().getLogFileName();
try {
appender = new DailyRollingFileAppender(patternLayout, fileName, "'.'yyyy-MM-dd");
} catch (IOException e) {
e.printStackTrace();
}
log.addAppender(rtracerAppender);
log.setLevel(Level.DEBUG);
log.setAdditivity(false);
}