Logback的AsyncAppender与RollingFileAppender流…

2020-01-01 16:07:35来源：博客园阅读 ()

Logback的AsyncAppender与RollingFileAppender流程解析

近期工作中涉及到文件记录、文件翻转等操作，思考有没有成熟的代码以便参考.
因此，第一时间就联想到Logback的AsyncAppender以及RollingFileAppender.

AsyncAppender:通过队列储存日志事件，启动Worker线程读取日志事件并写入关联的Appender中；
RollingFileAppender:当日志文件满足设定的翻滚条件时，对文件进行翻滚操作.

PS: AsyncAppender可以与RollingFileAppender结合使用，提升日志事件写入效率.

1 AsyncAppender

public class AsyncAppender extends AsyncAppenderBase<ILoggingEvent> {

    // 省略部分功能
    boolean includeCallerData = false;

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

    protected void preprocess(ILoggingEvent eventObject) {
        eventObject.prepareForDeferredProcessing();
        if (includeCallerData)
            eventObject.getCallerData();
    }
}

(1)isDiscardable:确定日志事件是否可以丢弃(当缓冲队列达到上限时，出于性能考虑需要丢弃诸如TRACE、DEBUG级别日志)；
(2)preprocess:预处理日志事件（包括格式化msg，线程名称，MDC中存储的数据），如果includeCallerData为true，则需要通过日志事件的堆栈信息，获取日志所在的文件，行号等信息；

因为AsyncAppender的绝大部分功能由AsyncAppenderBase中实现，因此接下来主要讲解AsyncAppenderBase的功能点.

public class AsyncAppenderBase<E> extends UnsynchronizedAppenderBase<E> implements AppenderAttachable<E> {

    BlockingQueue<E> blockingQueue;

    public static final int DEFAULT_QUEUE_SIZE = 256;
    int queueSize = DEFAULT_QUEUE_SIZE;

    int appenderCount = 0;

    static final int UNDEFINED = -1;
    int discardingThreshold = UNDEFINED;
    boolean neverBlock = false;

    Worker worker = new Worker();

    public static final int DEFAULT_MAX_FLUSH_TIME = 1000;
    int maxFlushTime = DEFAULT_MAX_FLUSH_TIME;
}

以上是AsyncAppenderBase的主要属性：

blockingQueue:阻塞队列，存储日志事件，等待被各Appender消费；
queueSize：队列大小，如果没有设置，则默认为256；
discardingThreshold：日志事件的丢弃阈值，当队列中的事件数量超过阈值后，则会丢弃诸如TRACE/DEBUG级别日志；
worker：消费者线程(单消费者足够了)；
maxFlushTime：程序结束时，worker线程的退出时间(从而保证尽量将队列中的日志事件写至文件中).

1.1 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;
                }
            }

            for (E e : parent.blockingQueue) {
                aai.appendLoopOnAppenders(e);
                parent.blockingQueue.remove(e);
            }

            aai.detachAndStopAllAppenders();
        }
    }

Worker线程比较简单，其主要功能就是判断Appeneder是否处于运行状态(parent.isStarted()):

运行：读取阻塞队列中的数据，通过AppenderAttachableImpl分发至各Appender中;
不运行：循环读取缓冲队列中的残余数据，通过AppenderAttachableImpl分发至各Appender中，最后关闭关联的Appender.

1.2 启动appender

 @Override
    public void start() {

        // 省略部分校验代码
        blockingQueue = new ArrayBlockingQueue<E>(queueSize);

        if (discardingThreshold == UNDEFINED)
            discardingThreshold = queueSize / 5;
        worker.setDaemon(true);
        worker.setName("AsyncAppender-Worker-" + getName());
        super.start();
        worker.start();
    }

主要步骤：

(1) 根据设置的队列大小，创建缓冲队列大小；

(2) 如果未设置discardingThreshold，则设置discardingThreshold阈值为缓冲队列大小的4/5(1-1/5);

(3) 设置worker线程为守护线程，设置线程名称；

(4) 启动Appender，启动worker线程读取数据(需要确保Appender在worker线程前启动).

1.3 关闭appender

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

        super.stop();

        // interrupt the worker thread so that it can terminate. Note that the interruption can be consumed
        // by sub-appenders
        worker.interrupt();

        InterruptUtil interruptUtil = new InterruptUtil(context);

        try {
            interruptUtil.maskInterruptFlag();
            worker.join(maxFlushTime);
            // check to see if the thread ended and if not add a warning message
            if (worker.isAlive()) {
                addWarn("Max queue flush timeout (" + maxFlushTime + " ms) exceeded. Approximately " + blockingQueue.size()
                                + " queued events were possibly discarded.");
            } else {
                addInfo("Queue flush finished successfully within timeout.");
            }
        } catch (InterruptedException e) {
            int remaining = blockingQueue.size();
            addError("Failed to join worker thread. " + remaining + " queued events may be discarded.", e);
        } finally {
            interruptUtil.unmaskInterruptFlag();
        }
    }

主要步骤:

(1) super.stop():关闭Appender,worker线程执行退出逻辑；

(2) worker.interrupt()：给worker线程设置中断标志(worker中未检测中断标志，因此保持继续运行状态)，设置的意义是让当前appender关联的sub-appender消费，从而安全的关闭sub-appender;

(3) interruptUtil.maskInterruptFlag():清除当前线程的interrupt状态；

(4) worker.join(maxFlushTime):等待worker线程退出，结束其生命周期；

(5) 判断worker线程是否存活，如果存活说明阻塞队列中仍存有部分日志事件未被写入文件等载体中,记录消息；

(6) interruptUtil.unmaskInterruptFlag():恢复worker线程的interrupt状态.

划重点:

上述代码中，对worker线程的中断标志进行了若干次操作：

(1) interrupt:中断worker关联的sub-appender；

(2) interruptUtil.maskInterruptFlag: 取消中断标志（在线程标志为true的状态下，join操作会立即返回），因此为了确保join操作有效，需要清除worker线程的interrupt标志；

(3) interruptUtil.unmaskInterruptFlag：结束操作，恢复worker线程的interrupt标志.

1.4 添加日志事件

日志事件的添加，实质是就是往阻塞队列中插入日志事件.
根据阻塞队列接口，分两种插入方式：

(1) offer：非阻塞插入，插入失败不进行处理(存在丢日志可能性)；

(2) put：阻塞插入(插入失败后会循环进行再次插入操作).

2 RollingFileAppender

public class RollingFileAppender<E> extends FileAppender<E> {
    File currentlyActiveFile;
    TriggeringPolicy<E> triggeringPolicy;
    RollingPolicy rollingPolicy;
}

主要属性：

currentlyActiveFile:当前日志写入文件；
triggeringPolicy：日志文件触发策略(触发条件包括：单个日志文件大小，文件中日志文件总体积，翻转时刻等条件)；
rollingPolicy：翻转策略(主要确定日志文件翻转文件名，存储路径等信息).

需要注意，如果没有日志事件写入，那么即使日志文件达到时间或者大小的触发条件，也不会创建相应的新日志文件.

2.1 添加日志事件

    protected void subAppend(E event) {
        synchronized (triggeringPolicy) {
            if (triggeringPolicy.isTriggeringEvent(currentlyActiveFile, event)) {
                rollover();
            }
        }
        super.subAppend(event);
    }

主要步骤:

(1) 判断当前文件是否达到触发条件，如果是则翻转文件(使用synchronized加锁以保证同一时间段，只有一个线程进行文件的翻转操作)；

(2) 调用基类的subAppend方法，将日志文件写入BufferedOutputStream中.

翻转文件：

    public void rollover() {
        lock.lock();
        try {
            this.closeOutputStream();
            attemptRollover();
            attemptOpenFile();
        } finally {
            lock.unlock();
        }
    }

由以上代码可知，翻转文件涉及到以下几个操作：

(1) 关闭当前BufferedOutputStream；

(2) attemptRollover: 进行文件翻转（重命名已写入文件名，根据需求压缩日志文件，根据日志文件夹总大小以及日期删除文件等）；

(3) attemptOpenFile:根据翻转条件，确定新日志文件名称，并创建对应的日志文件供后续写入.

2.2 SizeAndTimeBasedRollingPolicy

工作中，经常使用的文件翻转工具类为SizeAndTimeBasedRollingPolicy(实现了RollingPolicy以及TriggeringPolicy接口)，顾名思义，其根据时间和文件大小确定日志文件的翻转触发条件.

2.2.1 日志触发

需要注意，触发器实际定义于SizeAndTimeBasedFNATP类中.

    @Override
    public boolean isTriggeringEvent(File activeFile, final E event) {

        long time = getCurrentTime();

        // first check for roll-over based on time
        if (time >= nextCheck) {
            Date dateInElapsedPeriod = dateInCurrentPeriod;
            elapsedPeriodsFileName = tbrp.fileNamePatternWithoutCompSuffix.convertMultipleArguments(dateInElapsedPeriod, currentPeriodsCounter);
            currentPeriodsCounter = 0;
            setDateInCurrentPeriod(time);
            computeNextCheck();
            return true;
        }

        // next check for roll-over based on size
        if (invocationGate.isTooSoon(time)) {
            return false;
        }

        if (activeFile.length() >= maxFileSize.getSize()) {
            elapsedPeriodsFileName = tbrp.fileNamePatternWithoutCompSuffix.convertMultipleArguments(dateInCurrentPeriod, currentPeriodsCounter);
            currentPeriodsCounter++;
            return true;
        }
        return false;
    }

主要步骤:

(1) 获取当前时间点，并与nextCheck进行比较(日志出发时间点)；

(2) 如果当前时间点大于nextCheck，则计算得到新的日志文件名前缀,赋值至elapsedPeriodsFileName；清空currentPeriodsCounter(记录时间段内的日志文件总数)；计算下一个触发时间点后退出函数；

(3) 若当前时间点小于nextCheck，则进行文件大小的校验(通过isTooSoon判断函数触发是否过于频繁，如果时，则退出等待以后校验)；

(4) 比较当前日志文件和设置的最大文件大小比较，如果当前文件大小达到阈值，则计算新的日志文件名前缀，currentPeriodsCounter进行+1操作.

2.2.2 日志翻转

    public void rollover() throws RolloverFailure {

        // when rollover is called the elapsed period's file has
        // been already closed. This is a working assumption of this method.

        String elapsedPeriodsFileName = timeBasedFileNamingAndTriggeringPolicy.getElapsedPeriodsFileName();
        String elapsedPeriodStem = FileFilterUtil.afterLastSlash(elapsedPeriodsFileName);

        if (compressionMode == CompressionMode.NONE) {
            if (getParentsRawFileProperty() != null) {
                renameUtil.rename(getParentsRawFileProperty(), elapsedPeriodsFileName);
            } // else { nothing to do if CompressionMode == NONE and parentsRawFileProperty == null }
        } else {
            if (getParentsRawFileProperty() == null) {
                compressionFuture = compressor.asyncCompress(elapsedPeriodsFileName, elapsedPeriodsFileName, elapsedPeriodStem);
            } else {
                compressionFuture = renameRawAndAsyncCompress(elapsedPeriodsFileName, elapsedPeriodStem);
            }
        }

        if (archiveRemover != null) {
            Date now = new Date(timeBasedFileNamingAndTriggeringPolicy.getCurrentTime());
            this.cleanUpFuture = archiveRemover.cleanAsynchronously(now);
        }
    }

主要步骤：

(1) 获取isTriggeringEvent函数设置的elapsedPeriodsFileName文件名称;

(2) 如果不需要对日志文件进行压缩操作，则尝试将当前日志文件的名称重命名为elapsedPeriodsFileName；

(3) 如果需要对日志文件进行压缩，则尝试将日志文件进行异步压缩操作(需要注意，涉及到日志文件重命名操作);

(4) 设置archiveRemover，将当前时间点传入archiveRemover，通过其删除过期文件，或删除早期文件以保证文件夹大小在合理范围.

划重点:

(1) getParentsRawFileProperty: 配置文件中可以设置活动日志文件名称(简称rawFileName)，当日志文件达到触发条件时，将日志文件内容转移至翻转文件中，重新创建日志文件并命名为rawFileName；

(2) renameUtil.rename：日志文件转移时，会判断当前日志文件和翻转文件是否在同一块volume上，如果是则重命名文件即可，如果不是则复制当前文件内容至翻转文件中;

(3) 以上Future任务均是提交至Logback的线程池中执行，以保证日志记录的稳定性，避免成为应用的性能负担.