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

Simplified code and added concurrent tests
This commit is contained in:
Iwein Fuld
2009-03-09 19:14:21 +00:00
parent 242d8291ec
commit 01fe7a0baf
8 changed files with 223 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,7 +16,7 @@
package org.springframework.integration.channel;
import org.springframework.integration.dispatcher.AbstractSendOnceDispatcher;
import org.springframework.integration.dispatcher.AbstractWinningHandlerDispatcher;
import org.springframework.integration.dispatcher.LoadBalancingDispatcher;
/**
@@ -27,12 +27,12 @@ import org.springframework.integration.dispatcher.LoadBalancingDispatcher;
* @author Mark Fisher
* @author Iwein Fuld
*/
public class DirectChannel extends AbstractSubscribableChannel<AbstractSendOnceDispatcher> {
public class DirectChannel extends AbstractSubscribableChannel<AbstractWinningHandlerDispatcher> {
public DirectChannel() {
super(new LoadBalancingDispatcher());
}
public DirectChannel(AbstractSendOnceDispatcher dispatcher){
public DirectChannel(AbstractWinningHandlerDispatcher dispatcher){
super(dispatcher);
}

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

@@ -16,8 +16,9 @@
package org.springframework.integration.dispatcher;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -37,7 +38,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
protected final Log logger = LogFactory.getLog(this.getClass());
private final Queue<MessageHandler> handlers = new ConcurrentLinkedQueue<MessageHandler>();
private final List<MessageHandler> handlers = new ArrayList<MessageHandler>();
private volatile TaskExecutor taskExecutor;
@@ -54,15 +55,15 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
return this.taskExecutor;
}
protected Queue<MessageHandler> getHandlers() {
return handlers;
protected List<MessageHandler> getHandlers() {
return Collections.unmodifiableList(handlers);
}
public boolean addHandler(MessageHandler handler) {
if (this.handlers.contains(handler)) {
return false;
}
return this.handlers.offer(handler);
return this.handlers.add(handler);
}
public boolean removeHandler(MessageHandler handler) {

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

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

61
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractWinningHandlerDispatcher.java Normal file
View File

@@ -0,0 +1,61 @@
package org.springframework.integration.dispatcher;
import java.util.ArrayList;
import java.util.List;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageRejectedException;
/**
* Implementation of {@link MessageDispatcher} that will attempt to send a
* {@link Message} to at most one of its handlers. The handlers will be tried
* one by one. As soon as <em>one</em> of the handlers accepts the Message, the
* dispatcher will return <code>true</code> and ignore the rest of it's
* handlers.
* <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.
* <p/>
* The implementations of this class control the order in which the handlers
* will be tried through the <code>getNextHandler</code> method.
*
* @author Iwein Fuld
*
*/
public abstract class AbstractWinningHandlerDispatcher extends AbstractDispatcher {
public final boolean dispatch(Message<?> message) {
boolean success = false;
List<MessageHandler> triedHandlers = new ArrayList<MessageHandler>();
int size;
List<MessageHandler> handlers = this.getHandlers();
if (handlers.isEmpty()) {
throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
}
size = handlers.size();
for (int i = 0; triedHandlers.size() < size && success == false; i++) {
MessageHandler handler = handlers.get(getNextHandlerIndex(size, i));
if (this.sendMessageToHandler(message, handler)) {
success = true; //we have a winner.
}
triedHandlers.add(handler);
}
if (!success) {
throw new MessageRejectedException(message, "All of dispatcher's subscribers rejected Message.");
}
return success;
}
/**
* Return the next handler index. Subclasses have to implement this method
* to determine the order in which handlers will be invoked.
*
* @param size the total number of handlers
* @param loopIndex the current index of the loop
*/
protected abstract int getNextHandlerIndex(int size, int loopIndex);
}

40
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/FailOverDispatcher.java
View File

@@ -16,42 +16,22 @@
package org.springframework.integration.dispatcher;
import java.util.Iterator;
import org.springframework.integration.core.Message;
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'.
* <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.
* {@link AbstractWinningHandlerDispatcher} that will try it's handlers in the
* same order every dispatch.
*
* @author Mark Fisher
* @author Iwein Fuld
*/
public class FailOverDispatcher extends AbstractSendOnceDispatcher {
public class FailOverDispatcher extends AbstractWinningHandlerDispatcher {
public boolean dispatch(Message<?> message) {
if (this.getHandlers().size() == 0) {
throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
}
Iterator<MessageHandler> handlerIterator = this.getHandlers().iterator();
boolean sent = false;
while (sent == false && handlerIterator.hasNext()) {
if (this.sendMessageToHandler(message, handlerIterator.next())) {
sent = true;
}
}
if (!sent) {
throw new MessageRejectedException(message, "All of dispatcher's subscribers rejected Message.");
}
return sent;
/**
* Returns the current loop index, so each <code>dispatch</code> will try
* the handlers in the same order.
*/
@Override
protected int getNextHandlerIndex(int size, int loopIndex) {
return loopIndex;
}
}

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

@@ -15,52 +15,26 @@
*/
package org.springframework.integration.dispatcher;
import java.util.Queue;
import java.util.concurrent.locks.ReentrantLock;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageRejectedException;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Round-robin 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'. This
* implementation will load balance over its handlers using a round-robin
* strategy.
* <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.
* Round-robin implementation of {@link AbstractWinningHandlerDispatcher}. This
* implementation will keep track of the index in which its handlers have been
* tried and return a different handler every dispatch.
*
* @author Iwein Fuld
*/
public class LoadBalancingDispatcher extends AbstractSendOnceDispatcher {
public class LoadBalancingDispatcher extends AbstractWinningHandlerDispatcher {
private ReentrantLock queueLock = new ReentrantLock();
public boolean dispatch(Message<?> message) {
queueLock.lock();
Queue<MessageHandler> handlers = this.getHandlers();
if (handlers.isEmpty()) {
throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
}
boolean success = false;
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;
}
}
if (!success) {
throw new MessageRejectedException(message, "All of dispatcher's subscribers rejected Message.");
}
return success;
private AtomicInteger currentHandlerIndex = new AtomicInteger();
/**
* Keeps track of the last index over multiple <code>dispatch</code>
* invocations. Each invocation of this method will increment the index by
* one, overflowing at <code>size</code>. <code>loopIndex</code> is ignored.
*/
protected int getNextHandlerIndex(int size, int loopIndex) {
int indexTail = currentHandlerIndex.getAndIncrement() % size;
return indexTail < 0 ? indexTail + size : indexTail;
}
}

111
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/LoadBalancingDispatcherConcurrentTests.java Normal file
View File

@@ -0,0 +1,111 @@
package org.springframework.integration.dispatcher;
import static org.junit.Assert.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnit44Runner;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
@RunWith(MockitoJUnit44Runner.class)
public class LoadBalancingDispatcherConcurrentTests {
private static final int TOTAL_EXECUTIONS = 40;
private AbstractWinningHandlerDispatcher dispatcher = new LoadBalancingDispatcher();
private ThreadPoolTaskExecutor scheduler = new ThreadPoolTaskExecutor();
@Mock
private MessageHandler handler1;
@Mock
private MessageHandler handler2;
@Mock
private MessageHandler handler3;
@Mock
private MessageHandler handler4;
@Mock
private Message<?> message;
@Before
public void initialize() throws Exception {
scheduler.setCorePoolSize(10);
scheduler.setMaxPoolSize(10);
scheduler.initialize();
}
@Test(timeout = 1000)
public void noHandlerExhaustion() throws Exception {
dispatcher.addHandler(handler1);
dispatcher.addHandler(handler2);
dispatcher.addHandler(handler3);
dispatcher.addHandler(handler4);
final CountDownLatch start = new CountDownLatch(1);
final CountDownLatch allDone = new CountDownLatch(TOTAL_EXECUTIONS);
final Message<?> message = this.message;
final AtomicBoolean failed = new AtomicBoolean(false);
Runnable messageSenderTask = new Runnable() {
public void run() {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
if (!dispatcher.dispatch(message)) {
failed.set(true);
}
allDone.countDown();
}
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
scheduler.execute(messageSenderTask);
}
start.countDown();
allDone.await();
assertFalse("not all messages were accepted", failed.get());
}
@Test(timeout=2000)
public void unlockOnFailure() throws Exception {
// dispatcher has no subscribers (shouldn't lead to deadlock)
final CountDownLatch start = new CountDownLatch(1);
final CountDownLatch allDone = new CountDownLatch(TOTAL_EXECUTIONS);
final Message<?> message = this.message;
Runnable messageSenderTask = new Runnable() {
public void run() {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
try {
dispatcher.dispatch(message);
fail("this shouldn't happen");
}
catch (MessageDeliveryException e) {
//expected
}
allDone.countDown();
}
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
scheduler.execute(messageSenderTask);
}
start.countDown();
allDone.await();
}
}

17
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/LoadBalancingDispatcherTests.java
View File

@@ -17,10 +17,13 @@ package org.springframework.integration.dispatcher;
import static org.mockito.Mockito.*;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnit44Runner;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageHandler;
@@ -32,7 +35,7 @@ import org.springframework.integration.message.MessageHandler;
@RunWith(MockitoJUnit44Runner.class)
public class LoadBalancingDispatcherTests {
private LoadBalancingDispatcher dispatcher = new LoadBalancingDispatcher();
private AbstractWinningHandlerDispatcher dispatcher = new LoadBalancingDispatcher();
@Mock
private MessageHandler handler;
@@ -58,4 +61,16 @@ public class LoadBalancingDispatcherTests {
verify(handler).handleMessage(message);
verify(differentHandler).handleMessage(message);
}
@Test
public void overFlowCurrentHandlerIndex() throws Exception {
dispatcher.addHandler(handler);
dispatcher.addHandler(differentHandler);
DirectFieldAccessor accessor = new DirectFieldAccessor(dispatcher);
((AtomicInteger)accessor.getPropertyValue("currentHandlerIndex")).set(Integer.MAX_VALUE-5);
for(long i=0; i < 40; i++){
dispatcher.dispatch(message);
}
verify(handler, atLeast(18)).handleMessage(message);
verify(differentHandler, atLeast(18)).handleMessage(message);
}
}