这一节我们讲解 logback 异步日志打印中 ArrayBlockingQueue 的使用。

异步日志打印模型概述

在高并发、高流量并且响应时间要求比较小的系统中同步打印日志已经满足不了需求了，这是因为打印日志本身是需要写磁盘的，写磁盘的操作会暂时阻塞调用打印日志的业务线程，这会造成调用线程的 rt 增加。如图 11-1 所示为同步日志打印模型。

ArrayBlockingQueue 的使用 - 图1

图 11-1

同步日志打印模型的缺点是将日志写入磁盘的操作是业务线程同步调用完成的，那么是否可以让业务线程把要打印的日志任务放入一个队列后直接返回，然后使用一个线程专门负责从队列中获取日志任务并将其写入磁盘呢？这样的话，业务线程打印日志的耗时就仅仅是把日志任务放入队列的耗时了，其实这就是 logback 提供的异步日志打印模型要做的事，具体如图 11-2 所示。

ArrayBlockingQueue 的使用 - 图2

图 11-2

由图 11-2 可知，其实 logback 的异步日志模型是一个多生产者-单消费者模型，其通过使用队列把同步日志打印转换为了异步，业务线程只需要通过调用异步 appender 把日志任务放入日志队列，而日志线程则负责使用同步的 appender 进行具体的日志打印。日志打印线程只需要负责生产日志并将其放入队列，而不需要关心消费线程何时把日志具体写入磁盘。

异步日志与具体实现

1．异步日志

一般配置同步日志打印时会在 logback 的 xml 文件里面配置如下内容。

//(1)配置同步日志打印 appender
  <appender name=「PROJECT」 class=「ch.qos.logback.core.FileAppender」>
      <file>project.log</file>
      <encoding>UTF-8</encoding>
      <append>true</append>
      <rollingPolicy class=「ch.qos.logback.core.rolling.TimeBasedRollingPolicy」>
          <！ -- daily rollover -->
            <fileNamePattern>project.log.%d{yyyy-MM-dd}</fileNamePattern>
            <！ -- keep 7 days『 worth of history -->
            <maxHistory>7</maxHistory>
        </rollingPolicy>
        <layout class=「ch.qos.logback.classic.PatternLayout」>
            <pattern><！ [CDATA[
%n%-4r [%d{yyyy-MM-dd HH:mm:ss}] %X{productionMode} - %X{method}
%X{requestURIWithQueryString} [ip=%X{remoteAddr}， ref=%X{referrer}，
ua=%X{userAgent}， sid=%X{cookie.JSESSIONID}]%n  %-5level %logger{35} - %m%n
            ]]></pattern>
        </layout>
    </appender>
  //(2) 设置 logger
    <logger name=「PROJECT_LOGGER」 additivity=「false」>
        <level value=「WARN」 />
        <appender-ref ref=「PROJECT」 />
    </logger>

然后以如下方式使用。

ArrayBlockingQueue 的使用 - 图3

要把同步日志打印改为异步则需要修改 logback 的 xml 配置文件为如下所示。

    <appender name="PROJECT" class="ch.qos.logback.core.FileAppender">
        <file>project.log</file>
        <encoding>UTF-8</encoding>
        <append>true</append>
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <! -- daily rollover -->
            <fileNamePattern>project.log.%d{yyyy-MM-dd}</fileNamePattern>
            <! -- keep 7 days' worth of history -->
            <maxHistory>7</maxHistory>
        </rollingPolicy>
        <layout class="ch.qos.logback.classic.PatternLayout">
            <pattern><! [CDATA[
%n%-4r [%d{yyyy-MM-dd HH:mm:ss}] %X{productionMode} - %X{method}
%X{requestURIWithQueryString} [ip=%X{remoteAddr}, ref=%X{referrer},
ua=%X{userAgent}, sid=%X{cookie.JSESSIONID}]%n  %-5level %logger{35} - %m%n
            ]]></pattern>
        </layout>
    </appender>
    <appender name="asyncProject" class="ch.qos.logback.classic.AsyncAppender">
        <discardingThreshold>0</discardingThreshold>
        <queueSize>1024</queueSize>
        <neverBlock>true</neverBlock>
        <appender-ref ref="PROJECT" />
    </appender>
    <logger name="PROJECT_LOGGER" additivity="false">
        <level value="WARN" />
        <appender-ref ref="asyncProject" />
    </logger>

由以上代码可以看出，AsyncAppender 是实现异步日志的关键，下一节主要讲它的内部实现。

2．异步日志实现原理

本文使用的 logback-classic 的版本为 1.0.13。我们首先从 AsyncAppender 的类图结构来认识下 AsyncAppender 的组件构成，如图 11-3 所示。

ArrayBlockingQueue 的使用 - 图4

图 11-3

由图 11-3 可知，AsyncAppender 继承自 AsyncAppenderBase，其中后者具体实现了异步日志模型的主要功能，前者只是重写了其中的一些方法。由该图可知，logback 中的异步日志队列是一个阻塞队列，其实就是有界阻塞队列 ArrayBlockingQueue，其中 queueSize 表示有界队列的元素个数，默认为 256 个。

worker 是个线程，也就是异步日志打印模型中的单消费者线程。aai 是一个 appender 的装饰器，里面存放同步日志的 appender，其中 appenderCount 记录 aai 里面附加的同步 appender 的个数。neverBlock 用来指示当日志队列满时是否阻塞打印日志的线程。discardingThreshold 是一个阈值，当日志队列里面的空闲元素个数小于该值时，新来的某些级别的日志会被直接丢弃，下面会具体讲。

首先我们来看何时创建日志队列，以及何时启动消费线程，这需要看 AsyncAppenderBase 的 start 方法。该方法在解析完配置 AsyncAppenderBase 的 xml 的节点元素后被调用。

public void start() {
  ...
  //（1）日志队列为有界阻塞队列 
  blockingQueue = new ArrayBlockingQueue<E>（queueSize）；
  //(2)如果没设置 discardingThreshold 则设置为队列大小的 1/5
  if （discardingThreshold == UNDEFINED）
    discardingThreshold = queueSize / 5；
  //(3)设置消费线程为守护线程，并设置日志名称 
  worker.setDaemon（true）；
  worker.setName(「AsyncAppender-Worker-」 + worker.getName())；
  //(4)设置启动消费线程 
  super.start()；
  worker.start()；
}

由以上代码可知，logback 使用的是有界队列 ArrayBlockingQueue，之所以使用有界队列是考虑内存溢出问题。在高并发下写日志的 QPS 会很高，如果设置为无界队列，队列本身会占用很大的内存，很可能会造成 OOM。

这里消费日志队列的 worker 线程被设置为守护线程，这意味着当主线程运行结束并且当前没有用户线程时，该 worker 线程会随着 JVM 的退出而终止，而不管日志队列里面是否还有日志任务未被处理。另外，这里设置了线程的名称，这是个很好的习惯，因为在查找问题时会很有帮助，根据线程名字就可以定位线程。

既然是有界队列，那么肯定需要考虑队列满的问题，是丢弃老的日志任务，还是阻塞日志打印线程直到队列有空余元素呢？要回答这个问题，我们需要看看具体进行日志打印的 AsyncAppenderBase 的 append 方法。

protected void append（E eventObject） {
  //（5）调用 AsyncAppender 重写的 isDiscardable 方法 
  if （isQueueBelowDiscardingThreshold() && isDiscardable(eventObject)） {
      return；
  }
  ...
  //（6）将日志任务放入队列 
  put（eventObject）；
}
    private boolean isQueueBelowDiscardingThreshold() {
        return （blockingQueue.remainingCapacity() < discardingThreshold）；
    }

其中代码（5）调用了 AsyncAppender 重写的 isDiscardable 方法，该方法的具体内容为

    //(7)
    protected boolean isDiscardable(ILoggingEvent event) {
        Level level = event.getLevel();
        return level.toInt() <= Level.INFO_INT;
    }

结合代码（5）和代码（7）可知，如果当前日志的级别小于等于 INFO_INT 并且当前队列的剩余容量小于 discardingThreshold 则会直接丢弃这些日志任务。

下面看具体代码（6）中的 put 方法。

    private void put(E eventObject) {
        //(8)
        if (neverBlock) {
            blockingQueue.offer(eventObject);
        } else {
            try {//(9)
                blockingQueue.put(eventObject);
            } catch (InterruptedException e) {
                // Interruption of current thread when in doAppend method should not
                //  be consumed
                // by AsyncAppender
                Thread.currentThread().interrupt();
            }
        }
    }

如果 neverBlock 被设置为 false（默认为 false）则会调用阻塞队列的 put 方法，而 put 是阻塞的，也就是说如果当前队列满，则在调用 put 方法向队列放入一个元素时调用线程会被阻塞直到队列有空余空间。这里可以看下 put 方法的实现。

    public void put（E e） throws InterruptedException {
        ...
        final ReentrantLock lock = this.lock；
        lock.lockInterruptibly()；
        try {
            //如果队列满，则调用 await 方法阻塞当前调用线程 
            while （count == items.length）
                notFull.await()；
            enqueue（e）；
        } finally {
            lock.unlock()；
        }
    }

这里有必要解释下代码（9），当日志队列满时 put 方法会调用 await（）方法阻塞当前线程，而如果其他线程中断了该线程，那么该线程会抛出 InterruptedException 异常，并且当前的日志任务就会被丢弃。在 logback-classic 的 1.2.3 版本中，则添加了不对中断进行响应的方法。

    private void put(E eventObject) {
        if (neverBlock) {
            blockingQueue.offer(eventObject);
        } else {
            putUninterruptibly(eventObject);
          }
     }
     private void putUninterruptibly(E eventObject) {
        boolean interrupted = false;
        try {
            while (true) {
                try {
                    blockingQueue.put(eventObject);
                    break;
                } catch (InterruptedException e) {
                    interrupted = true;
                }
            }
        } finally {
            if (interrupted) {
                Thread.currentThread().interrupt();
            }
        }
    }

如果当前日志打印线程在调用 blockingQueue.put 时被其他线程中断，则只是记录中断标志，然后继续循环调用 blockingQueue.put，尝试把日志任务放入日志队列。新版本的这个实现通过使用循环保证了即使当前线程被中断，日志任务最终也会被放入日志队列。

如果 neverBlock 被设置为 true 则会调用阻塞队列的 offer 方法，而该方法是非阻塞的，所以如果当前队列满，则会直接返回，也就是丢弃当前日志任务。这里回顾下 offer 方法的实现。

    public boolean offer（E e） {
        ...
        final ReentrantLock lock = this.lock；
        lock.lock()；
        try {
            //如果队列满则直接返回 false。
            if （count == items.length）
                return false；
            else {
                enqueue（e）；
                return true；
            }
        } finally {
            lock.unlock()；
        }
    }

最后来看 addAppender 方法都做了什么。

    public void addAppender(Appender<E> newAppender) {
        if (appenderCount == 0) {
            appenderCount++;
            ...
            aai.addAppender(newAppender);
        } else {
            addWarn("One and only one appender may be attached to AsyncAppender.");
            addWarn("Ignoring additional appender named [" + newAppender.getName() + "]");
        }
    }

由如上代码可知，一个异步 appender 只能绑定一个同步 appender。这个 appender 会被放到 AppenderAttachableImpl 的 appenderList 列表里面。

到这里我们已经分析完了日志生产线程把日志任务放入日志队列的实现，下面一起来看消费线程是如何从队列里面消费日志任务并将其写入磁盘的。由于消费线程是一个线程，所以就从 worker 的 run 方法开始。

class Worker extends Thread {
      public void run() {
          AsyncAppenderBase<E> parent = AsyncAppenderBase.this；
          AppenderAttachableImpl<E> aai = parent.aai；
          //（10）一直循环直到该线程被中断 
          while （parent.isStarted()） {
              try {//（11）从阻塞队列获取元素 
                  E e = parent.blockingQueue.take()；
                  aai.appendLoopOnAppenders（e）；
              } catch （InterruptedException ie） {
                  break；
                }
            }
          //（12）到这里说明该线程被中断，则把队列里面的剩余日志任务 
          //刷新到磁盘 
          for （E e : parent.blockingQueue） {
              aai.appendLoopOnAppenders（e）；
              parent.blockingQueue.remove（e）；
            }
          ...
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               ..        }
  }

其中代码（11）使用 take 方法从日志队列获取一个日志任务，如果当前队列为空则当前线程会被阻塞直到队列不为空才返回。获取到日志任务后会调用 AppenderAttachableImpl 的 aai.appendLoopOnAppenders 方法，该方法会循环调用通过 addAppender 注入的同步日志，appener 具体实现把日志打印到磁盘。

小结

本节结合 logback 中异步日志的实现介绍了并发组件 ArrayBlockingQueue 的使用，包括 put、offer 方法的使用场景以及它们之间的区别，take 方法的使用，同时也介绍了如何使用 ArrayBlockingQueue 来实现一个多生产者-单消费者模型。另外使用 ArrayBlockingQueue 时需要注意合理设置队列的大小以免造成 OOM，队列满或者剩余元素比较少时，要根据具体场景制定一些抛弃策略以避免队列满时业务线程被阻塞。