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

Refactored ConcurrentHandler to accept a preconfigured ExecutorService rather than creating one from the ConcurrencyPolicy.

Browse Source
This commit is contained in:
Mark Fisher
2008-02-18 17:46:23 +00:00
parent 3d1303bb53
commit 0b5013939e
4 changed files with 71 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java
View File

@@ -17,7 +17,13 @@
package org.springframework.integration.endpoint;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
@@ -156,14 +162,19 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
public void afterPropertiesSet() {
if (this.concurrencyPolicy != null || this.handler instanceof ConcurrentHandler) {
if (!(this.handler instanceof ConcurrentHandler)) {
this.handler = new ConcurrentHandler(this.handler, this.concurrencyPolicy);
int capacity = concurrencyPolicy.getQueueCapacity();
BlockingQueue<Runnable> queue = (capacity < 1) ? new SynchronousQueue<Runnable>() :
new ArrayBlockingQueue<Runnable>(capacity);
ExecutorService executor = new ThreadPoolExecutor(concurrencyPolicy.getCoreSize(),
concurrencyPolicy.getMaxSize(), concurrencyPolicy.getKeepAliveSeconds(),
TimeUnit.SECONDS, queue);
this.handler = new ConcurrentHandler(this.handler, executor);
}
ConcurrentHandler concurrentHandler = (ConcurrentHandler) this.handler;
if (this.errorHandler != null) {
concurrentHandler.setErrorHandler(this.errorHandler);
}
concurrentHandler.setReplyHandler(this.replyHandler);
concurrentHandler.afterPropertiesSet();
}
this.initialized = true;
}

116
spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java
View File

@@ -16,15 +16,14 @@
package org.springframework.integration.handler;
import java.util.concurrent.ExecutorService;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.integration.message.Message;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;
@@ -35,50 +34,27 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class ConcurrentHandler implements MessageHandler, Lifecycle, InitializingBean {
public class ConcurrentHandler implements MessageHandler, DisposableBean {
private Log logger = LogFactory.getLog(this.getClass());
private final Log logger = LogFactory.getLog(this.getClass());
private MessageHandler handler;
private final MessageHandler handler;
private ThreadPoolTaskExecutor executor;
private final ExecutorService executor;
private volatile int currentQueueCapacity;
private volatile ErrorHandler errorHandler;
private final ConcurrencyPolicy concurrencyPolicy;
private ErrorHandler errorHandler;
private ReplyHandler replyHandler;
private volatile boolean running;
private Object lifecycleMonitor = new Object();
private volatile ReplyHandler replyHandler;
public ConcurrentHandler(MessageHandler handler) {
this(handler, null);
}
public ConcurrentHandler(MessageHandler handler, ConcurrencyPolicy concurrencyPolicy) {
public ConcurrentHandler(MessageHandler handler, ExecutorService executor) {
Assert.notNull(handler, "'handler' must not be null");
if (concurrencyPolicy != null) {
Assert.isTrue(concurrencyPolicy.getMaxSize() >= concurrencyPolicy.getCoreSize(),
"'coreSize' must not exceed 'maxSize'");
this.concurrencyPolicy = concurrencyPolicy;
}
else {
this.concurrencyPolicy = new ConcurrencyPolicy();
}
this.handler = handler;
}
public void setExecutor(ThreadPoolTaskExecutor executor) {
Assert.notNull(executor, "'executor' must not be null");
this.handler = handler;
this.executor = executor;
}
public void setErrorHandler(ErrorHandler errorHandler) {
this.errorHandler = errorHandler;
}
@@ -87,74 +63,12 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin
this.replyHandler = replyHandler;
}
public void afterPropertiesSet() {
refreshExecutor();
}
public void refreshExecutor() {
if (this.executor == null || currentQueueCapacity != this.concurrencyPolicy.getQueueCapacity()) {
this.initializeExecutor();
}
else {
this.refreshRuntimeModifiableExecutorProperties();
}
}
private void initializeExecutor() {
synchronized (this.lifecycleMonitor) {
if (this.executor != null) {
this.executor.shutdown();
}
this.executor = new ThreadPoolTaskExecutor();
this.currentQueueCapacity = this.concurrencyPolicy.getQueueCapacity();
this.executor.setQueueCapacity(this.currentQueueCapacity);
CustomizableThreadFactory threadFactory = new CustomizableThreadFactory();
threadFactory.setThreadNamePrefix("handler-");
this.executor.setThreadFactory(threadFactory);
this.refreshRuntimeModifiableExecutorProperties();
}
this.executor.afterPropertiesSet();
}
private void refreshRuntimeModifiableExecutorProperties() {
int coreSize = this.concurrencyPolicy.getCoreSize();
int maxSize = this.concurrencyPolicy.getMaxSize();
int keepAlive = this.concurrencyPolicy.getKeepAliveSeconds();
if (this.executor.getCorePoolSize() != coreSize) {
this.executor.setCorePoolSize(coreSize);
}
if (this.executor.getMaxPoolSize() != maxSize) {
this.executor.setMaxPoolSize(maxSize);
}
if (this.executor.getKeepAliveSeconds() != keepAlive) {
this.executor.setKeepAliveSeconds(keepAlive);
}
}
public boolean isRunning() {
return this.running;
}
public void start() {
synchronized (this.lifecycleMonitor) {
if (!this.running) {
this.afterPropertiesSet();
}
this.running = true;
}
}
public void stop() {
synchronized (this.lifecycleMonitor) {
if (this.isRunning()) {
this.executor.shutdown();
}
this.running = false;
}
public void destroy() {
this.executor.shutdown();
}
public Message<?> handle(Message<?> message) {
if (!this.isRunning()) {
if (this.executor.isShutdown()) {
throw new MessageHandlerNotRunningException();
}
try {

60
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java
View File

@@ -21,6 +21,9 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -58,8 +61,8 @@ public class DefaultMessageDispatcherTests {
SimpleChannel channel = new SimpleChannel();
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1)));
dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1)));
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()));
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()));
SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler();
scheduler.start();
dispatcher.setMessagingTaskScheduler(scheduler);
@@ -79,8 +82,8 @@ public class DefaultMessageDispatcherTests {
SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true));
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1)));
dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1)));
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()));
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()));
dispatcher.start();
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
@@ -99,13 +102,11 @@ public class DefaultMessageDispatcherTests {
SimpleChannel channel = new SimpleChannel();
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1)) {
@Override
public void start() {
}
});
dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1)));
dispatcher.addHandler(new ConcurrentHandler(handler3, new ConcurrencyPolicy(1, 1)));
ConcurrentHandler inactiveHandler = new ConcurrentHandler(handler1, createExecutor());
inactiveHandler.destroy();
dispatcher.addHandler(inactiveHandler);
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()));
dispatcher.addHandler(new ConcurrentHandler(handler3, createExecutor()));
dispatcher.start();
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
@@ -125,13 +126,11 @@ public class DefaultMessageDispatcherTests {
SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true));
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1)));
dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1)) {
@Override
public void start() {
}
});
dispatcher.addHandler(new ConcurrentHandler(handler3, new ConcurrencyPolicy(1, 1)));
ConcurrentHandler inactiveHandler = new ConcurrentHandler(handler2, createExecutor());
inactiveHandler.destroy();
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()));
dispatcher.addHandler(inactiveHandler);
dispatcher.addHandler(new ConcurrentHandler(handler3, createExecutor()));
dispatcher.start();
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
@@ -161,14 +160,14 @@ public class DefaultMessageDispatcherTests {
channel.getDispatcherPolicy().setRetryInterval(3);
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1)));
dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1)) {
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()));
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()) {
@Override
public Message<?> handle(Message<?> message) {
throw new MessageHandlerRejectedExecutionException();
}
});
dispatcher.addHandler(new ConcurrentHandler(handler3, new ConcurrencyPolicy(1, 1)));
dispatcher.addHandler(new ConcurrentHandler(handler3, createExecutor()));
SimpleChannel errorChannel = new SimpleChannel();
SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler();
scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel));
@@ -249,7 +248,7 @@ public class DefaultMessageDispatcherTests {
channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false);
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1)) {
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()) {
@Override
public Message<?> handle(Message<?> message) {
rejectedCounter1.incrementAndGet();
@@ -257,7 +256,7 @@ public class DefaultMessageDispatcherTests {
throw new MessageHandlerRejectedExecutionException();
}
});
dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1)) {
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()) {
@Override
public Message<?> handle(Message<?> message) {
rejectedCounter2.incrementAndGet();
@@ -293,7 +292,7 @@ public class DefaultMessageDispatcherTests {
SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1)) {
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()) {
@Override
public Message<?> handle(Message<?> message) {
rejectedCounter1.incrementAndGet();
@@ -301,7 +300,7 @@ public class DefaultMessageDispatcherTests {
throw new MessageHandlerRejectedExecutionException();
}
});
dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1)) {
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()) {
@Override
public Message<?> handle(Message<?> message) {
if (rejectedCounter2.get() == 1) {
@@ -312,7 +311,7 @@ public class DefaultMessageDispatcherTests {
throw new MessageHandlerRejectedExecutionException();
}
});
dispatcher.addHandler(new ConcurrentHandler(handler3, new ConcurrencyPolicy(1, 1)) {
dispatcher.addHandler(new ConcurrentHandler(handler3, createExecutor()) {
@Override
public Message<?> handle(Message<?> message) {
rejectedCounter3.incrementAndGet();
@@ -347,7 +346,7 @@ public class DefaultMessageDispatcherTests {
SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
dispatcher.addHandler(new ConcurrentHandler(handler1, new ConcurrencyPolicy(1, 1)) {
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()) {
@Override
public Message<?> handle(Message<?> message) {
if (rejectedCounter1.get() == 2) {
@@ -358,7 +357,7 @@ public class DefaultMessageDispatcherTests {
throw new MessageHandlerRejectedExecutionException();
}
});
dispatcher.addHandler(new ConcurrentHandler(handler2, new ConcurrencyPolicy(1, 1)) {
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()) {
@Override
public Message<?> handle(Message<?> message) {
if (rejectedCounter2.get() == 4) {
@@ -474,4 +473,9 @@ public class DefaultMessageDispatcherTests {
assertEquals("endpoint2 should have accepted the message", 1, counter2.get());
}
private static ExecutorService createExecutor() {
return new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
}
}

14
spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java
View File

@@ -22,6 +22,9 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -168,7 +171,7 @@ public class DefaultMessageEndpointTests {
};
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
endpoint.setChannelRegistry(channelRegistry);
endpoint.setHandler(new ConcurrentHandler(handler));
endpoint.setHandler(new ConcurrentHandler(handler, createExecutor()));
endpoint.setDefaultOutputChannelName("replyChannel");
endpoint.start();
endpoint.handle(new StringMessage(1, "test"));
@@ -219,7 +222,7 @@ public class DefaultMessageEndpointTests {
};
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
endpoint.setChannelRegistry(channelRegistry);
endpoint.setHandler(new ConcurrentHandler(handler));
endpoint.setHandler(new ConcurrentHandler(handler, createExecutor()));
endpoint.setDefaultOutputChannelName("replyChannel");
endpoint.start();
endpoint.handle(new StringMessage(1, "test"));
@@ -244,7 +247,7 @@ public class DefaultMessageEndpointTests {
};
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
endpoint.setChannelRegistry(channelRegistry);
endpoint.setHandler(new ConcurrentHandler(handler));
endpoint.setHandler(new ConcurrentHandler(handler, createExecutor()));
endpoint.start();
StringMessage message = new StringMessage(1, "test");
message.getHeader().setReplyChannelName("replyChannel");
@@ -449,4 +452,9 @@ public class DefaultMessageEndpointTests {
endpoint.stop();
}
private static ExecutorService createExecutor() {
return new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
}
}