From 23bd1dff38f2036141e7d6d6934e5780e9867eaf Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sun, 8 Feb 2015 10:11:56 -0500 Subject: [PATCH] AMQP-474:Log Appenders appId Replacement Once Only JIRA: https://jira.spring.io/browse/AMQP-474 Instead of replacing the placeholder on every log event, replace it once during initialization. Polishing; Restore ampqProps.appId AMQP-474: Polishing Conflicts: spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j/AmqpAppender.java spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java --- .../amqp/rabbit/log4j/AmqpAppender.java | 72 +++++++------------ .../amqp/rabbit/logback/AmqpAppender.java | 19 ++--- .../log4j/AmqpAppenderIntegrationTests.java | 53 -------------- 3 files changed, 29 insertions(+), 115 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j/AmqpAppender.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j/AmqpAppender.java index 635d4c9a..277774f0 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j/AmqpAppender.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j/AmqpAppender.java @@ -30,7 +30,6 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.log4j.AppenderSkeleton; import org.apache.log4j.Layout; import org.apache.log4j.Level; -import org.apache.log4j.MDC; import org.apache.log4j.PatternLayout; import org.apache.log4j.spi.ErrorCode; import org.apache.log4j.spi.LocationInfo; @@ -144,18 +143,13 @@ public class AmqpAppender extends AppenderSkeleton { /** * Log4J Layout to use to generate routing key. */ - private Layout routingKeyLayout = new PatternLayout(routingKeyPattern); + private Layout routingKeyLayout; /** * Used to synchronize access to pattern layouts. */ private final Object layoutMutex = new Object(); - /** - * Whether or not we've tried to declare this exchange yet. - */ - private final AtomicBoolean exchangeDeclared = new AtomicBoolean(false); - /** * Configuration arbitrary application ID. */ @@ -313,7 +307,6 @@ public class AmqpAppender extends AppenderSkeleton { public void setRoutingKeyPattern(String routingKeyPattern) { this.routingKeyPattern = routingKeyPattern; - this.routingKeyLayout = new PatternLayout(routingKeyPattern); } public boolean isDeclareExchange() { @@ -404,16 +397,18 @@ public class AmqpAppender extends AppenderSkeleton { this.charset = charset; } - /** - * Submit the required number of senders into the pool. - */ - protected void startSenders() { - senderPool = Executors.newCachedThreadPool(); - synchronized(this) { - } // (logically) flush all variables to main memory - for (int i = 0; i < senderPoolSize; i++) { - senderPool.submit(new EventSender()); - } + @Override + public void activateOptions() { + this.routingKeyLayout = new PatternLayout(this.routingKeyPattern + .replaceAll("%X\\{applicationId\\}", this.applicationId)); + this.connectionFactory = new CachingConnectionFactory(); + this.connectionFactory.setHost(host); + this.connectionFactory.setPort(port); + this.connectionFactory.setUsername(username); + this.connectionFactory.setPassword(password); + this.connectionFactory.setVirtualHost(virtualHost); + maybeDeclareExchange(); + startSenders(); } /** @@ -443,26 +438,19 @@ public class AmqpAppender extends AppenderSkeleton { } } + /** + * Submit the required number of senders into the pool. + */ + protected void startSenders() { + this.senderPool = Executors.newCachedThreadPool(); + for (int i = 0; i < senderPoolSize; i++) { + senderPool.submit(new EventSender()); + } + } + @Override public void append(LoggingEvent event) { - if (null == senderPool && this.initializing.compareAndSet(false, true)) { - try { - connectionFactory = new CachingConnectionFactory(); - connectionFactory.setHost(host); - connectionFactory.setPort(port); - connectionFactory.setUsername(username); - connectionFactory.setPassword(password); - connectionFactory.setVirtualHost(virtualHost); - maybeDeclareExchange(); - exchangeDeclared.set(true); - - startSenders(); - } - finally { - this.initializing.set(false); - } - } - events.add(new Event(event, event.getProperties())); + this.events.add(new Event(event, event.getProperties())); } @Override @@ -498,12 +486,6 @@ public class AmqpAppender extends AppenderSkeleton { */ protected class EventSender implements Runnable { - public EventSender() { - synchronized(AmqpAppender.this) { - // (logically) invalidate the CPU cache so we see all outer class fields correctly - } - } - @Override public void run() { try { @@ -530,7 +512,6 @@ public class AmqpAppender extends AppenderSkeleton { // Set applicationId, if we're using one if (null != applicationId) { amqpProps.setAppId(applicationId); - MDC.put(APPLICATION_ID, applicationId); } // Set timestamp @@ -598,11 +579,6 @@ public class AmqpAppender extends AppenderSkeleton { + " after " + maxSenderRetries + " retries", e, ErrorCode.WRITE_FAILURE, logEvent); } } - finally { - if (null != applicationId) { - MDC.remove(APPLICATION_ID); - } - } } } catch (Throwable t) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java index 2dd497c4..d267f780 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java @@ -216,7 +216,7 @@ public class AmqpAppender extends AppenderBase { private TargetLengthBasedClassNameAbbreviator abbreviator; public void setRoutingKeyPattern(String routingKeyPattern) { - this.routingKeyLayout.setPattern("%nopex" + routingKeyPattern); + this.routingKeyLayout.setPattern("%nopex{}" + routingKeyPattern); } public String getHost() { @@ -378,6 +378,8 @@ public class AmqpAppender extends AppenderBase { @Override public void start() { super.start(); + this.routingKeyLayout.setPattern(this.routingKeyLayout.getPattern() + .replaceAll("%property\\{applicationId\\}", this.applicationId)); this.routingKeyLayout.setContext(getContext()); this.routingKeyLayout.start(); this.locationLayout.setContext(getContext()); @@ -390,8 +392,6 @@ public class AmqpAppender extends AppenderBase { this.connectionFactory.setVirtualHost(this.virtualHost); maybeDeclareExchange(); this.senderPool = Executors.newCachedThreadPool(); - synchronized(this) { - } // (logically) flush all variables to main memory for (int i = 0; i < this.senderPoolSize; i++) { this.senderPool.submit(new EventSender()); } @@ -458,12 +458,6 @@ public class AmqpAppender extends AppenderBase { */ protected class EventSender implements Runnable { - public EventSender() { - synchronized(AmqpAppender.this) { - // (logically) invalidate the CPU cache so we see all outer class fields correctly - } - } - @Override public void run() { try { @@ -505,15 +499,12 @@ public class AmqpAppender extends AppenderBase { String.format("%s.%s()[%s]", location[0], location[1], location[2])); } String msgBody; - String routingKey; + String routingKey = routingKeyLayout.doLayout(logEvent); // Set applicationId, if we're using one - if (null != applicationId) { + if (applicationId != null) { amqpProps.setAppId(applicationId); - logEvent.getLoggerContextVO().getPropertyMap().put(APPLICATION_ID, applicationId); } - routingKey = routingKeyLayout.doLayout(logEvent); - if (abbreviator != null && logEvent instanceof LoggingEvent) { ((LoggingEvent) logEvent).setLoggerName(abbreviator.abbreviate(name)); msgBody = layout.doLayout(logEvent); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j/AmqpAppenderIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j/AmqpAppenderIntegrationTests.java index 06d9823e..05da118f 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j/AmqpAppenderIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j/AmqpAppenderIntegrationTests.java @@ -89,59 +89,6 @@ public class AmqpAppenderIntegrationTests { Log4jConfigurer.initLogging("classpath:log4j.properties"); } - @Test - public void testInit() { - final AtomicInteger count = new AtomicInteger(); - final LoggingEvent event = new LoggingEvent - ("foo", null, 1, Level.INFO, "bar", - "baz", null, null, null, - new HashMap()); - AmqpAppender appender = new AmqpAppender() { - - @Override - protected void maybeDeclareExchange() { - super.maybeDeclareExchange(); - if (count.incrementAndGet() < 2) { - // ensure we don't try to initialize again while initializing - append(event); - } - } - - }; - appender.append(event); - assertEquals(1, count.get()); - } - - @Test - public void testInitRetry() { - final AtomicInteger count = new AtomicInteger(); - final LoggingEvent event = new LoggingEvent - ("foo", null, 1, Level.INFO, "bar", - "baz", null, null, null, - new HashMap()); - AmqpAppender appender = new AmqpAppender() { - - @Override - protected void maybeDeclareExchange() { - super.maybeDeclareExchange(); - if (count.incrementAndGet() < 2) { - throw new RuntimeException("foo"); - } - } - - }; - try { - appender.append(event); - fail("Expected exception"); - } - catch (RuntimeException e) { - assertEquals("foo", e.getMessage()); - } - // ensure we initialize again if the first time failed - appender.append(event); - assertEquals(2, count.get()); - } - @Test public void testAppender() throws InterruptedException { TestListener testListener = (TestListener) applicationContext.getBean("testListener", 4);