log4j异步那些事(1)--AsyncAppender

简介Logger和Appender的异步化配置和基本原理

前面的博客里，我简单介绍过了Log4j2的简单配置和其中基本组件LogManager，LoggerContext以及Configuration的机制和流程。而还有两个关键的组件Logger和Appender，他们是对msg做真正处理的关键组件，在众多类型的Logger和Appender中，我主要想把目光集中在其中比较特定的，异步的Logger和Appender。如果大家平时只是简单使用Log4j,可能对异步的Logger和Appender有些分不清，或者并未配置过这类的异步化组件。我将用两篇文章分别介绍异步Appender和异步Logger。

Logger和Appender的关系

前面的博客中，我介绍过，记录日志的代码，这里为了讲述方便，我重新把代码放上来

Logger logger = LogManager.getLogger(loggerName);
logger.info(msg);

这篇文章主要聚焦于异步性的实现，所以对于获取logger和调用info操作的通用流程我直接用两张流程图给出：

了解了通用流程后，我将分别介绍异步Appender和异步Logger的原理

异步Appender AsyncAppender

配置

我们以官方文档的两个配置为例来介绍异步Appender的配置，配置文件如下

<Configuration status="warn" name="MyApp" packages="">
  <Appenders>
    <File name="MyFile" fileName="logs/app.log">
      <PatternLayout>
        <Pattern>%d %p %c{1.} [%t] %m%n</Pattern>
      </PatternLayout>
    </File>
    <Async name="Async">
      <AppenderRef ref="MyFile"/>
    </Async>
  </Appenders>
  <Loggers>
    <Root level="error">
      <AppenderRef ref="Async"/>
    </Root>
  </Loggers>
</Configuration>

<Configuration name="LinkedTransferQueueExample">
  <Appenders>
    <List name="List"/>
    <Async name="Async" bufferSize="262144">
      <AppenderRef ref="List"/>
      <LinkedTransferQueue/>
    </Async>
  </Appenders>
  <Loggers>
    <Root>
      <AppenderRef ref="Async"/>
    </Root>
  </Loggers>
</Configuration>

配置中配置了两个Appender，一个为FileAppender另一个则是异步Appender，使用 <Async> 标签进行声明，在AsyncAppender中引用了FileAppender，那么存在疑问，一般的Appender都会有一个明确的输出位置，而对于这个异步Appender都需要引用一个其他的Appender才将msg最终输出。下面的介绍将会彻底解开这个疑问。

架构

下图展示了AsyncAppender的架构

AsyncAppender的核心部件是一个阻塞队列，logger将数据通过append方法放入到阻塞队列中，随后后台线程从队列中取出数据然后进行后续的操作

机制

上文简单介绍了架构，下面从源码角度来详细阐述AsyncAppender的流程

成员变量

public final class AsyncAppender extends AbstractAppender {

    private static final int DEFAULT_QUEUE_SIZE = 128;
    private static final LogEvent SHUTDOWN = new AbstractLogEvent() {
    };

    private static final AtomicLong THREAD_SEQUENCE = new AtomicLong(1);

    private final BlockingQueue<LogEvent> queue;
    private final int queueSize;
    private final boolean blocking;
    private final long shutdownTimeout;
    private final Configuration config;
    private final AppenderRef[] appenderRefs;
    private final String errorRef;
    private final boolean includeLocation;
    private AppenderControl errorAppender;
    private AsyncThread thread;
    private AsyncQueueFullPolicy asyncQueueFullPolicy;
}

可以看到其中的一些关键属性有一个阻塞队列queue，一个后台线程thread，一个AppenderRef的数组appenderRefs以及一个关键属性blocking。对于一个任何一个Appender对象，我们都应该关注他的append()而方法，对于一个后台线程，重要的方法则是run方法，对于AsyncAppender，这两个方法刚好对应了AsyncAppender的两个核心步骤，即放入消息以及处理消息，下面将分别说明。

append()方法

废话不多说，线上代码：

@Override
public void append(final LogEvent logEvent) {
    if (!isStarted()) {
        throw new IllegalStateException("AsyncAppender " + getName() + " is not active");
    }
    if (!Constants.FORMAT_MESSAGES_IN_BACKGROUND) { // LOG4J2-898: user may choose
        logEvent.getMessage().getFormattedMessage(); // LOG4J2-763: ask message to freeze parameters

    }
    final Log4jLogEvent memento = Log4jLogEvent.createMemento(logEvent, includeLocation);
    if (!transfer(memento)) {
        if (blocking) {
            // delegate to the event router (which may discard, enqueue and block, or log in current thread)
            final EventRoute route = asyncQueueFullPolicy.getRoute(thread.getId(), memento.getLevel());
            route.logMessage(this, memento);
        } else {
            error("Appender " + getName() + " is unable to write primary appenders. queue is full");
            logToErrorAppenderIfNecessary(false, memento);
        }
    }
}

可以看到这个appende方法流程并不复杂，只有以下两步：

1、创建LogEvent的复制对象memento
2、将event放入队列

虽然只有简单两步，也需要注意一个边界情况，那就是当阻塞队列满时Appender的处理，这里我首先给出流程，然后结合代码进行简要说明。

如流程图所示，首先会判断用户是否设置了blocking选项，如果未选择blocking选项，则Appender直接会将msg放入errorAppender中，如果用户没有配置这些Appender，则会直接丢弃这些消息，如果设置了这个属性，则会按照一定的策略来处理这些消息。策略可以分为3种，他们分别为：

1、Default---等待直到队列有空闲，退化为同步操作
2、Discard---按照日志级别丢弃一部分日志
3、用户自定义(需要实现AsyncQueueFullPolicy接口)

run()方法

当使用append方法将消息放入阻塞队列后，后台的线程将会一步的进行处理，这也就是异步线程的run方法的功能所在，首先简单看其数据结构

private class AsyncThread extends Log4jThread {
    private volatile boolean shutdown = false;
    private final List<AppenderControl> appenders;
    private final BlockingQueue<LogEvent> queue;
}

AsyncThread这是一个内部类，其中包含一个外部Appender类的阻塞队列，还有对应的AsyncAppender所引用的Appender。接下来我们详细看其中的run方法

public void run() {
     while (!shutdown) {
         LogEvent event;
         try {
             event = queue.take();
             if (event == SHUTDOWN) {
                 shutdown = true;
                 continue;
             }
         } catch (final InterruptedException ex) {
             break; // LOG4J2-830
         }
         event.setEndOfBatch(queue.isEmpty());
         final boolean success = callAppenders(event);
         if (!success && errorAppender != null) {
             try {
                 errorAppender.callAppender(event);
             } catch (final Exception ex) {
                 // Silently accept the error.
             }
         }
     }
     // Process any remaining items in the queue.
     LOGGER.trace("AsyncAppender.AsyncThread shutting down. Processing remaining {} queue events.",
         queue.size());
     int count = 0;
     int ignored = 0;
     while (!queue.isEmpty()) {
         try {
             final LogEvent event = queue.take();
             if (event instanceof Log4jLogEvent) {
                 final Log4jLogEvent logEvent = (Log4jLogEvent) event;
                 logEvent.setEndOfBatch(queue.isEmpty());
                 callAppenders(logEvent);
                 count++;
             } else {
                 ignored++;
                 LOGGER.trace("Ignoring event of class {}", event.getClass().getName());
             }
         } catch (final InterruptedException ex) {
             // May have been interrupted to shut down.
             // Here we ignore interrupts and try to process all remaining events.
         }
     }
     LOGGER.trace("AsyncAppender.AsyncThread stopped. Queue has {} events remaining. "
         + "Processed {} and ignored {} events since shutdown started.", queue.size(), count, ignored);
}

可以看到，异步线程的逻辑比较简单，该线程会一直尝试从阻塞队列中获取logEvent数据，如果能够成功获取数据，则会调用AppenderRef所引用Appender的append方法，通过这个方法，我们可以看到，实际上，AsyncAppender可以看做一个中转站，其作用仅仅将消息的处理异步化，当消息放入阻塞队列后，info方法就能返回成功，这样能够大幅提高日志记录的吞吐，同时，用户可以自行权衡性能与日志收集质量上进行权衡(设置blocking选项)，此外，用户还可以设置不同类型的阻塞队列已到达更好的日志记录吞吐。

再回配置

最后，让我们整体来看AsyncAppender所支持的所有配置项以及其中每个配置项的作用

至此，我结合源码简单介绍了AsyncAppnder的使用配置以及基本原理，在下一篇文章中，我将介绍另一个异步化组件AsyncLogger