日志 note_深入源码解析日志框架Log4j2(二)

异步

AsyncAppender

log4j2突出于其他日志的优势，异步日志实现。我们先从日志打印看进去。找到Logger，随便找一个log日志的方法。

public void debug(final Marker marker, final Message msg) {logIfEnabled(FQCN, Level.DEBUG, marker, msg, msg != null ? msg.getThrowable() : null);}

一路跟进

@PerformanceSensitive// NOTE: This is a hot method. Current implementation compiles to 29 bytes of byte code.// This is within the 35 byte MaxInlineSize threshold. Modify with care!private void logMessageTrackRecursion(final String fqcn,final Level level,final Marker marker,final Message msg,final Throwable throwable) {try {incrementRecursionDepth(); // LOG4J2-1518, LOG4J2-2031tryLogMessage(fqcn, level, marker, msg, throwable);} finally {decrementRecursionDepth();}}

可以看出这个在打日志之前做了调用次数的记录。跟进tryLogMessage，

@PerformanceSensitive// NOTE: This is a hot method. Current implementation compiles to 26 bytes of byte code.// This is within the 35 byte MaxInlineSize threshold. Modify with care!private void tryLogMessage(final String fqcn,final Level level,final Marker marker,final Message msg,final Throwable throwable) {try {logMessage(fqcn, level, marker, msg, throwable);} catch (final Exception e) {// LOG4J2-1990 Log4j2 suppresses all exceptions that occur once application called the loggerhandleLogMessageException(e, fqcn, msg);}}

继续跟进：

@Overridepublic void logMessage(final String fqcn, final Level level, final Marker marker, final Message message,final Throwable t) {final Message msg = message == null ? new SimpleMessage(Strings.EMPTY) : message;final ReliabilityStrategy strategy = privateConfig.loggerConfig.getReliabilityStrategy();strategy.log(this, getName(), fqcn, marker, level, msg, t);}

这里可以看到在实际打日志的时候，会从config中获取打日志的策略，跟踪ReliabilityStrategy的创建，发现默认的实现类为DefaultReliabilityStrategy，跟进看实际打日志的方法

@Overridepublic void log(final Supplier<LoggerConfig> reconfigured, final String loggerName, final String fqcn, final Marker marker, final Level level,final Message data, final Throwable t) {loggerConfig.log(loggerName, fqcn, marker, level, data, t);}

这里实际打日志的方法居然是交给一个config去实现的。。。感觉有点奇怪。。跟进看看

@PerformanceSensitive("allocation")public void log(final String loggerName, final String fqcn, final Marker marker, final Level level,final Message data, final Throwable t) {List<Property> props = null;if (!propertiesRequireLookup) {props = properties;} else {if (properties != null) {props = new ArrayList<>(properties.size());final LogEvent event = Log4jLogEvent.newBuilder().setMessage(data).setMarker(marker).setLevel(level).setLoggerName(loggerName).setLoggerFqcn(fqcn).setThrown(t).build();for (int i = 0; i < properties.size(); i++) {final Property prop = properties.get(i);final String value = prop.isValueNeedsLookup() // since LOG4J2-1575? config.getStrSubstitutor().replace(event, prop.getValue()) //: prop.getValue();props.add(Property.createProperty(prop.getName(), value));}}}final LogEvent logEvent = logEventFactory.createEvent(loggerName, marker, fqcn, level, data, props, t);try {log(logEvent, LoggerConfigPredicate.ALL);} finally {// LOG4J2-1583 prevent scrambled logs when logging calls are nested (logging in toString())ReusableLogEventFactory.release(logEvent);}}

可以清楚的看到try之前是在创建LogEvent，try里面做的才是真正的log（好tm累），一路跟进。

private void processLogEvent(final LogEvent event, LoggerConfigPredicate predicate) {event.setIncludeLocation(isIncludeLocation());if (predicate.allow(this)) {callAppenders(event);}logParent(event, predicate);}

接下来就是callAppender了，我们直接开始看AsyncAppender的append方法：

/*** Actual writing occurs here.** @param logEvent The LogEvent.*/@Overridepublic void append(final LogEvent logEvent) {if (!isStarted()) {throw new IllegalStateException("AsyncAppender " + getName() + " is not active");}final Log4jLogEvent memento = Log4jLogEvent.createMemento(logEvent, includeLocation);InternalAsyncUtil.makeMessageImmutable(logEvent.getMessage());if (!transfer(memento)) {if (blocking) {if (AbstractLogger.getRecursionDepth() > 1) { // LOG4J2-1518, LOG4J2-2031// If queue is full AND we are in a recursive call, call appender directly to prevent deadlockAsyncQueueFullMessageUtil.logWarningToStatusLogger();logMessageInCurrentThread(logEvent);} else {// delegate to the event router (which may discard, enqueue and block, or log in current thread)final EventRoute route = asyncQueueFullPolicy.getRoute(thread.getId(), memento.getLevel());route.logMessage(this, memento);}} else {error("Appender " + getName() + " is unable to write primary appenders. queue is full");logToErrorAppenderIfNecessary(false, memento);}}}

这里主要的步骤就是：

生成logEvent
将logEvent放入BlockingQueue，就是transfer方法
如果BlockingQueue满了则启用相应的策略

同样的，这里也有一个线程用来做异步消费的事情

private class AsyncThread extends Log4jThread {private volatile boolean shutdown = false;private final List<AppenderControl> appenders;private final BlockingQueue<LogEvent> queue;public AsyncThread(final List<AppenderControl> appenders, final BlockingQueue<LogEvent> queue) {super("AsyncAppender-" + THREAD_SEQUENCE.getAndIncrement());this.appenders = appenders;this.queue = queue;setDaemon(true);}@Overridepublic void run() {while (!shutdown) {LogEvent event;try {event = queue.take();if (event == SHUTDOWN_LOG_EVENT) {shutdown = true;continue;}} catch (final InterruptedException ex) {break; // LOG4J2-830}event.setEndOfBatch(queue.isEmpty());final boolean success = callAppenders(event);if (!success && errorAppender != null) {try {errorAppender.callAppender(event);} catch (final Exception ex) {// Silently accept the error.}}}// Process any remaining items in the queue.LOGGER.trace("AsyncAppender.AsyncThread shutting down. Processing remaining {} queue events.",queue.size());int count = 0;int ignored = 0;while (!queue.isEmpty()) {try {final LogEvent event = queue.take();if (event instanceof Log4jLogEvent) {final Log4jLogEvent logEvent = (Log4jLogEvent) event;logEvent.setEndOfBatch(queue.isEmpty());callAppenders(logEvent);count++;} else {ignored++;LOGGER.trace("Ignoring event of class {}", event.getClass().getName());}} catch (final InterruptedException ex) {// May have been interrupted to shut down.// Here we ignore interrupts and try to process all remaining events.}}LOGGER.trace("AsyncAppender.AsyncThread stopped. Queue has {} events remaining. "+ "Processed {} and ignored {} events since shutdown started.", queue.size(), count, ignored);}/*** Calls {@link AppenderControl#callAppender(LogEvent) callAppender} on all registered {@code AppenderControl}* objects, and returns {@code true} if at least one appender call was successful, {@code false} otherwise. Any* exceptions are silently ignored.** @param event the event to forward to the registered appenders* @return {@code true} if at least one appender call succeeded, {@code false} otherwise*/boolean callAppenders(final LogEvent event) {boolean success = false;for (final AppenderControl control : appenders) {try {control.callAppender(event);success = true;} catch (final Exception ex) {// If no appender is successful the error appender will get it.}}return success;}public void shutdown() {shutdown = true;if (queue.isEmpty()) {queue.offer(SHUTDOWN_LOG_EVENT);}if (getState() == State.TIMED_WAITING || getState() == State.WAITING) {this.interrupt(); // LOG4J2-1422: if underlying appender is stuck in wait/sleep/join/park call}}}

直接看run方法：

阻塞获取logEvent
将logEvent分发出去
如果线程要退出了，将blockingQueue里面的event消费完在退出。

AsyncLogger

接从AsyncLogger的logMessage看进去：

public void logMessage(final String fqcn, final Level level, final Marker marker, final Message message,final Throwable thrown) {if (loggerDisruptor.isUseThreadLocals()) {logWithThreadLocalTranslator(fqcn, level, marker, message, thrown);} else {// LOG4J2-1172: avoid storing non-JDK classes in ThreadLocals to avoid memory leaks in web appslogWithVarargTranslator(fqcn, level, marker, message, thrown);}}

跟进logWithThreadLocalTranslator，

private void logWithThreadLocalTranslator(final String fqcn, final Level level, final Marker marker,final Message message, final Throwable thrown) {// Implementation note: this method is tuned for performance. MODIFY WITH CARE!final RingBufferLogEventTranslator translator = getCachedTranslator();initTranslator(translator, fqcn, level, marker, message, thrown);initTranslatorThreadValues(translator);publish(translator);}

这里的逻辑很简单，就是将日志相关的信息转换成RingBufferLogEvent（RingBuffer是Disruptor的无所队列），然后将其发布到RingBuffer中。发布到RingBuffer中，那肯定也有消费逻辑。这时候有两种方式可以找到这个消费的逻辑。

找disruptor被使用的地方，然后查看，但是这样做会很容易迷惑
按照Log4j2的尿性，这种Logger都有对应的start方法，我们可以从start方法入手寻找

在start方法中，我们找到了一段代码：

final RingBufferLogEventHandler[] handlers = {new RingBufferLogEventHandler()};disruptor.handleEventsWith(handlers);

直接看看这个RingBufferLogEventHandler的实现：

public class RingBufferLogEventHandler implementsSequenceReportingEventHandler<RingBufferLogEvent>, LifecycleAware {private static final int NOTIFY_PROGRESS_THRESHOLD = 50;private Sequence sequenceCallback;private int counter;private long threadId = -1;@Overridepublic void setSequenceCallback(final Sequence sequenceCallback) {this.sequenceCallback = sequenceCallback;}@Overridepublic void onEvent(final RingBufferLogEvent event, final long sequence,final boolean endOfBatch) throws Exception {event.execute(endOfBatch);event.clear();// notify the BatchEventProcessor that the sequence has progressed.// Without this callback the sequence would not be progressed// until the batch has completely finished.if (++counter > NOTIFY_PROGRESS_THRESHOLD) {sequenceCallback.set(sequence);counter = 0;}}/*** Returns the thread ID of the background consumer thread, or {@code -1} if the background thread has not started* yet.* @return the thread ID of the background consumer thread, or {@code -1}*/public long getThreadId() {return threadId;}@Overridepublic void onStart() {threadId = Thread.currentThread().getId();}@Overridepublic void onShutdown() {}
}

顺着接口找上去，发现一个接口：

/*** Callback interface to be implemented for processing events as they become available in the {@link RingBuffer}** @param <T> event implementation storing the data for sharing during exchange or parallel coordination of an event.* @see BatchEventProcessor#setExceptionHandler(ExceptionHandler) if you want to handle exceptions propagated out of the handler.*/
public interface EventHandler<T>
{/*** Called when a publisher has published an event to the {@link RingBuffer}** @param event      published to the {@link RingBuffer}* @param sequence   of the event being processed* @param endOfBatch flag to indicate if this is the last event in a batch from the {@link RingBuffer}* @throws Exception if the EventHandler would like the exception handled further up the chain.*/void onEvent(T event, long sequence, boolean endOfBatch) throws Exception;
}

通过注释可以发现，这个onEvent就是处理逻辑，回到RingBufferLogEventHandler的onEvent方法，发现里面有一个execute方法，跟进：

public void execute(final boolean endOfBatch) {this.endOfBatch = endOfBatch;asyncLogger.actualAsyncLog(this);}

这个方法就是实际打日志了，AsyncLogger看起来还是比较简单的，只是使用了一个Disruptor。

插件化

之前在很多代码里面都可以看到

final PluginManager manager = new PluginManager(CATEGORY);
manager.collectPlugins(pluginPackages);

其实整个log4j2为了获得更好的扩展性，将自己的很多组件都做成了插件，然后在配置的时候去加载plugin。跟进collectPlugins。

public void collectPlugins(final List<String> packages) {final String categoryLowerCase = category.toLowerCase();final Map<String, PluginType<?>> newPlugins = new LinkedHashMap<>();// First, iterate the Log4j2Plugin.dat files found in the main CLASSPATHMap<String, List<PluginType<?>>> builtInPlugins = PluginRegistry.getInstance().loadFromMainClassLoader();if (builtInPlugins.isEmpty()) {// If we didn't find any plugins above, someone must have messed with the log4j-core.jar.// Search the standard package in the hopes we can find our core plugins.builtInPlugins = PluginRegistry.getInstance().loadFromPackage(LOG4J_PACKAGES);}mergeByName(newPlugins, builtInPlugins.get(categoryLowerCase));// Next, iterate any Log4j2Plugin.dat files from OSGi Bundlesfor (final Map<String, List<PluginType<?>>> pluginsByCategory : PluginRegistry.getInstance().getPluginsByCategoryByBundleId().values()) {mergeByName(newPlugins, pluginsByCategory.get(categoryLowerCase));}// Next iterate any packages passed to the static addPackage method.for (final String pkg : PACKAGES) {mergeByName(newPlugins, PluginRegistry.getInstance().loadFromPackage(pkg).get(categoryLowerCase));}// Finally iterate any packages provided in the configuration (note these can be changed at runtime).if (packages != null) {for (final String pkg : packages) {mergeByName(newPlugins, PluginRegistry.getInstance().loadFromPackage(pkg).get(categoryLowerCase));}}LOGGER.debug("PluginManager '{}' found {} plugins", category, newPlugins.size());plugins = newPlugins;}

处理逻辑如下：

从Log4j2Plugin.dat中加载所有的内置的plugin
然后将OSGi Bundles中的Log4j2Plugin.dat中的plugin加载进来
再加载传入的package路径中的plugin
最后加载配置中的plugin

逻辑还是比较简单的，但是我在看源码的时候发现了一个很有意思的东西，就是在加载log4j2 core插件的时候，也就是

PluginRegistry.getInstance().loadFromMainClassLoader()

这个方法，跟进到decodeCacheFiles：

private Map<String, List<PluginType<?>>> decodeCacheFiles(final ClassLoader loader) {final long startTime = System.nanoTime();final PluginCache cache = new PluginCache();try {final Enumeration<URL> resources = loader.getResources(PluginProcessor.PLUGIN_CACHE_FILE);if (resources == null) {LOGGER.info("Plugin preloads not available from class loader {}", loader);} else {cache.loadCacheFiles(resources);}} catch (final IOException ioe) {LOGGER.warn("Unable to preload plugins", ioe);}final Map<String, List<PluginType<?>>> newPluginsByCategory = new HashMap<>();int pluginCount = 0;for (final Map.Entry<String, Map<String, PluginEntry>> outer : cache.getAllCategories().entrySet()) {final String categoryLowerCase = outer.getKey();final List<PluginType<?>> types = new ArrayList<>(outer.getValue().size());newPluginsByCategory.put(categoryLowerCase, types);for (final Map.Entry<String, PluginEntry> inner : outer.getValue().entrySet()) {final PluginEntry entry = inner.getValue();final String className = entry.getClassName();try {final Class<?> clazz = loader.loadClass(className);final PluginType<?> type = new PluginType<>(entry, clazz, entry.getName());types.add(type);++pluginCount;} catch (final ClassNotFoundException e) {LOGGER.info("Plugin [{}] could not be loaded due to missing classes.", className, e);} catch (final LinkageError e) {LOGGER.info("Plugin [{}] could not be loaded due to linkage error.", className, e);}}}final long endTime = System.nanoTime();final DecimalFormat numFormat = new DecimalFormat("#0.000000");final double seconds = (endTime - startTime) * 1e-9;LOGGER.debug("Took {} seconds to load {} plugins from {}",numFormat.format(seconds), pluginCount, loader);return newPluginsByCategory;}

可以发现加载时候是从一个文件（PLUGIN_CACHE_FILE）获取所有要获取的plugin。看到这里的时候我有一个疑惑就是，为什么不用反射的方式直接去扫描，而是要从文件中加载进来，而且文件是写死的，很不容易扩展啊。然后我找了一下PLUGIN_CACHE_FILE这个静态变量的用处，发现了PluginProcessor这个类，这里用到了注解处理器。

/*** Annotation processor for pre-scanning Log4j 2 plugins.*/
@SupportedAnnotationTypes("org.apache.logging.log4j.core.config.plugins.*")
public class PluginProcessor extends AbstractProcessor {// TODO: this could be made more abstract to allow for compile-time and run-time plugin processing/*** The location of the plugin cache data file. This file is written to by this processor, and read from by* {@link org.apache.logging.log4j.core.config.plugins.util.PluginManager}.*/public static final String PLUGIN_CACHE_FILE ="META-INF/org/apache/logging/log4j/core/config/plugins/Log4j2Plugins.dat";private final PluginCache pluginCache = new PluginCache();@Overridepublic boolean process(final Set<? extends TypeElement> annotations, final RoundEnvironment roundEnv) {System.out.println("Processing annotations");try {final Set<? extends Element> elements = roundEnv.getElementsAnnotatedWith(Plugin.class);if (elements.isEmpty()) {System.out.println("No elements to process");return false;}collectPlugins(elements);writeCacheFile(elements.toArray(new Element[elements.size()]));System.out.println("Annotations processed");return true;} catch (final IOException e) {e.printStackTrace();error(e.getMessage());return false;} catch (final Exception ex) {ex.printStackTrace();error(ex.getMessage());return false;}}
}

（不太重要的方法省略）我们可以看到在process方法中，PluginProcessor会先收集所有的Plugin，然后在写入文件。这样做的好处就是可以省去反射时候的开销。然后我又看了一下Plugin这个注解，发现它的RetentionPolicy是RUNTIME，一般来说PluginProcessor是搭配RetentionPolicy.SOURCE，CLASS使用的，而且既然你把自己的Plugin扫描之后写在文件中了，RetentionPolicy就没有必要是RUNTIME了吧，这个是一个很奇怪的地方。