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

INT-711 Created LoadBalancingStrategy and improved encapsulation for the unicasting dispatcher (now contains the TaskExecutor LB strategy and a failover boolean flag).

Browse Source
This commit is contained in:
Mark Fisher
2009-07-11 05:49:34 +00:00
parent 2cd12df50e
commit aeb8c7e3a9
13 changed files with 239 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java
View File

@@ -16,9 +16,8 @@
package org.springframework.integration.channel;
import org.springframework.integration.dispatcher.AbstractUnicastDispatcher;
import org.springframework.integration.dispatcher.RoundRobinDispatcher;
import org.springframework.util.Assert;
import org.springframework.integration.dispatcher.LoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
/**
* A channel that invokes a single subscriber for each sent Message.
@@ -30,21 +29,36 @@ import org.springframework.util.Assert;
*/
public class DirectChannel extends AbstractSubscribableChannel {
private final AbstractUnicastDispatcher dispatcher;
private final UnicastingDispatcher dispatcher = new UnicastingDispatcher();
/**
* Create a channel with no {@link LoadBalancingStrategy}.
* The dispatcher for such a channel will invoke its
* MessageHandlers in a fixed-order.
*/
public DirectChannel() {
this.dispatcher = new RoundRobinDispatcher();
}
public DirectChannel(AbstractUnicastDispatcher dispatcher) {
Assert.notNull(dispatcher, "dispatcher must not be null");
this.dispatcher = dispatcher;
/**
* Create a DirectChannel with a {@link LoadBalancingStrategy}. The
* strategy <emphasis>must not</emphasis> be null.
*/
public DirectChannel(LoadBalancingStrategy loadBalancingStrategy) {
this.dispatcher.setLoadBalancingStrategy(loadBalancingStrategy);
}
/**
* Specify whether the channel's dispatcher should have failover enabled.
* By default, it will. Set this value to 'false' to disable it.
*/
public void setFailover(boolean failover) {
this.dispatcher.setFailover(failover);
}
@Override
protected AbstractUnicastDispatcher getDispatcher() {
protected UnicastingDispatcher getDispatcher() {
return this.dispatcher;
}

67
org.springframework.integration/src/main/java/org/springframework/integration/channel/ExecutorChannel.java
View File

@@ -16,21 +16,13 @@
package org.springframework.integration.channel;
import java.util.concurrent.Executors;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.dispatcher.AbstractUnicastDispatcher;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.integration.dispatcher.RoundRobinDispatcher;
import org.springframework.integration.message.MessageHandler;
import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
import org.springframework.util.Assert;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
/**
* An implementation of {@link MessageChannel} that delegates to an instance of
* {@link AbstractUnicastDispatcher} and wraps all send invocations within a
* {@link UnicastingDispatcher} and wraps all send invocations within a
* {@link TaskExecutor}.
*
* @author Mark Fisher
@@ -38,66 +30,17 @@ import org.springframework.util.Assert;
*/
public class ExecutorChannel extends AbstractSubscribableChannel {
private final ExecutorDecoratingDispatcher dispatcher;
private final UnicastingDispatcher dispatcher;
public ExecutorChannel() {
this(null, null);
}
public ExecutorChannel(TaskExecutor taskExecutor) {
this(null, taskExecutor);
}
public ExecutorChannel(AbstractUnicastDispatcher dispatcher) {
this(dispatcher, null);
}
public ExecutorChannel(AbstractUnicastDispatcher dispatcher, TaskExecutor taskExecutor) {
if (dispatcher == null) {
dispatcher = new RoundRobinDispatcher();
}
this.dispatcher = new ExecutorDecoratingDispatcher(dispatcher, taskExecutor);
this.dispatcher = new UnicastingDispatcher(taskExecutor);
}
@Override
protected MessageDispatcher getDispatcher() {
protected UnicastingDispatcher getDispatcher() {
return this.dispatcher;
}
private static class ExecutorDecoratingDispatcher implements MessageDispatcher {
private final AbstractUnicastDispatcher targetDispatcher;
private final TaskExecutor taskExecutor;
ExecutorDecoratingDispatcher(AbstractUnicastDispatcher dispatcher, TaskExecutor taskExecutor) {
Assert.notNull(dispatcher, "'dispatcher' must not be null");
this.targetDispatcher = dispatcher;
this.taskExecutor = taskExecutor != null ? taskExecutor
: new ConcurrentTaskExecutor(Executors.newSingleThreadExecutor());
}
public boolean addHandler(MessageHandler handler) {
return this.targetDispatcher.addHandler(handler);
}
public boolean removeHandler(MessageHandler handler) {
return this.targetDispatcher.removeHandler(handler);
}
public final boolean dispatch(final Message<?> message) {
this.taskExecutor.execute(new Runnable() {
public void run() {
targetDispatcher.dispatch(message);
}
});
return true;
}
}
}

29
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java
View File

@@ -18,9 +18,8 @@ package org.springframework.integration.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
@@ -37,6 +36,7 @@ public class PointToPointChannelParser extends AbstractChannelParser {
private static final String DISPATCHER_PACKAGE = IntegrationNamespaceUtils.BASE_PACKAGE + ".dispatcher";
@Override
protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = null;
@@ -81,23 +81,16 @@ public class PointToPointChannelParser extends AbstractChannelParser {
else {
builder = BeanDefinitionBuilder.genericBeanDefinition(CHANNEL_PACKAGE + ".DirectChannel");
}
parseDispatcher(element.getAttribute("dispatcher"), builder, parserContext);
}
return builder;
}
private void parseDispatcher(String dispatcherAttribute, BeanDefinitionBuilder builder, ParserContext parserContext) {
if (dispatcherAttribute != null) {
if (dispatcherAttribute.equals("failover")) {
BeanDefinitionBuilder dispatcherBuilder = BeanDefinitionBuilder
.genericBeanDefinition(DISPATCHER_PACKAGE + ".FailOverDispatcher");
dispatcherBuilder.setRole(BeanDefinition.ROLE_SUPPORT);
builder.addConstructorArgReference(BeanDefinitionReaderUtils.registerWithGeneratedName(dispatcherBuilder
.getBeanDefinition(), parserContext.getRegistry()));
// this attribute is deprecated, but if set, we need to create a UnicastingDispatcher
// without any LoadBalancerStrategy and the failover flag set to true (default).
String dispatcherAttribute = element.getAttribute("dispatcher");
if (!"failover".equals(dispatcherAttribute)) {
// round-robin dispatcher by default, but TODO first we need to check for the dispatcher element.
builder.addConstructorArgValue(new RootBeanDefinition(
DISPATCHER_PACKAGE + ".RoundRobinLoadBalancingStrategy", null, null));
}
}
// rely on the default for round-robin
}
return builder;
}
private boolean parseQueueCapacity(BeanDefinitionBuilder builder, Element queueElement) {

7
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliverException.java → org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AggregateMessageDeliveryException.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.dispatcher;
import java.util.Collections;
import java.util.List;
import org.springframework.integration.core.Message;
@@ -30,12 +31,12 @@ import org.springframework.integration.message.MessageDeliveryException;
* @since 1.0.3
*/
@SuppressWarnings("serial")
public class AggregateMessageDeliverException extends MessageDeliveryException {
public class AggregateMessageDeliveryException extends MessageDeliveryException {
private final List<? extends Exception> aggregatedExceptions;
public AggregateMessageDeliverException(Message<?> undeliveredMessage,
public AggregateMessageDeliveryException(Message<?> undeliveredMessage,
String description, List<? extends Exception> aggregatedExceptions) {
super(undeliveredMessage, description);
this.aggregatedExceptions = aggregatedExceptions;
@@ -43,7 +44,7 @@ public class AggregateMessageDeliverException extends MessageDeliveryException {
public List<? extends Exception> getAggregatedExceptions() {
return this.aggregatedExceptions;
return Collections.unmodifiableList(this.aggregatedExceptions);
}
}

24
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java → org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/LoadBalancingStrategy.java
View File

@@ -1,5 +1,4 @@
/*
* Copyright 2002-2009 the original author or authors.
/* Copyright 2002-2009 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.
@@ -16,29 +15,20 @@
package org.springframework.integration.dispatcher;
import java.util.Iterator;
import java.util.List;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageHandler;
/**
* {@link AbstractUnicastDispatcher} that will try its handlers in the
* same order every dispatch.
* Strategy for determining the iteration order of a MessageHandler list.
*
* @author Mark Fisher
* @author Iwein Fuld
* @since 1.0.3
*/
public class FailOverDispatcher extends AbstractUnicastDispatcher {
public interface LoadBalancingStrategy {
@Override
protected void handleExceptions(List<RuntimeException> allExceptions,
Message<?> message, boolean isLast) {
if (isLast) {
if (allExceptions != null && allExceptions.size() == 1) {
throw allExceptions.get(0);
}
throw new AggregateMessageDeliverException(message,
"All attempts to deliver Message to MessageHandlers failed.", allExceptions);
}
}
public Iterator<MessageHandler> getHandlerIterator(Message<?> message, List<MessageHandler> handlers);
}

28
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/RoundRobinDispatcher.java → org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/RoundRobinLoadBalancingStrategy.java
View File

@@ -25,32 +25,26 @@ import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageHandler;
/**
* Round-robin implementation of {@link AbstractUnicastDispatcher}. This
* Round-robin implementation of {@link LoadBalancingStrategy}. This
* implementation will keep track of the index of the handler that has been
* tried first and use a different starting handler every dispatch.
*
* @author Iwein Fuld
* @author Mark Fisher
* @since 1.0.3
*/
public class RoundRobinDispatcher extends AbstractUnicastDispatcher {
private volatile boolean failover = true;
public class RoundRobinLoadBalancingStrategy implements LoadBalancingStrategy {
private final AtomicInteger currentHandlerIndex = new AtomicInteger();
public void setFailover(boolean failover) {
this.failover = failover;
}
/**
* Returns an iterator that starts at a new point in the list every time the
* first part of the list that is skipped will be used at the end of the
* iteration, so it guarantees all handlers are returned once on subsequent
* <code>next()</code> invocations.
*/
@Override
protected Iterator<MessageHandler> getHandlerIterator() {
List<MessageHandler> handlers = this.getHandlers();
public final Iterator<MessageHandler> getHandlerIterator(final Message<?> message, final List<MessageHandler> handlers) {
int size = handlers.size();
if (size == 0) {
return handlers.iterator();
@@ -72,16 +66,4 @@ public class RoundRobinDispatcher extends AbstractUnicastDispatcher {
return indexTail < 0 ? indexTail + size : indexTail;
}
@Override
protected void handleExceptions(List<RuntimeException> allExceptions,
Message<?> message, boolean isLast) {
if (isLast || !this.failover) {
if (allExceptions != null && allExceptions.size() == 1) {
throw allExceptions.get(0);
}
throw new AggregateMessageDeliverException(message,
"Failed to deliver Message to any MessageHandler.", allExceptions);
}
}
}

148
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java Normal file
View File

@@ -0,0 +1,148 @@
/* Copyright 2002-2009 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.dispatcher;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
/**
* Implementation of {@link MessageDispatcher} that will attempt to send a
* {@link Message} to at most one of its handlers. The handlers will be tried
* as determined by the {@link LoadBalancingStrategy} if one is configured. As
* soon as <em>one</em> of the handlers accepts the Message, the dispatcher will
* return <code>true</code> and ignore the rest of its handlers.
* <p/>
* If the dispatcher has no handlers, a {@link MessageDeliveryException} will be
* thrown. If all handlers throw Exceptions, the dispatcher will throw an
* {@link AggregateMessageDeliveryException}.
* <p/>
* A load-balancing strategy may be provided to this class to control the order in
* which the handlers will be tried.
*
* @author Iwein Fuld
* @author Mark Fisher
* @since 1.0.2
*/
public class UnicastingDispatcher extends AbstractDispatcher {
private volatile boolean failover = true;
private volatile LoadBalancingStrategy loadBalancingStrategy;
private final TaskExecutor taskExecutor;
public UnicastingDispatcher() {
this.taskExecutor = null;
}
public UnicastingDispatcher(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
/**
* Specify whether this dispatcher should failover when a single
* {@link MessageHandler} throws an Exception. The default value is
* <code>true</code>.
*/
public void setFailover(boolean failover) {
this.failover = failover;
}
/**
* Provide a {@link LoadBalancingStrategy} for this dispatcher.
*/
public void setLoadBalancingStrategy(LoadBalancingStrategy loadBalancingStrategy) {
Assert.notNull(loadBalancingStrategy, "loadBalancingStrategy must not be null");
this.loadBalancingStrategy = loadBalancingStrategy;
}
public final boolean dispatch(final Message<?> message) {
if (this.taskExecutor != null) {
this.taskExecutor.execute(new Runnable() {
public void run() {
doDispatch(message);
}
});
return true;
}
return this.doDispatch(message);
}
private boolean doDispatch(Message<?> message) {
boolean success = false;
Iterator<MessageHandler> handlerIterator = this.getHandlerIterator(message);
if (!handlerIterator.hasNext()) {
throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
}
List<RuntimeException> exceptions = new ArrayList<RuntimeException>();
while (success == false && handlerIterator.hasNext()) {
MessageHandler handler = handlerIterator.next();
try {
handler.handleMessage(message);
success = true; // we have a winner.
}
catch (Exception e) {
RuntimeException runtimeException = (e instanceof RuntimeException)
? (RuntimeException) e
: new MessageDeliveryException(message,
"Dispatcher failed to deliver Message.", e);
exceptions.add(runtimeException);
this.handleExceptions(exceptions, message, !handlerIterator.hasNext());
}
}
return success;
}
/**
* Returns the iterator that will be used to loop over the handlers.
* Delegates to a {@link LoadBalancingStrategy} if available. Otherwise,
* it simply returns the Iterator for the existing handler List.
*/
private Iterator<MessageHandler> getHandlerIterator(Message<?> message) {
if (this.loadBalancingStrategy != null) {
return this.loadBalancingStrategy.getHandlerIterator(message, this.getHandlers());
}
return this.getHandlers().iterator();
}
/**
* Handles Exceptions that occur while dispatching. If this dispatcher has
* failover enabled, it will only throw an Exception when the handler list
* is exhausted. The 'isLast' flag will be <emphasis>true</emphasis> if the
* Exception occurred during the final iteration of the MessageHandlers.
* If failover is disabled for this dispatcher, it will re-throw any
* Exception immediately.
*/
private void handleExceptions(List<RuntimeException> allExceptions, Message<?> message, boolean isLast) {
if (isLast || !this.failover) {
if (allExceptions != null && allExceptions.size() == 1) {
throw allExceptions.get(0);
}
throw new AggregateMessageDeliveryException(message,
"All attempts to deliver Message to MessageHandlers failed.", allExceptions);
}
}
}