无锡网站开发定制开发,qq电脑版登录入口,国内网站设计欣赏,做网站网站建设专业公司异步AsyncAppenderlog4j2突出于其他日志的优势#xff0c;异步日志实现。我们先从日志打印看进去。找到Logger#xff0c;随便找一个log日志的方法。public void debug(final Marker marker, final Message msg) {logIfEnabled(FQCN, Level.DEBUG, marker, msg, msg ! null…异步AsyncAppenderlog4j2突出于其他日志的优势异步日志实现。我们先从日志打印看进去。找到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();}}可以看出这个在打日志之前做了调用次数的记录。跟进tryLogMessagePerformanceSensitive// 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 SupplierLoggerConfig 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) {ListProperty 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之前是在创建LogEventtry里面做的才是真正的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 ListAppenderControl appenders;private final BlockingQueueLogEvent queue;public AsyncThread(final ListAppenderControl appenders, final BlockingQueueLogEvent 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);}}跟进logWithThreadLocalTranslatorprivate 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);}这里的逻辑很简单就是将日志相关的信息转换成RingBufferLogEventRingBuffer是Disruptor的无所队列然后将其发布到RingBuffer中。发布到RingBuffer中那肯定也有消费逻辑。这时候有两种方式可以找到这个消费的逻辑。找disruptor被使用的地方然后查看但是这样做会很容易迷惑按照Log4j2的尿性这种Logger都有对应的start方法我们可以从start方法入手寻找在start方法中我们找到了一段代码final RingBufferLogEventHandler[] handlers {new RingBufferLogEventHandler()};disruptor.handleEventsWith(handlers);直接看看这个RingBufferLogEventHandler的实现public class RingBufferLogEventHandler implementsSequenceReportingEventHandlerRingBufferLogEvent, 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 EventHandlerT
{/*** 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 ListString packages) {final String categoryLowerCase category.toLowerCase();final MapString, PluginType? newPlugins new LinkedHashMap();// First, iterate the Log4j2Plugin.dat files found in the main CLASSPATHMapString, ListPluginType? builtInPlugins PluginRegistry.getInstance().loadFromMainClassLoader();if (builtInPlugins.isEmpty()) {// If we didnt 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 MapString, ListPluginType? 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()这个方法跟进到decodeCacheFilesprivate MapString, ListPluginType? decodeCacheFiles(final ClassLoader loader) {final long startTime System.nanoTime();final PluginCache cache new PluginCache();try {final EnumerationURL 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 MapString, ListPluginType? newPluginsByCategory new HashMap();int pluginCount 0;for (final Map.EntryString, MapString, PluginEntry outer : cache.getAllCategories().entrySet()) {final String categoryLowerCase outer.getKey();final ListPluginType? types new ArrayList(outer.getValue().size());newPluginsByCategory.put(categoryLowerCase, types);for (final Map.EntryString, 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.SOURCECLASS使用的而且既然你把自己的Plugin扫描之后写在文件中了RetentionPolicy就没有必要是RUNTIME了吧这个是一个很奇怪的地方。小结总算是把Log4j2的代码看完了发现它的设计理念很值得借鉴为了灵活性所有的东西都设计成插件式。互联网技术日益发展各种中间件层出不穷而作为工程师的我们更需要做的是去思考代码与代码之间的关系毫无疑问的是解耦是最具有美感的关系。添加Java高级架构交流群 378461078关注微信公众号“托尼的技术成长之路”
