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

INT-684, Added thread safety to OrderedAwareLinkedHashSet and removed synchronization logic from AbstractDispatcher.addHandler(..) method

Browse Source
This commit is contained in:
Oleg Zhurakousky
2009-07-01 22:30:05 +00:00
parent 03acaacea2
commit ca5d0a9aa9
3 changed files with 227 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -89,9 +89,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
}
public boolean addHandler(MessageHandler handler) {
synchronized (this.handlerListMonitor) {
return this.handlers.add(handler);
}
return this.handlers.add(handler);
}
public boolean removeHandler(MessageHandler handler) {

137
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/OrderedAwareLinkedHashSet.java
View File

@@ -19,7 +19,8 @@ import java.util.Collection;
import java.util.Comparator;
import java.util.LinkedHashSet;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.springframework.core.OrderComparator;
import org.springframework.core.Ordered;
@@ -31,8 +32,8 @@ import org.springframework.util.CollectionUtils;
* Special Set that maintains the following semantics:
* All elements that are un-ordered (do not implement {@link Ordered} interface or annotated {@link Order} annotation)
* will be stored in the order in which they were added, maintaining the semantics of the {@link LinkedHashSet}.
* However, there is a special {@link Comparator} (instantiated by default) for this implementation of {@link Set},
* which is aware of the {@link Ordered} interface or and {@link Order} annotation. Those elements will have
* However, for all {@link Ordered} elements a {@link Comparator} (instantiated by default) for this implementation of {@link Set},
* will be used. Those elements will have
* precedence over un-ordered elements. If elements have the same order but themselves do not equal to one another
* they will be placed to the right (appended next to) of the element with the same order, thus preserving the order
* of the insertion and maintaining {@link LinkedHashSet} semantics.
@@ -41,77 +42,111 @@ import org.springframework.util.CollectionUtils;
* @since 1.0.3
*/
class OrderedAwareLinkedHashSet<E> extends LinkedHashSet<E> {
private TreeSet orderedSet = new TreeSet(new EqualsAwareOrderComparator());
OrderComparator comparator = new OrderComparator();
Lock lock = new ReentrantLock();
/**
* Every time when Ordered element is added via this method
* this Set will be re-sorted, otherwise the element is simply added to the end of the stack.
* If adding multiple objects for performance reasons it is recommended to use
* addAll(Collection c) method which first adds all the elements to this set
* and then calls reinitializeThis() method.
* Added element must not be null;
*/
public boolean add(E o){
boolean present = this.doAdd(o);
if (o instanceof Ordered){
this.reinitializeThis();
public boolean add(E o){
lock.lock();
try {
Assert.notNull(o,"Can not add NULL object");
boolean present = false;
if (o instanceof Ordered){
present = this.reinitializeThis(o);
} else {
present = super.add(o);
}
return present;
} finally {
lock.unlock();
}
return present;
}
/**
* Adds all elements in this Collection and then resorts this set
* via call to the reinitializeThis() method
*/
public boolean addAll(Collection<? extends E> c){
Assert.notNull(c,"Can not merge with NULL set");
public boolean addAll(Collection<? extends E> c){
lock.lock();
try {
Assert.notNull(c,"Can not merge with NULL set");
for (E object : c) {
this.add(object);
}
return true;
} finally {
lock.unlock();
}
}
/**
*
*/
public boolean remove(Object o){
lock.lock();
try {
return super.remove(o);
} finally {
lock.unlock();
}
}
/**
*
*/
public boolean removeAll(Collection<?> c){
if (CollectionUtils.isEmpty(c)){
return false;
}
for (E object : c) {
this.doAdd(object);
}
this.reinitializeThis();
// due to the nature of previous method
// at this point collection will always be modified
return true;
}
/**
*
* @param o
* @return
*/
private boolean doAdd(E o){
if (o instanceof Ordered){
return orderedSet.add(o);
} else {
return super.add(o);
lock.lock();
try {
super.removeAll(c);
return true;
} finally {
lock.unlock();
}
}
/**
*
*/
private void reinitializeThis(){
E[] tempUnorderedElements = (E[]) super.toArray();
E[] tempOrderedElements = (E[]) orderedSet.toArray();
private boolean reinitializeThis(Object adding){
boolean added = false;
E[] tempUnorderedElements = (E[]) this.toArray();
if (super.contains(adding)){
return false;
}
super.clear();
for (E object : tempOrderedElements) {
super.add(object);
}
for (E object : tempUnorderedElements) {
super.add(object);
}
}
/**
* Will reuse most of the functionality of OrderComparator, however if
* elements have the same order, then 1 will be returned, thus positioning
* such element to the right of the existing element.
*/
private static class EqualsAwareOrderComparator extends OrderComparator {
public int compare(Object o1, Object o2) {
int value = super.compare(o1, o2);
// see if objects are not equal
if (value == 0 && !o1.equals(o2)){
return 1;
if (tempUnorderedElements.length == 0){
added = super.add((E) adding);
} else {
Set tempSet = new LinkedHashSet();
for (E current : tempUnorderedElements) {
if (current instanceof Ordered && adding instanceof Ordered){
if (((Ordered)adding).getOrder() < ((Ordered)current).getOrder()){
added = super.add((E) adding);
super.add(current);
} else {
super.add(current);
}
} else {
tempSet.add(current);
}
}
if (!added){
added = super.add((E) adding);
}
for (Object object : tempSet) {
super.add((E) object);
}
return value;
}
return added;
}
}

140
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/OrderedAwareLinkedHashSetTests.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.dispatcher;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import org.junit.Test;
@@ -130,6 +131,145 @@ public class OrderedAwareLinkedHashSetTests {
assertEquals(o10, elements[9]);
}
@Test
public void testConcurrent(){
for(int i = 0; i < 1000; i++){
this.doConcurrent();
}
}
private void doConcurrent(){
final OrderedAwareLinkedHashSet setToTest = new OrderedAwareLinkedHashSet();
final Object o1 = new Foo(3);
final Object o2 = new Foo(1);
final Object o3 = new Foo(2);
final Object o4 = new Foo(2);
final Object o5 = new Foo(Ordered.LOWEST_PRECEDENCE);
final Object o6 = new Foo(Ordered.LOWEST_PRECEDENCE);
final Object o7 = new Foo(Ordered.HIGHEST_PRECEDENCE);
final Object o8 = new Foo(Ordered.HIGHEST_PRECEDENCE);
final Object o9 = new Foo(4);
final Object o10 = new Foo(2);
Thread t1 = new Thread(new Runnable() {
public void run() {
setToTest.add(o1);
setToTest.add(o3);
setToTest.add(o5);
setToTest.add(o7);
setToTest.add(o9);
}
});
Thread t2 = new Thread(new Runnable() {
public void run() {
setToTest.add(o2);
setToTest.add(o4);
setToTest.add(o6);
setToTest.add(o8);
setToTest.add(o10);
}
});
Thread t3 = new Thread(new Runnable() {
public void run() {
setToTest.add(1);
setToTest.add(new Foo(2));
setToTest.add(3);
setToTest.add(new Foo(9));
setToTest.add(8);
}
});
t1.start();
t2.start();
t3.start();
try {
t1.join();
t2.join();
t3.join();
} catch (Exception e) {
e.printStackTrace();
throw new RuntimeException(e);
}
assertEquals(15, setToTest.size());
}
/**
* Will test addAll operation including the removal and adding an object in the concurrent environment
*/
@Test
public void testConcurrentAll(){
for(int i = 0; i < 1000; i++){
this.doConcurrentAll();
}
}
public void doConcurrentAll(){
final List tempList = new ArrayList();
Object o1 = new Foo(3);
Object o2 = new Foo(1);
Object o3 = "Bla";
Object o4 = new Foo(2);
final Object o5 = new Foo(Ordered.LOWEST_PRECEDENCE);
Object o6 = new Foo(Ordered.LOWEST_PRECEDENCE);
final Object o7 = new Foo(Ordered.HIGHEST_PRECEDENCE);
Object o8 = new Foo(Ordered.HIGHEST_PRECEDENCE);
Object o9 = new Foo(4);
Object o10 = "Baz";
tempList.add(o1);
tempList.add(o2);
tempList.add(o3);
tempList.add(o4);
tempList.add(o5);
tempList.add(o6);
tempList.add(o7);
tempList.add(o8);
tempList.add(o9);
tempList.add(o10);
final OrderedAwareLinkedHashSet orderAwareSet = new OrderedAwareLinkedHashSet();
Thread t1 = new Thread(new Runnable() {
public void run() {
orderAwareSet.addAll(tempList);
orderAwareSet.remove(o5);
orderAwareSet.remove(o7);
}
});
final List tempList2 = new ArrayList();
final Foo foo5 = new Foo(5);
Foo foo6 = new Foo(6);
tempList2.add(foo6);
tempList2.add(foo5);
tempList2.add(new Foo(30));
tempList2.add(new Foo(10));
tempList2.add(1);
tempList2.add(new Foo(28));
tempList2.add(10);
tempList2.add(13);
tempList2.add(new Foo(63));
Thread t2 = new Thread(new Runnable() {
public void run() {
orderAwareSet.addAll(tempList2);
orderAwareSet.remove(foo5);
}
});
Thread t3 = new Thread(new Runnable() {
public void run() {
orderAwareSet.add("hello");
orderAwareSet.add("hello again");
}
});
t1.start();
t2.start();
t3.start();
try {
t1.join();
t2.join();
t3.join();
} catch (Exception e) {
e.printStackTrace();
throw new RuntimeException(e);
}
Object[] elements = orderAwareSet.toArray();
assertEquals(18, elements.length);
}
private static class Foo implements Ordered {
private int order;
public Foo(int order){