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

IN PROGRESS - issue INT-567: Add round-robin dispatching strategy

Browse Source 
http://jira.springframework.org/browse/INT-567

- refactored AbstractDispatcher to use Queue instead of Set
- added AbstractSendOnceDispatcher (for DirectChannel parametrization)
- refactored DirectChannel to use LoadBalancing Dispatcher by default
- renamed SimpleDispatcher to FailOverDispatcher
This commit is contained in:
Iwein Fuld
2009-03-07 19:29:01 +00:00
parent dabb76ac26
commit 242d8291ec
10 changed files with 181 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,7 +16,8 @@
package org.springframework.integration.channel;
import org.springframework.integration.dispatcher.SimpleDispatcher;
import org.springframework.integration.dispatcher.AbstractSendOnceDispatcher;
import org.springframework.integration.dispatcher.LoadBalancingDispatcher;
/**
* A channel that invokes a single subscriber for each sent Message.
@@ -24,11 +25,15 @@ import org.springframework.integration.dispatcher.SimpleDispatcher;
*
* @author Dave Syer
* @author Mark Fisher
* @author Iwein Fuld
*/
public class DirectChannel extends AbstractSubscribableChannel<SimpleDispatcher> {
public class DirectChannel extends AbstractSubscribableChannel<AbstractSendOnceDispatcher> {
public DirectChannel() {
super(new SimpleDispatcher());
super(new LoadBalancingDispatcher());
}
public DirectChannel(AbstractSendOnceDispatcher dispatcher){
super(dispatcher);
}
}

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

@@ -18,7 +18,9 @@ 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.xml.ParserContext;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
@@ -33,6 +35,7 @@ public class PointToPointChannelParser extends AbstractChannelParser {
private static final String CHANNEL_PACKAGE = IntegrationNamespaceUtils.BASE_PACKAGE + ".channel";
private static final String DISPATCHER_PACKAGE = IntegrationNamespaceUtils.BASE_PACKAGE + ".dispatcher";
@Override
protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) {
@@ -55,10 +58,24 @@ 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("fail-over")) {
BeanDefinitionBuilder dispatcherBuilder = BeanDefinitionBuilder
.genericBeanDefinition(DISPATCHER_PACKAGE + ".FailOverDispatcher");
dispatcherBuilder.setRole(BeanDefinition.ROLE_SUPPORT);
builder.addConstructorArgReference(BeanDefinitionReaderUtils.registerWithGeneratedName(dispatcherBuilder
.getBeanDefinition(), parserContext.getRegistry()));
}
}
// rely on the default for round-robin
}
private void parseQueueCapacity(BeanDefinitionBuilder builder, Element queueElement) {
String capacity = queueElement.getAttribute("capacity");
if (StringUtils.hasText(capacity)) {

14
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/spring-integration-1.0.xsd
View File

@@ -59,6 +59,7 @@
</xsd:appinfo>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="channelType">
<xsd:sequence>
@@ -163,6 +164,14 @@
</xsd:annotation>
<xsd:attribute name="id" type="xsd:ID" use="required" />
<xsd:attribute name="datatype" type="xsd:string" />
<xsd:attribute name="dispatcher">
<xsd:simpleType>
<xsd:restriction base="xsd:string">
<xsd:enumeration value="fail-over" />
<xsd:enumeration value="round-robin" />
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
<xsd:element name="gateway">
@@ -789,7 +798,8 @@
</xsd:attribute>
<xsd:attribute name="method" type="xsd:string" />
<xsd:attribute name="resolution-required" type="xsd:boolean" />
<xsd:attribute name="ignore-channel-name-resolution-failures" type="xsd:boolean" />
<xsd:attribute name="ignore-channel-name-resolution-failures"
type="xsd:boolean" />
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -823,7 +833,7 @@
</xsd:attribute>
<xsd:attribute name="completion-strategy-method"
type="xsd:string" />
<xsd:attribute name="correlation-strategy" type="xsd:string">
<xsd:attribute name="correlation-strategy" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">

49
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java
View File

@@ -16,14 +16,11 @@
package org.springframework.integration.dispatcher;
import java.util.Collection;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageHandler;
@@ -40,22 +37,14 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
protected final Log logger = LogFactory.getLog(this.getClass());
private final Set<MessageHandler> handlers = new CopyOnWriteArraySet<MessageHandler>();
private final Queue<MessageHandler> handlers = new ConcurrentLinkedQueue<MessageHandler>();
private volatile TaskExecutor taskExecutor;
public boolean addHandler(MessageHandler handler) {
return this.handlers.add(handler);
}
public boolean removeHandler(MessageHandler handler) {
return this.handlers.remove(handler);
}
/**
* Specify a {@link TaskExecutor} for invoking the handlers.
* If none is provided, the invocation will occur in the thread
* that runs this polling dispatcher.
* Specify a {@link TaskExecutor} for invoking the handlers. If none is
* provided, the invocation will occur in the thread that runs this polling
* dispatcher.
*/
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
@@ -65,10 +54,21 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
return this.taskExecutor;
}
protected Set<MessageHandler> getHandlers() {
return Collections.unmodifiableSet(handlers);
protected Queue<MessageHandler> getHandlers() {
return handlers;
}
public boolean addHandler(MessageHandler handler) {
if (this.handlers.contains(handler)) {
return false;
}
return this.handlers.offer(handler);
}
public boolean removeHandler(MessageHandler handler) {
return this.handlers.remove(handler);
}
public String toString() {
return this.getClass().getSimpleName() + " with handlers: " + this.handlers;
}
@@ -78,7 +78,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
* "Selective Consumer" throws a {@link MessageRejectedException}.
*/
protected boolean sendMessageToHandler(Message<?> message, MessageHandler handler) {
Assert.notNull(message, "'message' must not be null");
Assert.notNull(message, "'message' must not be null.");
Assert.notNull(handler, "'handler' must not be null.");
try {
handler.handleMessage(message);
@@ -86,7 +86,12 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
}
catch (MessageRejectedException e) {
if (logger.isDebugEnabled()) {
logger.debug("Handler '" + handler + "' rejected Message, if other handlers are available this dispatcher may try to send to those.", e);
logger
.debug(
"Handler '"
+ handler
+ "' rejected Message, if other handlers are available this dispatcher may try to send to those.",
e);
}
return false;
}

9
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractSendOnceDispatcher.java Normal file
View File

@@ -0,0 +1,9 @@
package org.springframework.integration.dispatcher;
import org.springframework.integration.core.Message;
public abstract class AbstractSendOnceDispatcher extends AbstractDispatcher {
public abstract boolean dispatch(Message<?> message);
}

34
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java → org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java
View File

@@ -16,42 +16,42 @@
package org.springframework.integration.dispatcher;
import java.util.Iterator;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageRejectedException;
/**
* Basic implementation of {@link MessageDispatcher} that will attempt
* to send a {@link Message} to one of its handlers. As soon as <em>one</em>
* of the handlers accepts the Message, the dispatcher will return 'true'.
* Basic implementation of {@link MessageDispatcher} that will attempt to send a
* {@link Message} to one of its handlers. As soon as <em>one</em> of the
* handlers accepts the Message, the dispatcher will return 'true'.
* <p>
* If the dispatcher has no handlers, a {@link MessageDeliveryException}
* will be thrown. If all handlers reject the Message, the dispatcher will
* throw a MessageRejectedException.
* If the dispatcher has no handlers, a {@link MessageDeliveryException} will be
* thrown. If all handlers reject the Message, the dispatcher will throw a
* MessageRejectedException.
*
* @author Mark Fisher
* @author Iwein Fuld
*/
public class SimpleDispatcher extends AbstractDispatcher {
public class FailOverDispatcher extends AbstractSendOnceDispatcher {
public boolean dispatch(Message<?> message) {
if (this.getHandlers().size() == 0) {
throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
}
int count = 0;
int rejectedExceptionCount = 0;
for (MessageHandler handler : this.getHandlers()) {
count++;
if (this.sendMessageToHandler(message, handler)) {
return true;
Iterator<MessageHandler> handlerIterator = this.getHandlers().iterator();
boolean sent = false;
while (sent == false && handlerIterator.hasNext()) {
if (this.sendMessageToHandler(message, handlerIterator.next())) {
sent = true;
}
rejectedExceptionCount++;
}
if (rejectedExceptionCount == count) {
if (!sent) {
throw new MessageRejectedException(message, "All of dispatcher's subscribers rejected Message.");
}
return false;
return sent;
}
}

25
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/LoadBalancingDispatcher.java
View File

@@ -15,11 +15,8 @@
*/
package org.springframework.integration.dispatcher;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.ReentrantLock;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageDeliveryException;
@@ -39,21 +36,24 @@ import org.springframework.integration.message.MessageRejectedException;
*
* @author Iwein Fuld
*/
public class LoadBalancingDispatcher extends AbstractDispatcher {
public class LoadBalancingDispatcher extends AbstractSendOnceDispatcher {
private final Queue<MessageHandler> handlerQueue = new ConcurrentLinkedQueue<MessageHandler>();
private ReentrantLock queueLock = new ReentrantLock();
public boolean dispatch(Message<?> message) {
Set<MessageHandler> handlers = new HashSet<MessageHandler>(this.getHandlers());
queueLock.lock();
Queue<MessageHandler> handlers = this.getHandlers();
if (handlers.isEmpty()) {
throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
}
if (this.handlerQueue.isEmpty()){
handlerQueue.addAll(handlers);
}
boolean success = false;
while (!handlerQueue.isEmpty() && success == false) {
MessageHandler handler = handlerQueue.poll();
int size = handlers.size();
queueLock.unlock();
for (int i = 0; i < size && success == false; i++) {
queueLock.lock();
MessageHandler handler = handlers.poll();
handlers.offer(handler);
queueLock.unlock();
if (this.sendMessageToHandler(message, handler)) {
success = true;
}
@@ -63,5 +63,4 @@ public class LoadBalancingDispatcher extends AbstractDispatcher {
}
return success;
}
}