AsyncAppenderBase.java
/**
* Logback: the reliable, generic, fast and flexible logging framework.
* Copyright (C) 1999-2015, QOS.ch. All rights reserved.
*
* This program and the accompanying materials are dual-licensed under
* either the terms of the Eclipse Public License v1.0 as published by
* the Eclipse Foundation
*
* or (per the licensee's choosing)
*
* under the terms of the GNU Lesser General Public License version 2.1
* as published by the Free Software Foundation.
*/
package ch.qos.logback.core;
import ch.qos.logback.core.spi.AppenderAttachable;
import ch.qos.logback.core.spi.AppenderAttachableImpl;
import ch.qos.logback.core.util.InterruptUtil;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/**
* This appender and derived classes, log events asynchronously. In order to
* avoid loss of logging events, this appender should be closed. It is the
* user's responsibility to close appenders, typically at the end of the
* application lifecycle.
* <p>
* This appender buffers events in a {@link BlockingQueue}. {@link Worker}
* thread created by this appender takes events from the head of the queue, and
* dispatches them to the single appender attached to this appender.
* <p>
* Please refer to the
* <a href="http://logback.qos.ch/manual/appenders.html#AsyncAppender">logback
* manual</a> for further information about this appender.
* </p>
*
* @param <E>
* @author Ceki Gülcü
* @author Torsten Juergeleit
* @since 1.0.4
*/
public class AsyncAppenderBase<E> extends UnsynchronizedAppenderBase<E> implements AppenderAttachable<E> {
AppenderAttachableImpl<E> aai = new AppenderAttachableImpl<E>();
BlockingQueue<E> 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.
* <p>
* 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 subclasses 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<E>(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);
}
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();
}
}
}
public int getQueueSize() {
return queueSize;
}
public void setQueueSize(int queueSize) {
this.queueSize = queueSize;
}
public int getDiscardingThreshold() {
return discardingThreshold;
}
public void setDiscardingThreshold(int discardingThreshold) {
this.discardingThreshold = discardingThreshold;
}
public int getMaxFlushTime() {
return maxFlushTime;
}
public void setMaxFlushTime(int maxFlushTime) {
this.maxFlushTime = maxFlushTime;
}
/**
* Returns the number of elements currently in the blocking queue.
*
* @return number of elements currently in the queue.
*/
public int getNumberOfElementsInQueue() {
return blockingQueue.size();
}
public void setNeverBlock(boolean neverBlock) {
this.neverBlock = neverBlock;
}
public boolean isNeverBlock() {
return neverBlock;
}
/**
* The remaining capacity available in the blocking queue.
* <p>
* See also {@link java.util.concurrent.BlockingQueue#remainingCapacity()
* BlockingQueue#remainingCapacity()}
*
* @return the remaining capacity
*
*/
public int getRemainingCapacity() {
return blockingQueue.remainingCapacity();
}
public void addAppender(Appender<E> newAppender) {
if (appenderCount == 0) {
appenderCount++;
addInfo("Attaching appender named [" + newAppender.getName() + "] to AsyncAppender.");
aai.addAppender(newAppender);
} else {
addWarn("One and only one appender may be attached to AsyncAppender.");
addWarn("Ignoring additional appender named [" + newAppender.getName() + "]");
}
}
public Iterator<Appender<E>> iteratorForAppenders() {
return aai.iteratorForAppenders();
}
public Appender<E> getAppender(String name) {
return aai.getAppender(name);
}
public boolean isAttached(Appender<E> eAppender) {
return aai.isAttached(eAppender);
}
public void detachAndStopAllAppenders() {
aai.detachAndStopAllAppenders();
}
public boolean detachAppender(Appender<E> eAppender) {
return aai.detachAppender(eAppender);
}
public boolean detachAppender(String name) {
return aai.detachAppender(name);
}
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 {
List<E> elements = new ArrayList<E>();
E e0 = parent.blockingQueue.take();
elements.add(e0);
parent.blockingQueue.drainTo(elements);
for (E e : elements) {
aai.appendLoopOnAppenders(e);
}
} catch (InterruptedException e1) {
// exit if interrupted
break;
}
}
addInfo("Worker thread will flush remaining events before exiting. ");
for (E e : parent.blockingQueue) {
aai.appendLoopOnAppenders(e);
parent.blockingQueue.remove(e);
}
aai.detachAndStopAllAppenders();
}
}
}