聊聊logback的AsyncAppender

序

本文主要研究一下logback的AsyncAppender

AsyncAppender

ch/qos/logback/classic/AsyncAppender.java

public class AsyncAppender extends AsyncAppenderBase {

    boolean includeCallerData = false;

    /**
     * Events of level TRACE, DEBUG and INFO are deemed to be discardable.
     * @param event
     * @return true if the event is of level TRACE, DEBUG or INFO false otherwise.
     */
    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();
    }

    public boolean isIncludeCallerData() {
        return includeCallerData;
    }

    public void setIncludeCallerData(boolean includeCallerData) {
        this.includeCallerData = includeCallerData;
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

AsyncAppender继承了AsyncAppenderBase，它新增了includeCallerData配置，另外覆盖了isDiscardable、preprocess方法，isDiscardable针对TRACE、DEBUG的级别返回true，INFO返回false；preprocess则判断是否includeCallerData，是的话则执行eventObject.getCallerData()

AsyncAppenderBase

ch/qos/logback/core/AsyncAppenderBase.java

public class AsyncAppenderBase extends UnsynchronizedAppenderBase implements AppenderAttachable {

    AppenderAttachableImpl aai = new AppenderAttachableImpl();
    BlockingQueue blockingQueue;

    /**
     * The default buffer size.
     */
    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();

    /**
     * The default maximum queue flush time allowed during appender stop. If the 
     * worker takes longer than this time it will exit, discarding any remaining 
     * items in the queue
     */
    public static final int DEFAULT_MAX_FLUSH_TIME = 1000;
    int maxFlushTime = DEFAULT_MAX_FLUSH_TIME;

    /**
     * Is the eventObject passed as parameter discardable? The base class's implementation of this method always returns
     * 'false' but sub-classes may (and do) override this method.
     * 

     * 
Note that only if the buffer is nearly full are events discarded. Otherwise, when the buffer is "not full"
     * all events are logged.
     *
     * @param eventObject
     * @return - true if the event can be discarded, false otherwise
     */
    protected boolean isDiscardable(E eventObject) {
        return false;
    }

    /**
     * Pre-process the event prior to queueing. The base class does no pre-processing but sub-classes can
     * override this behavior.
     *
     * @param eventObject
     */
    protected void preprocess(E eventObject) {
    }

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

        if (discardingThreshold == UNDEFINED)
            discardingThreshold = queueSize / 5;
        addInfo("Setting discardingThreshold to " + discardingThreshold);
        worker.setDaemon(true);
        worker.setName("AsyncAppender-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();

        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();
        }
    }


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

    protected boolean isDiscardable(E eventObject) {
        return false;
    }

    protected void preprocess(E eventObject) {
    }

    private boolean isQueueBelowDiscardingThreshold() {
        return (blockingQueue.remainingCapacity() < discardingThreshold);
    }

    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();
            }
        }
    }

    //......
}    
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160

AsyncAppenderBase继承了UnsynchronizedAppenderBase，实现了AppenderAttachable接口，它定义了queueSize、discardingThreshold、neverBlock等属性，其start方法会根据queueSize创建ArrayBlockingQueue，discardingThreshold默认为queueSize / 5，之后启动Wroker；stop方法则执行worker.interrupt()，然后等待maxFlushTime让log进行flush；其append方法会先判断isQueueBelowDiscardingThreshold及isDiscardable，都为true则直接返回，否则执行preprocess、put方法

Worker

ch/qos/logback/core/AsyncAppenderBase.java

    class Worker extends Thread {

        public void run() {
            AsyncAppenderBase parent = AsyncAppenderBase.this;
            AppenderAttachableImpl 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);
                parent.blockingQueue.remove(e);
            }

            aai.detachAndStopAllAppenders();
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

Worker的run方法会不断循环从blockingQueue阻塞取出原生，然后添加到AppenderAttachableImpl；在started为false的时候跳槽循环，然后遍历blockingQueue，添加到AppenderAttachableImpl，然后将其从blockingQueue；最后执行detachAndStopAllAppenders

AppenderAttachableImpl

ch/qos/logback/core/spi/AppenderAttachableImpl.java

    public int appendLoopOnAppenders(E e) {
        int size = 0;
        final Appender[] appenderArray = appenderList.asTypedArray();
        final int len = appenderArray.length;
        for (int i = 0; i < len; i++) {
            appenderArray[i].doAppend(e);
            size++;
        }
        return size;
    }

    /**
     * Remove and processPriorToRemoval all previously attached appenders.
     */
    public void detachAndStopAllAppenders() {
        for (Appender a : appenderList) {
            a.stop();
        }
        appenderList.clear();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

AppenderAttachableImpl的appendLoopOnAppenders方法会遍历所有的appenderList执行doAppend方法；其detachAndStopAllAppenders则遍历appenderList，挨个执行stop，最后clear掉整个appenderList

小结

logback的AsyncAppender使用ArrayBlockingQueue(默认size为256)来进行缓冲，每次append的时候会先判断isQueueBelowDiscardingThreshold及isDiscardable，为true则直接返回/丢弃，之后执行preprocess，最后执行put，put的时候有个参数neverBlock，为true则使用的是offer方法，队列满的时候会被丢弃，为false则是阻塞的方法，等到put成功才返回；另外它有个worker线程，不断从blockingQueue阻塞take元素出来然后写入到appenderList，在关闭时还会遍历队列写入到appenderList然后从队列移除，最后清空队列。