Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQP-474:Log Appenders appId Replacement Once Only

Browse Source 
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
This commit is contained in:
Gary Russell
2015-02-08 10:11:56 -05:00
committed by Artem Bilan
parent 71051fdbbd
commit 23bd1dff38
3 changed files with 29 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j/AmqpAppender.java
View File

@@ -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) {

19
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java
View File

@@ -216,7 +216,7 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
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<ILoggingEvent> {
@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<ILoggingEvent> {
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<ILoggingEvent> {
*/
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<ILoggingEvent> {
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);

53
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j/AmqpAppenderIntegrationTests.java
View File

@@ -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<String, String>());
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<String, String>());
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);