logback异步输出日志的配置和源码分析

时间：2022-08-27 15:28:27

1，异步输出日志的配置

logback中的异步输出日志使用了AsyncAppender这个appender

配置方式如下：

<appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
        <fileNamePattern>logs/context-log.%d{yyyy-MM-dd}.log</fileNamePattern>
        <maxHistory>30</maxHistory>
    </rollingPolicy>
    <encoder charset="UTF-8">
        <pattern>[%-5level] %date --%thread-- [%logger] %msg %n</pattern>
    </encoder>
</appender>

<appender name ="ASYNC_FILE" class= "ch.qos.logback.classic.AsyncAppender">
    <discardingThreshold >0</discardingThreshold>
    <queueSize>1234</queueSize>
    <appender-ref ref = "FILE"/>
</appender>

AsyncAppender的父类是AsyncAppenderBase，用到的代码基本都在这个父类里面。

2，异步输出日志时会把信息放到BlockingQueue中

当执行logger.info()方法时，日志的内容会被放到AsyncAppenderBase里定义的一个BlockingQueue中，调用的是AsyncAppenderBase的append方法：

@Override
protected void append(E eventObject) {
  if (isQueueBelowDiscardingThreshold() && isDiscardable(eventObject)) {
    return;
  }
  preprocess(eventObject);
  put(eventObject);
}

然后append方法里的put方法：

private void put(E eventObject) {
  try {
    blockingQueue.put(eventObject);
  } catch (InterruptedException e) {
  }
}


3，从BlockingQueue中获取信息并写入到文件

在AsyncAppender的父类AsyncAppenderBase里面定义了一个叫Worker的内部类，这个类负责从BlockingQueue中取出信息并处理，Worker的定义如下：

class Worker extends Thread {

  public void run() {
    AsyncAppenderBase<E> parent = AsyncAppenderBase.this;
    AppenderAttachableImpl<E> aai = parent.aai;

    // loop while the parent is started
    while (parent.isStarted()) {
      try {
        E e = parent.blockingQueue.take();
        aai.appendLoopOnAppenders(e);
      } catch (InterruptedException ie) {
        break;
      }
    }

    addInfo("Worker thread will flush remaining events before exiting. ");
    for (E e : parent.blockingQueue) {
      aai.appendLoopOnAppenders(e);
    }

    aai.detachAndStopAllAppenders();
  }
}

另外AsyncAppenderBase还定义了Worker线程的start和stop方法，是重写的父类UnsynchronizedAppenderBase中的方法： 

@Override
public void start() {
  if (appenderCount == 0) {
    addError("No attached appenders found.");
    return;
  }
  if (queueSize < 1) {
    addError("Invalid queue size [" + queueSize + "]");
    return;
  }
  blockingQueue = new ArrayBlockingQueue<E>(queueSize);

  if (discardingThreshold == UNDEFINED)
    discardingThreshold = queueSize / 5;
  addInfo("Setting discardingThreshold to " + discardingThreshold);
  worker.setDaemon(true);
  worker.setName("AsyncAppender-Worker-" + worker.getName());
  // make sure this instance is marked as "started" before staring the worker Thread
  super.start();
  worker.start();
}

@Override
public void stop() {
  if (!isStarted())
    return;

  // mark this appender as stopped so that Worker can also processPriorToRemoval if it is invoking aii.appendLoopOnAppenders
  // and sub-appenders consume the interruption
  super.stop();

  // interrupt the worker thread so that it can terminate. Note that the interruption can be consumed
  // by sub-appenders
  worker.interrupt();
  try {
    worker.join(1000);
  } catch (InterruptedException e) {
    addError("Failed to join worker thread", e);
  }
}

可以看到，在while循环期间，Worker从blockingQueue里面拿出一个元素并进行处理，调用了AppenderAttachableImpl的appendLoopOnAppenders方法，该方法是这么写的： 

/**
 * Call the <code>doAppend</code> method on all attached appenders.
 */
public int appendLoopOnAppenders(E e) {
  int size = 0;
    for (Appender<E> appender : appenderList) {
      appender.doAppend(e);
      size++;
    }
  return size;
}

代码里面调用了所有Appender的doAppend方法，Appender是个接口，看一下其中一个实现了Appender接口的类UnsynchronizedAppenderBase，这里面的doAppend方法是这样的： 

public void doAppend(E eventObject) {
  // WARNING: The guard check MUST be the first statement in the
  // doAppend() method.

  // prevent re-entry.
  if (Boolean.TRUE.equals(guard.get())) {
    return;
  }

  try {
    guard.set(Boolean.TRUE);

    if (!this.started) {
      if (statusRepeatCount++ < ALLOWED_REPEATS) {
        addStatus(new WarnStatus(
            "Attempted to append to non started appender [" + name + "].",
            this));
      }
      return;
    }

    if (getFilterChainDecision(eventObject) == FilterReply.DENY) {
      return;
    }

    // ok, we now invoke derived class' implementation of append
    this.append(eventObject);

  } catch (Exception e) {
    if (exceptionCount++ < ALLOWED_REPEATS) {
      addError("Appender [" + name + "] failed to append.", e);
    }
  } finally {
    guard.set(Boolean.FALSE);
  }
}

这段代码，经过了一堆的判断和设置，调用了append方法，自己类里的append方法是个抽象方法，具体的实现写在了他的子类OutputStreamAppender里，代码如下： 

@Override
protected void append(E eventObject) {
  if (!isStarted()) {
    return;
  }

  subAppend(eventObject);
}

然后是subAppend方法：

/**
 * Actual writing occurs here.
 * <p>
 * Most subclasses of <code>WriterAppender</code> will need to override this
 * method.
 * 
 * @since 0.9.0
 */
protected void subAppend(E event) {
  if (!isStarted()) {
    return;
  }
  try {
    // this step avoids LBCLASSIC-139
    if (event instanceof DeferredProcessingAware) {
      ((DeferredProcessingAware) event).prepareForDeferredProcessing();
    }
    // the synchronization prevents the OutputStream from being closed while we
    // are writing. It also prevents multiple threads from entering the same
    // converter. Converters assume that they are in a synchronized block.
    lock.lock();
    try {
      writeOut(event);
    } finally {
      lock.unlock();
    }
  } catch (IOException ioe) {
    // as soon as an exception occurs, move to non-started state
    // and add a single ErrorStatus to the SM.
    this.started = false;
    addStatus(new ErrorStatus("IO failure in appender", this, ioe));
  }
}

然后是writeOut方法： 

protected void writeOut(E event) throws IOException {
  this.encoder.doEncode(event);
}

调用了encoder的doEncode方法，encoder的类是Encoder，Encoder是OutputStreamAppender定义的最终负责写日志的接口，由EchoEncoder类实现，他的doEncode方法是这么写的： 

public void doEncode(E event) throws IOException {
  String val = event + CoreConstants.LINE_SEPARATOR;
  outputStream.write(val.getBytes());
  // necessary if ResilientFileOutputStream is buffered
  outputStream.flush();
}

用OutputStream写日志，完了flush

至此日志写入完毕

 

4，一些可能有用的配置方式

1，blockingQueue长度。

blockingQueue长度决定了队列能放多少信息，在默认的配置下，如果blockingQueue放满了，后续想要输出日志的线程会被阻塞，直到Worker线程处理掉队列中的信息为止。根据实际情况适当调整队列长度，可以防止线程被阻塞。

2，immediateFlush=false。不立即清空输出流

immediateFlush参数可以配置在<appender>里面，默认是true，代表是否立即刷新OutputStream中的信息。如果设置为false，会在OutputStream放满或隔断时间进行flush，具体由OutputStream类决定。据说设置为false之后输出日志的效率能提高为原来的4倍。

官网说：setting thisproperty to 'false' is likely to quadruple (your mileage may vary) loggingthroughput.

3，neverBlock=true。队列满了也不卡线程

neverBlock参数可以配置在<appender>里面，默认是false，代表在队列放满的情况下是否卡住线程。也就是说，如果配置neverBlock=true，当队列满了之后，后面阻塞的线程想要输出的消息就直接被丢弃，从而线程不会阻塞。这个配置用于线程很重要，不能卡顿，而且日志又不是很重要的场景，因为很有可能会丢日志。

4，自定义appender

开发者可以自己写一个appender类，需要继承AppenderBase<LoggingEvent>类并重写append(LoggingEventeventObject)方法，然后像别的appender一样配置到logback.xml里面，就可以定义自己的日志输出方式了。

 

 

标签：

相关文章