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

INT-3752: Add AggregatingMetricsFactory

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3752
JIRA: https://jira.spring.io/browse/INT-3754

The statistics are based on this count so, for example the
`meanSendDuration` reflects the mean of total duration from the first send to
the last in each batch.

Also change the linked lists to `Deque`s.

Also remove `isTraceEnabled()` calls - too expensive in high volume environments.

Polishing

Assign the count to a local variable so the mod (%) operation is short
circuited rather than short circuiting the call to isFullStatsEnabled().

INT-3752: Fix Aggregating Metrics

Previously, the duration was for the first message in each sample.

We need to capture the total elapsed time for the sample and use it
for the duration calculation.

Add 'newCount' to the context so the after... method can calculate
the duration at the end of the sample.

The start field does not need to be volatile because its updates
are synchronized.

INT-3752: Fix Javadocs
This commit is contained in:
Gary Russell
2015-06-26 15:13:34 -04:00
committed by Artem Bilan
parent 5298368168
commit a480354210
14 changed files with 472 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
spring-integration-core/src/main/java/org/springframework/integration/channel/management/AggregatingMessageChannelMetrics.java Normal file
View File

@@ -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;
}
}
}

34
spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java
View File

@@ -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;
}

2
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java
View File

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

6
spring-integration-core/src/main/java/org/springframework/integration/handler/management/AbstractMessageHandlerMetrics.java
View File

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

99
spring-integration-core/src/main/java/org/springframework/integration/handler/management/AggregatingMessageHandlerMetrics.java Normal file
View File

@@ -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;
}
}
}

20
spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java
View File

@@ -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;
}

7
spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java
View File

@@ -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<Double> samples = new LinkedList<Double>();
private final Deque<Double> samples = new ArrayDeque<Double>();
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

13
spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java
View File

@@ -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<Long> times = new LinkedList<Long>();
private final Deque<Long> times = new ArrayDeque<Long>();
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;

17
spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java
View File

@@ -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<Long> times = new LinkedList<Long>();
private final Deque<Long> times = new ArrayDeque<Long>();
private final List<Integer> values = new LinkedList<Integer>();
private final Deque<Integer> values = new ArrayDeque<Integer>();
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;

9
spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java
View File

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

53
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/AggregatingMetricsFactory.java Normal file
View File

@@ -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);
}
}

32
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests-context.xml Normal file
View File

@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jmx="http://www.springframework.org/schema/integration/jmx"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jmx http://www.springframework.org/schema/integration/jmx/spring-integration-jmx.xsd">
<context:mbean-server />
<int-jmx:mbean-export metrics-factory="aggregatingMetricsFactory" />
<int:channel id="input" />
<int:bridge input-channel="input" output-channel="queue" />
<int:channel id="queue">
<int:queue />
</int:channel>
<bean id="aggregatingMetricsFactory" class="org.springframework.integration.monitor.AggregatingMetricsFactory">
<constructor-arg value="1000" />
</bean>
<int:channel id="delay" />
<int:service-activator input-channel="delay" expression="T(Thread).sleep(50)" />
</beans>

107
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests.java Normal file
View File

@@ -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<String>("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<String> message = new GenericMessage<String>("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);
}
}
}

23
src/reference/asciidoc/jmx.adoc
View File

@@ -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]
----
<context:mbean-server />
<int-jmx:mbean-export metrics-factory="aggregatingMetricsFactory" />
<bean id="aggregatingMetricsFactory" class="org.springframework.integration.monitor.AggregatingMetricsFactory">
<constructor-arg value="1000" /> <!-- sample size -->
</bean>
----
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*