观察者模式与多线程

观察者模式是很常用、常见当然也相当重要的一种设计模式，其中主要包括的概念有：

1.Observer（观察者，又具化作ActionListener）

2.Observable（被观察者）

3.Action（观察者所感兴趣的被观察者的行为）

 

被观察者主动与观察者建立契约关系（添加观察者），观察者本身具有行为发生被告知资格，即拥有行为响应方法，如actionPerformed，下面贴出一段程序：

 

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;

// Broken - invokes alien method from synchronized block!
public class ObservableSet<E> extends ForwardingSet<E> {
	public ObservableSet(Set<E> set) {
		super(set);
	}

	private final List<SetObserver<E>> observers = new ArrayList<SetObserver<E>>();

	public void addObserver(SetObserver<E> observer) {
		synchronized (observers) {
			observers.add(observer);
		}
	}

	public boolean removeObserver(SetObserver<E> observer) {
		synchronized (observers) {
			return observers.remove(observer);
		}
	}

	private void notifyElementAdded(E element) {
		synchronized (observers) {
			for (SetObserver<E> observer : observers)
				observer.added(this, element);
		}
	}

	@Override
	public boolean add(E element) {
		boolean added = super.add(element);
		if (added)
			notifyElementAdded(element);
		return added;
	}

	@Override
	public boolean addAll(Collection<? extends E> c) {
		boolean result = false;
		for (E element : c)
			result |= add(element); // calls notifyElementAdded
		return result;
	}
}

 

public interface SetObserver<E> {
	// Invoked when an element is added to the observable set
	void added(ObservableSet<E> set, E element);
}

 

import java.util.Collection;
import java.util.Iterator;
import java.util.Set;

// Reusable forwarding class
public class ForwardingSet<E> implements Set<E> {
	private final Set<E> s;

	public ForwardingSet(Set<E> s) {
		this.s = s;
	}

	public void clear() {
		s.clear();
	}

	public boolean contains(Object o) {
		return s.contains(o);
	}

	public boolean isEmpty() {
		return s.isEmpty();
	}

	public int size() {
		return s.size();
	}

	public Iterator<E> iterator() {
		return s.iterator();
	}

	public boolean add(E e) {
		return s.add(e);
	}

	public boolean remove(Object o) {
		return s.remove(o);
	}

	public boolean containsAll(Collection<?> c) {
		return s.containsAll(c);
	}

	public boolean addAll(Collection<? extends E> c) {
		return s.addAll(c);
	}

	public boolean removeAll(Collection<?> c) {
		return s.removeAll(c);
	}

	public boolean retainAll(Collection<?> c) {
		return s.retainAll(c);
	}

	public Object[] toArray() {
		return s.toArray();
	}

	public <T> T[] toArray(T[] a) {
		return s.toArray(a);
	}

	@Override
	public boolean equals(Object o) {
		return s.equals(o);
	}

	@Override
	public int hashCode() {
		return s.hashCode();
	}

	@Override
	public String toString() {
		return s.toString();
	}
}

 

下面是测试类：

import java.util.HashSet;

public class Test {

	public static void main(String[] args) {
		ObservableSet<Integer> set = new ObservableSet<Integer>(new HashSet<Integer>());
		set.addObserver(new SetObserver<Integer>() {
			public void added(ObservableSet<Integer> s, Integer e) {
				System.out.println(e);
			}
		});
		for (int i = 0; i < 100; i++)
			set.add(i);
	}
}

 如此便有了一个观察者模式的简单示例，但如果往测试类中加一句：

import java.util.HashSet;

public class Test {

	public static void main(String[] args) {
		ObservableSet<Integer> set = new ObservableSet<Integer>(new HashSet<Integer>());
		set.addObserver(new SetObserver<Integer>() {
			public void added(ObservableSet<Integer> s, Integer e) {
				System.out.println(e);
				if (e == 23) s.removeObserver(this);
			}
		});
		for (int i = 0; i < 100; i++)
			set.add(i);
	}
}

 会发现，报以下错误：

Exception in thread "main" java.util.ConcurrentModificationException
    at java.util.AbstractList$Itr.checkForComodification(AbstractList.java:372)
    at java.util.AbstractList$Itr.next(AbstractList.java:343)
    at ObservableSet.notifyElementAdded(ObservableSet.java:28)
    at ObservableSet.add(ObservableSet.java:37)
    at Test.main(Test.java:14)

 

The problem is that notifyElementAdded is
in the process of iterating over the observers list when it invokes the observer’s
added method. The added method calls the observable set’s removeObserver
method, which in turn calls observers.remove. Now we are in trouble. We are
trying to remove an element from a list in the midst of iterating over it, which is
illegal. The iteration in the notifyElementAdded method is in a synchronized
block to prevent concurrent modification, but it doesn’t prevent the iterating thread
itself from calling back into the observable set and modifying its observers list.

问题是以下两个方法被同时调用了，当然是同一个线程，所以synchronized关键字对这种行为无效：

public boolean removeObserver(SetObserver<E> observer) {
    synchronized (observers) {
        return observers.remove(observer);
    }
}

private void notifyElementAdded(E element) {
    synchronized (observers) {
        for (SetObserver<E> observer : observers)
	    observer.added(this, element);
    }
}

 问题本质是遍历线程对在处于遍历操作中的List进行remove操作，但改成如下又会造成死锁：

// Observer that uses a background thread needlessly
set.addObserver(new SetObserver<Integer>() {
    public void added(final ObservableSet<Integer> s, Integer e) {
        System.out.println(e);
        if (e == 23) {
            ExecutorService executor = Executors.newSingleThreadExecutor();
            final SetObserver<Integer> observer = this;
            try {
                executor.submit(new Runnable() {
                    public void run() {
                        s.removeObserver(observer);
                    }
                }).get();
            } catch (ExecutionException ex) {
                throw new AssertionError(ex.getCause());
            } catch (InterruptedException ex) {
                throw new AssertionError(ex.getCause());
            } finally {
                executor.shutdown();
            }
        }    
    }
});

 

The background thread
calls s.removeObserver, which attempts to lock observers, but it can’t acquire
the lock, because the main thread already has the lock. All the while, the main
thread is waiting for the background thread to finish removing the observer, which
explains the deadlock.

 

但这样就不会造成死锁：

import java.util.HashSet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Test {

	public static void main(String[] args) {
		ObservableSet<Integer> set = new ObservableSet<Integer>(new HashSet<Integer>());
		// Observer that uses a background thread needlessly
		set.addObserver(new SetObserver<Integer>() {
			public void added(final ObservableSet<Integer> s, Integer e) {
				System.out.println(e);
				if (e == 23) {
					ExecutorService executor = Executors.newSingleThreadExecutor();
					final SetObserver<Integer> observer = this;
					try {
						executor.submit(new Runnable() {
							public void run() {
								s.removeObserver(observer);
							}
						});
					} finally {
						executor.shutdown();
					}
				}
			}
		});
		
		for (int i = 0; i < 100; i++) {
			set.add(i);
		}
	}
}

 原因应该是backgroud thread在竞争锁时能等待main thread执行完毕（其不阻塞），但也不会打印完整100个数，因为observer会被remove，再调用notifyElementAdded自然也就没有意义，当然原来的方法改变notifyElementAdded也不会造成死锁：

// Alien method moved outside of synchronized block - open calls
private void notifyElementAdded(E element) {
    List<SetObserver<E>> snapshot = null;
    synchronized(observers) {
        snapshot = new ArrayList<SetObserver<E>>(observers);
    }
    for (SetObserver<E> observer : snapshot)
        observer.added(this, element);
}

 还有更好的方法：

// Thread-safe observable set with CopyOnWriteArrayList
private final List<SetObserver<E>> observers =
new CopyOnWriteArrayList<SetObserver<E>>();

public void addObserver(SetObserver<E> observer) {
    observers.add(observer);
}

public boolean removeObserver(SetObserver<E> observer) {
    return observers.remove(observer);
}

private void notifyElementAdded(E element) {
    for (SetObserver<E> observer : observers)
        observer.added(this, element);
}

 In fact, there’s a better way to move the alien method invocations out of the
synchronized block. Since release 1.5, the Java libraries have provided a concurrent
collection (Item 69) known as CopyOnWriteArrayList, which is tailor-made
for this purpose. It is a variant of ArrayList in which all write operations are
implemented by making a fresh copy of the entire underlying array. Because the
internal array is never modified, iteration requires no locking and is very fast. For
most uses, the performance of CopyOnWriteArrayList would be atrocious, but
it’s perfect for observer lists, which are rarely modified and often traversed.

应该是创造了在遍历过程中不会被它线程更改数组结构的机制，而在遍历后或再次遍历时能将写效果实现于读效果之上