簡述：

《Effective Java》第67條避免過度同步

知識點：

1. synchronized 集合的時候，刪除集合元素出現，併發修改和死鎖的問題

2. CopyOnWriteArray, 一種在寫操作時都會進行拷貝的併發集合（concurrent collection）

程式碼：

ForwardingSet.java

package com.anialy.test.concurrency;

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

public class ForwardingSet<E> implements Set<E> {

	private final Set<E> s;

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

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

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

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

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

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

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

	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 retainAll(Collection<?> c) {
		return c.retainAll(c);
	}

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

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

}
 

package com.anialy.test.concurrency;

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

public class ObservableSet<E> extends ForwardingSet<E> {

	public ObservableSet(Set<E> s) {
		super(s);
	}

	private final List<SetObserver<E>> observers = 
			new ArrayList<SetObserver<E>>();
	
	public void addObserver(SetObserver<E> observer){
		synchronized (observer) {
			observers.add(observer);
		}
	}
	
	public boolean removeObserver(SetObserver<E> observer){
		synchronized (observer) {
			return observers.remove(observer);
		}
	}
	
	private void notifyElemetnAdded(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)
			notifyElemetnAdded(element);
		return added;
	}
	
	@Override
	public boolean addAll(Collection<? extends E> c) {
		boolean result = false;
		for(E element : c){
			result |= add(element);
		}
		return result;
	}
}

 

SetObserver.java

package com.anialy.test.concurrency;

public interface SetObserver<E> {
	void added(ObservableSet<E> set, E element);
}

package com.anialy.test.concurrency;

import java.util.HashSet;

public class Test {
	public static void main(String[] args) {
		ObservableSet<Integer> set =
				new ObservableSet<Integer>(new HashSet<Integer>());
		
		// add 方法呼叫後會觸發notifyElemetnAdded(E element)方法
		// 執行 SetObserver added方法
		set.addObserver(new SetObserver<Integer>() {
			public void added(ObservableSet<Integer> set, Integer element) {
				System.out.println(element);
			}
		});
		
		for(int i = 0; i < 100; i++){
			set.add(i);
		}
	}
}
 

然後，修改SetObserver的added方法，其中移除SetObserver介面，預期是希望此時的set集合不再繼續列印剩餘數字

package com.anialy.test.concurrency;

import java.util.HashSet;

public class Test {
	public static void main(String[] args) {
		ObservableSet<Integer> set =
				new ObservableSet<Integer>(new HashSet<Integer>());
		
		// add 方法呼叫後會觸發notifyElemetnAdded(E element)方法
		// 執行 SetObserver added方法
		set.addObserver(new SetObserver<Integer>() {
			public void added(ObservableSet<Integer> set, Integer element) {
				System.out.println(element);
				if(element == 23)
					set.removeObserver(this);
			}
		});
		
		for(int i = 0; i < 100; i++){
			set.add(i);
		}
	}
}

書中解釋的原因：

我們正在企圖遍歷列表的過程中，講一個元素從列表中刪除，這是非法的。notifyElementAdded方法中的迭代是在一個同步塊中，

可以防止併發的修改，但是無法防止迭代執行緒本身回撥到可觀察的集合中，也無法防止修改它的observers列表

之後使用Executors.newSingleThreadExecutor()庫函式

package com.anialy.test.concurrency;

import java.util.HashSet;
import java.util.concurrent.ExecutionException;
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>());

		// add 方法呼叫後會觸發notifyElemetnAdded(E element)方法
		// 執行 SetObserver added方法
		set.addObserver(new SetObserver<Integer>() {
			public void added(final ObservableSet<Integer> set, Integer element) {
				System.out.println(element);
				if(element == 23){
					ExecutorService executor = Executors.newSingleThreadExecutor();
					final SetObserver<Integer> observer = this;
					try {
						executor.submit(new Runnable(){
							public void run() {
								set.removeObserver(observer);
							}
						}).get();
					} catch (InterruptedException e) {
						e.printStackTrace();
					} catch (ExecutionException e) {
						e.printStackTrace();
					}
				}
			}
		});

		for(int i = 0; i < 100; i++){
			set.add(i);
		}
	}
}

正如這裡所示，

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

在這期間，主執行緒一直在等待後臺執行緒來完成對觀察這的刪除。

解決方式：

1. 對於notifyElementAdded中的collection在遍歷前做一個快照

	private void notifyElemetnAdded(E element){
		List<SetObserver<E>> snapshot = null;
		synchronized (observers) {
			snapshot = new ArrayList<SetObserver<E>>(observers);
		}
		for(SetObserver<E> observer : snapshot){
			observer.added(this, element);
		}
	}

其是ArrayList的一種變體, 通過重新拷貝整個底層陣列，在這裡實現所有的寫操作

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

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

可以看看CopyOnWriteArrayList的remove及add操作,每個寫操作都會進行拷貝，就不會出現之前鎖住集合而後遍歷的情況了

    /**
     * Inserts the specified element at the specified position in this
     * list. Shifts the element currently at that position (if any) and
     * any subsequent elements to the right (adds one to their indices).
     *
     * @throws IndexOutOfBoundsException {@inheritDoc}
     */
    public void add(int index, E element) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            Object[] elements = getArray();
            int len = elements.length;
            if (index > len || index < 0)
                throw new IndexOutOfBoundsException("Index: "+index+
                                                    ", Size: "+len);
            Object[] newElements;
            int numMoved = len - index;
            if (numMoved == 0)
                newElements = Arrays.copyOf(elements, len + 1);
            else {
                newElements = new Object[len + 1];
                System.arraycopy(elements, 0, newElements, 0, index);
                System.arraycopy(elements, index, newElements, index + 1,
                                 numMoved);
            }
            newElements[index] = element;
            setArray(newElements);
        } finally {
            lock.unlock();
        }
    }

    /**
     * Removes the element at the specified position in this list.
     * Shifts any subsequent elements to the left (subtracts one from their
     * indices).  Returns the element that was removed from the list.
     *
     * @throws IndexOutOfBoundsException {@inheritDoc}
     */
    public E remove(int index) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            Object[] elements = getArray();
            int len = elements.length;
            E oldValue = get(elements, index);
            int numMoved = len - index - 1;
            if (numMoved == 0)
                setArray(Arrays.copyOf(elements, len - 1));
            else {
                Object[] newElements = new Object[len - 1];
                System.arraycopy(elements, 0, newElements, 0, index);
                System.arraycopy(elements, index + 1, newElements, index,
                                 numMoved);
                setArray(newElements);
            }
            return oldValue;
        } finally {
            lock.unlock();
        }
    }