diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/AggregatingMessageChannelMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/AggregatingMessageChannelMetrics.java new file mode 100644 index 0000000000..ad9cac5a42 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/AggregatingMessageChannelMetrics.java @@ -0,0 +1,111 @@ +/* + * Copyright 2009-2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package org.springframework.integration.channel.management; + +import org.springframework.integration.support.management.ExponentialMovingAverage; +import org.springframework.integration.support.management.ExponentialMovingAverageRate; +import org.springframework.integration.support.management.ExponentialMovingAverageRatio; +import org.springframework.integration.support.management.MetricsContext; + +/** + * An implementation of {@link MessageChannelMetrics} that aggregates the total response + * time over a sample, to avoid fetching the system time twice for every message. + * + * @author Gary Russell + * @since 4.2 + */ +public class AggregatingMessageChannelMetrics extends DefaultMessageChannelMetrics { + + private static final int DEFAULT_SAMPLE_SIZE = 1000; + + private final int sampleSize; + + private long start; + + public AggregatingMessageChannelMetrics() { + this(null, DEFAULT_SAMPLE_SIZE); + } + + /** + * Construct an instance with default metrics with {@code window=10, period=1 second, + * lapsePeriod=1 minute}. + * @param name the name. + * @param sampleSize the sample size over which to aggregate the duration. + */ + public AggregatingMessageChannelMetrics(String name, int sampleSize) { + super(name); + this.sampleSize = sampleSize; + } + + /** + * Construct an instance with the supplied metrics. For proper representation of metrics, the + * supplied sendDuration must have a {@code factor=1000000.} and the the other arguments + * must be created with the {@code millis} constructor argument set to true. + * @param name the name. + * @param sendDuration an {@link ExponentialMovingAverage} for calculating the send duration. + * @param sendErrorRate an {@link ExponentialMovingAverageRate} for calculating the send error rate. + * @param sendSuccessRatio an {@link ExponentialMovingAverageRatio} for calculating the success ratio. + * @param sendRate an {@link ExponentialMovingAverageRate} for calculating the send rate. + * @param sampleSize the sample size over which to aggregate the duration. + */ + public AggregatingMessageChannelMetrics(String name, ExponentialMovingAverage sendDuration, + ExponentialMovingAverageRate sendErrorRate, ExponentialMovingAverageRatio sendSuccessRatio, + ExponentialMovingAverageRate sendRate, int sampleSize) { + super(name, sendDuration, sendErrorRate, sendSuccessRatio, sendRate); + this.sampleSize = sampleSize; + } + + @Override + public synchronized MetricsContext beforeSend() { + long count = this.sendCount.getAndIncrement(); + if (isFullStatsEnabled() && count % this.sampleSize == 0) { + this.start = System.nanoTime(); + this.sendRate.increment(start); + } + return new AggregatingChannelMetricsContext(this.start, count + 1); + } + + @Override + public void afterSend(MetricsContext context, boolean result) { + AggregatingChannelMetricsContext aggregatingContext = (AggregatingChannelMetricsContext) context; + long newCount = aggregatingContext.newCount; + if (result) { + if (isFullStatsEnabled() && newCount % this.sampleSize == 0) { + long now = System.nanoTime(); + this.sendSuccessRatio.success(now); + this.sendDuration.append(now - aggregatingContext.start); + } + } + else { + if (isFullStatsEnabled() && newCount % this.sampleSize == 0) { + long now = System.nanoTime(); + this.sendSuccessRatio.failure(now); + this.sendErrorRate.increment(now); + } + this.sendErrorCount.incrementAndGet(); + } + } + + protected static class AggregatingChannelMetricsContext extends DefaultChannelMetricsContext { + + protected long newCount; + + public AggregatingChannelMetricsContext(long start, long newCount) { + super(start); + this.newCount = newCount; + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java index 1124dd367c..4d20542080 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java @@ -15,9 +15,6 @@ package org.springframework.integration.channel.management; import java.util.concurrent.atomic.AtomicLong; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - import org.springframework.integration.support.management.ExponentialMovingAverage; import org.springframework.integration.support.management.ExponentialMovingAverageRate; import org.springframework.integration.support.management.ExponentialMovingAverageRatio; @@ -34,25 +31,23 @@ import org.springframework.integration.support.management.Statistics; */ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics { - protected final Log logger = LogFactory.getLog(getClass()); - public static final long ONE_SECOND_SECONDS = 1; public static final long ONE_MINUTE_SECONDS = 60; public static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10; - private final ExponentialMovingAverage sendDuration; + protected final ExponentialMovingAverage sendDuration; - private final ExponentialMovingAverageRate sendErrorRate; + protected final ExponentialMovingAverageRate sendErrorRate; - private final ExponentialMovingAverageRatio sendSuccessRatio; + protected final ExponentialMovingAverageRatio sendSuccessRatio; - private final ExponentialMovingAverageRate sendRate; + protected final ExponentialMovingAverageRate sendRate; - private final AtomicLong sendCount = new AtomicLong(); + protected final AtomicLong sendCount = new AtomicLong(); - private final AtomicLong sendErrorCount = new AtomicLong(); + protected final AtomicLong sendErrorCount = new AtomicLong(); protected final AtomicLong receiveCount = new AtomicLong(); @@ -106,10 +101,6 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics @Override public MetricsContext beforeSend() { - if (logger.isTraceEnabled()) { - logger.trace("Recording send on channel(" + this.name + ")"); - } - long start = 0; if (isFullStatsEnabled()) { start = System.nanoTime(); @@ -134,10 +125,6 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics } this.sendErrorCount.incrementAndGet(); } - if (logger.isTraceEnabled()) { - logger.trace("Elapsed: " + this.name - + ": " + ((System.nanoTime() / 1000000. - ((DefaultChannelMetricsContext) context).start)) + "ms"); - } } @Override @@ -229,9 +216,6 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics @Override public void afterReceive() { - if (logger.isTraceEnabled()) { - logger.trace("Recording receive on channel(" + this.name + ") "); - } this.receiveCount.incrementAndGet(); } @@ -267,11 +251,11 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics + "]", name, sendCount.get()); } - private static class DefaultChannelMetricsContext implements MetricsContext { + protected static class DefaultChannelMetricsContext implements MetricsContext { - private final long start; + protected final long start; - public DefaultChannelMetricsContext(long start) { + protected DefaultChannelMetricsContext(long start) { this.start = start; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index ef2efffabd..13c1c30592 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -110,7 +110,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im message = MessageHistory.write(message, this, this.getMessageBuilderFactory()); } if (countsEnabled) { - start = handlerMetrics.beforeHandle(message); + start = handlerMetrics.beforeHandle(); } this.handleMessageInternal(message); if (countsEnabled) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AbstractMessageHandlerMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AbstractMessageHandlerMetrics.java index ef92188342..9e287efcbe 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AbstractMessageHandlerMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AbstractMessageHandlerMetrics.java @@ -21,7 +21,6 @@ import org.apache.commons.logging.LogFactory; import org.springframework.integration.support.management.ConfigurableMetrics; import org.springframework.integration.support.management.MetricsContext; import org.springframework.integration.support.management.Statistics; -import org.springframework.messaging.Message; /** * Abstract base class for handler metrics implementations. @@ -57,14 +56,13 @@ public abstract class AbstractMessageHandlerMetrics implements ConfigurableMetri /** * Begin a handle event. - * @param message the message to handle. * @return the context to be used in the {@link #afterHandle(MetricsContext, boolean)}. */ - public abstract MetricsContext beforeHandle(Message message); + public abstract MetricsContext beforeHandle(); /** * End a handle event - * @param context the context from the previous {@link #beforeHandle(Message)}. + * @param context the context from the previous {@link #beforeHandle()}. * @param success true for success, false otherwise. */ public abstract void afterHandle(MetricsContext context, boolean success); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AggregatingMessageHandlerMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AggregatingMessageHandlerMetrics.java new file mode 100644 index 0000000000..c6cb917383 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AggregatingMessageHandlerMetrics.java @@ -0,0 +1,99 @@ +/* + * Copyright 2002-2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.handler.management; + +import org.springframework.integration.support.management.ExponentialMovingAverage; +import org.springframework.integration.support.management.MetricsContext; + +/** + * An implementation of {@link MessageHandlerMetrics} that aggregates the total response + * time over a sample, to avoid fetching the system time twice for every message. + * + * @author Gary Russell + * @since 2.0 + */ +public class AggregatingMessageHandlerMetrics extends DefaultMessageHandlerMetrics { + + private static final int DEFAULT_SAMPLE_SIZE = 1000; + + private final int sampleSize; + + private long start; + + public AggregatingMessageHandlerMetrics() { + this(null, DEFAULT_SAMPLE_SIZE); + } + + /** + * Construct an instance with the default moving average window (10). + * @param name the name. + * @param sampleSize the sample size over which to aggregate the duration. + */ + public AggregatingMessageHandlerMetrics(String name, int sampleSize) { + super(name); + this.sampleSize = sampleSize; + } + + /** + * Construct an instance with the supplied {@link ExponentialMovingAverage} calculating + * the duration of processing by the message handler (and any downstream synchronous + * endpoints). + * @param name the name. + * @param duration an {@link ExponentialMovingAverage} for calculating the duration. + * @param sampleSize the sample size over which to aggregate the duration. + */ + public AggregatingMessageHandlerMetrics(String name, ExponentialMovingAverage duration, int sampleSize) { + super(name, duration); + this.sampleSize = sampleSize; + } + + @Override + public synchronized MetricsContext beforeHandle() { + long count = this.handleCount.getAndIncrement(); + if (isFullStatsEnabled() && count % this.sampleSize == 0) { + this.start = System.nanoTime(); + } + this.activeCount.incrementAndGet(); + return new AggregatingHandlerMetricsContext(this.start, count + 1); + } + + @Override + public void afterHandle(MetricsContext context, boolean success) { + this.activeCount.decrementAndGet(); + AggregatingHandlerMetricsContext aggregatingContext = (AggregatingHandlerMetricsContext) context; + if (success) { + if (isFullStatsEnabled() && aggregatingContext.newCount % this.sampleSize == 0) { + this.duration.append(System.nanoTime() - aggregatingContext.start); + } + } + else { + this.errorCount.incrementAndGet(); + } + } + + protected static class AggregatingHandlerMetricsContext extends DefaultHandlerMetricsContext { + + protected long newCount; + + public AggregatingHandlerMetricsContext(long start, long newCount) { + super(start); + this.newCount = newCount; + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java index a574d490c9..6e0e0cb1d9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java @@ -21,7 +21,6 @@ import java.util.concurrent.atomic.AtomicLong; import org.springframework.integration.support.management.ExponentialMovingAverage; import org.springframework.integration.support.management.MetricsContext; import org.springframework.integration.support.management.Statistics; -import org.springframework.messaging.Message; /** * Default implementation; use the full constructor to customize the moving averages. @@ -35,13 +34,13 @@ public class DefaultMessageHandlerMetrics extends AbstractMessageHandlerMetrics private static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10; - private final AtomicLong activeCount = new AtomicLong(); + protected final AtomicLong activeCount = new AtomicLong(); - private final AtomicLong handleCount = new AtomicLong(); + protected final AtomicLong handleCount = new AtomicLong(); - private final AtomicLong errorCount = new AtomicLong(); + protected final AtomicLong errorCount = new AtomicLong(); - private final ExponentialMovingAverage duration; + protected final ExponentialMovingAverage duration; public DefaultMessageHandlerMetrics() { this(null); @@ -69,10 +68,7 @@ public class DefaultMessageHandlerMetrics extends AbstractMessageHandlerMetrics } @Override - public MetricsContext beforeHandle(Message message) { - if (logger.isTraceEnabled()) { - logger.trace("messageHandler(" + this.name + ") message(" + message + ") :"); - } + public MetricsContext beforeHandle() { long start = 0; if (isFullStatsEnabled()) { start = System.nanoTime(); @@ -158,11 +154,11 @@ public class DefaultMessageHandlerMetrics extends AbstractMessageHandlerMetrics return this.duration.getStatistics(); } - private static class DefaultHandlerMetricsContext implements MetricsContext { + protected static class DefaultHandlerMetricsContext implements MetricsContext { - private final long start; + protected final long start; - public DefaultHandlerMetricsContext(long start) { + protected DefaultHandlerMetricsContext(long start) { this.start = start; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java index da0d4879de..87b0bda91c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java @@ -13,8 +13,9 @@ package org.springframework.integration.support.management; +import java.util.ArrayDeque; import java.util.ArrayList; -import java.util.LinkedList; +import java.util.Deque; import java.util.List; @@ -41,7 +42,7 @@ public class ExponentialMovingAverage { private volatile double max; - private final List samples = new LinkedList(); + private final Deque samples = new ArrayDeque(); private final int retention; @@ -86,7 +87,7 @@ public class ExponentialMovingAverage { */ public synchronized void append(double value) { if (this.samples.size() == this.retention) { - samples.remove(0); + this.samples.poll(); } samples.add(value); count++;//NOSONAR - false positive, we're synchronized diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java index 5a2f250d00..cdedb4517c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java @@ -13,8 +13,9 @@ package org.springframework.integration.support.management; +import java.util.ArrayDeque; import java.util.ArrayList; -import java.util.LinkedList; +import java.util.Deque; import java.util.List; @@ -52,7 +53,7 @@ public class ExponentialMovingAverageRate { private final double period; - private final List times = new LinkedList(); + private final Deque times = new ArrayDeque(); private final int retention; @@ -109,7 +110,7 @@ public class ExponentialMovingAverageRate { */ public synchronized void increment(long t) { if (this.times.size() == this.retention) { - this.times.remove(0); + this.times.poll(); } this.times.add(t); this.count++;//NOSONAR - false positive, we're synchronized @@ -202,11 +203,9 @@ public class ExponentialMovingAverageRate { return count / (count / calc().getMean() + value); } - private double lastTime() { + private synchronized double lastTime() { if (this.times.size() > 0) { - synchronized (this) { - return this.times.get(this.times.size() - 1) / this.factor; - } + return this.times.peekFirst() / this.factor; } else { return this.t0; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java index a4a2fcfe6a..d82e759eeb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java @@ -13,9 +13,10 @@ package org.springframework.integration.support.management; +import java.util.ArrayDeque; import java.util.ArrayList; +import java.util.Deque; import java.util.Iterator; -import java.util.LinkedList; import java.util.List; @@ -50,9 +51,9 @@ public class ExponentialMovingAverageRatio { private final double lapse; - private final List times = new LinkedList(); + private final Deque times = new ArrayDeque(); - private final List values = new LinkedList(); + private final Deque values = new ArrayDeque(); private final int retention; @@ -124,8 +125,8 @@ public class ExponentialMovingAverageRatio { private synchronized void append(int value, long t) { if (this.times.size() == this.retention) { - this.times.remove(0); - this.values.remove(0); + this.times.poll(); + this.values.poll(); } this.times.add(t); this.values.add(value); @@ -221,11 +222,9 @@ public class ExponentialMovingAverageRatio { return alpha * mean + 1 - alpha; } - private double lastTime() { + private synchronized double lastTime() { if (this.times.size() > 0) { - synchronized (this) { - return this.times.get(this.times.size() - 1); - } + return this.times.peekFirst(); } else { return this.t0 * this.factor; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java index 02a648da5a..db6da494b9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java @@ -25,12 +25,21 @@ import org.junit.Test; /** * @author Dave Syer * @author Artem Bilan + * @author Gary Russell * */ public class ExponentialMovingAverageTests { private final ExponentialMovingAverage history = new ExponentialMovingAverage(10); + + @Test @Ignore // used to compare LinkedList to ArrayDeque which was 35% faster + public void perf() { + for (int i = 0; i < 100000000; i++) { + history.append(0.0); + } + } + @Test public void testGetCount() { assertEquals(0, history.getCount()); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/AggregatingMetricsFactory.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/AggregatingMetricsFactory.java new file mode 100644 index 0000000000..f30a571535 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/AggregatingMetricsFactory.java @@ -0,0 +1,53 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.monitor; + +import org.springframework.integration.channel.management.AbstractMessageChannelMetrics; +import org.springframework.integration.channel.management.AggregatingMessageChannelMetrics; +import org.springframework.integration.handler.management.AbstractMessageHandlerMetrics; +import org.springframework.integration.handler.management.AggregatingMessageHandlerMetrics; + + + +/** + * Implementation that returns aggregating metrics. + * + * @author Gary Russell + * @since 4.2 + * + */ +public class AggregatingMetricsFactory implements MetricsFactory { + + private final int sampleSize; + + /** + * @param sampleSize the number of messages over which to aggregate the elapsed time. + */ + public AggregatingMetricsFactory(int sampleSize) { + this.sampleSize = sampleSize; + } + + @Override + public AbstractMessageChannelMetrics createChannelMetrics(String name) { + return new AggregatingMessageChannelMetrics(name, this.sampleSize); + } + + @Override + public AbstractMessageHandlerMetrics createHandlerMetrics(String name) { + return new AggregatingMessageHandlerMetrics(name, this.sampleSize); + } + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests-context.xml new file mode 100644 index 0000000000..d513159e94 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests-context.xml @@ -0,0 +1,32 @@ + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests.java new file mode 100644 index 0000000000..4fac604c46 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests.java @@ -0,0 +1,107 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.monitor; + +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; + +import org.junit.Ignore; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.channel.management.AggregatingMessageChannelMetrics; +import org.springframework.integration.handler.BridgeHandler; +import org.springframework.integration.handler.ServiceActivatingHandler; +import org.springframework.integration.handler.management.AggregatingMessageHandlerMetrics; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 4.2 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class AggregatingMetricsTests { + + @Autowired + private MessageChannel input; + + @Autowired + private AbstractMessageChannel delay; + + @Autowired + private QueueChannel output; + + @Autowired + private BridgeHandler handler; + + @Autowired + private ServiceActivatingHandler delayer; + + @Test + public void testCounts() { + Message message = new GenericMessage("foo"); + int count = 2000; + for (int i = 0; i < count; i++) { + input.send(message); + } + assertEquals(count, this.output.getQueueSize()); + assertEquals(count, this.output.getSendCount()); + assertEquals(Long.valueOf(count / 1000).longValue(), this.output.getSendDuration().getCountLong()); + assertEquals(count, this.handler.getHandleCount()); + assertEquals(Long.valueOf(count / 1000).longValue(), this.handler.getDuration().getCountLong()); + } + + @Test + public void testElapsed() { + int sampleSize = 2; + this.delay.configureMetrics(new AggregatingMessageChannelMetrics("foo", sampleSize)); + this.delay.enableStats(true); + this.delayer.configureMetrics(new AggregatingMessageHandlerMetrics("bar", sampleSize)); + this.delayer.enableStats(true); + GenericMessage message = new GenericMessage("foo"); + int count = 4; + for (int i = 0; i < count; i++) { + this.delay.send(message); + } + assertEquals(count, this.delay.getSendCount()); + assertEquals(count / sampleSize, this.delay.getSendDuration().getCount()); + assertThat((int) this.delay.getMeanSendDuration() / sampleSize, greaterThanOrEqualTo(50)); + assertEquals(count, this.delayer.getHandleCount()); + assertEquals(count / sampleSize, this.delayer.getDuration().getCount()); + assertThat((int) this.delayer.getMeanDuration() / sampleSize, greaterThanOrEqualTo(50)); + } + + @Test @Ignore + public void perf() { + AggregatingMessageHandlerMetrics metrics = new AggregatingMessageHandlerMetrics(); + for (int i = 0; i < 100000000; i++) { + metrics.afterHandle(metrics.beforeHandle(), true); + } + } + +} diff --git a/src/reference/asciidoc/jmx.adoc b/src/reference/asciidoc/jmx.adoc index 4fd4c7e3e4..c08e73f215 100644 --- a/src/reference/asciidoc/jmx.adoc +++ b/src/reference/asciidoc/jmx.adoc @@ -509,6 +509,29 @@ By default, a `DefaultMetricsFactory` provides default implementation of `Messag To override the default `MetricsFactory` use the MBean exporter's `metrics-factory` attribute to provide a reference to your `MetricsFactory` bean instance. You can either customize the default implementations as described in the next bullet, or provide completely different implementations by overriding `AbstractMessageChannelMetrics` and/or `AbstractMessageHandlerMetrics`. +In addition to the default metrics factory described above, the framework provides the `AggregatingMetricsFactory`. +This factory creates `AggregatingMessageChannelMetrics` and `AggregatingMessageHandlerMetrics`. +In very high volume scenarios, the cost of capturing statistics can be prohibitive (2 calls to the system time and +storing the data). +The aggregating metrics aggregate the response time over a sample of messages. +This can save significant CPU time. + +CAUTION: The statistics will be skewed if messages arrive in bursts. +These metrics are intended for use with high, constant-volume, message rates. + +[source, xml] +---- + + + + + + + +---- + +The above configuration aggregates the response time over 1000 messages. +Counts (send, error) are maintained per-message but the statistics are per 1000 messages. * *Customizing the Default Channel/Handler Statistics*