日志 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);}}}這里主要的步驟就是:
同樣的,這里也有一個線程用來做異步消費的事情
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方法:
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;}處理邏輯如下:
邏輯還是比較簡單的,但是我在看源碼的時候發現了一個很有意思的東西,就是在加載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了吧,這個是一個很奇怪的地方。
小結
總算是把Log4j2的代碼看完了,發現它的設計理念很值得借鑒,為了靈活性,所有的東西都設計成插件式?;ヂ摼W技術日益發展,各種中間件層出不窮,而作為工程師的我們更需要做的是去思考代碼與代碼之間的關系,毫無疑問的是,解耦是最具有美感的關系。
添加Java高級架構交流群 378461078
關注微信公眾號“托尼的技術成長之路”
總結
以上是生活随笔為你收集整理的日志 note_深入源码解析日志框架Log4j2(二)的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 使用matlab播放特定频率的声音
- 下一篇: printf打印百分号%