From 0b5013939eae5aa7b8f94b8caa0b61671092649b Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 18 Feb 2008 17:46:23 +0000 Subject: [PATCH] Refactored ConcurrentHandler to accept a preconfigured ExecutorService rather than creating one from the ConcurrencyPolicy. --- .../endpoint/DefaultMessageEndpoint.java | 15 ++- .../handler/ConcurrentHandler.java | 116 +++--------------- .../DefaultMessageDispatcherTests.java | 60 ++++----- .../endpoint/DefaultMessageEndpointTests.java | 14 ++- 4 files changed, 71 insertions(+), 134 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java index a008671ceb..4fc35d6db5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java @@ -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 queue = (capacity < 1) ? new SynchronousQueue() : + new ArrayBlockingQueue(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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java index d6c0a04b43..b716d3c4b1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java @@ -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 { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java index 2712d97e42..1b7bf5cdae 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java @@ -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()); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java index 3387eeaefe..adec715a28 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java @@ -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()); + } + }