java 多线程 同步 观察者 并发集合的一个例子

Posted 2021-01-29 听哥哥的话

//第一版
package com.hra.riskprice;

import com.hra.riskprice.SysEnum.Factor_Type;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import javax.swing.text.html.HTMLDocument;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.TimeUnit;

 class ForwardingSet<E> implements  Set<E>{
    private final Set<E> s;
    public ForwardingSet(Set<E> s){this.s=s;}

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

    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);
    }

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

    @Override
    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();
    }

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

 interface  SetObserver<E>{

    void added(ObservableSet<E> set,E element);
}

 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);
        }
    }
    public void notifyElementAdded(E element){
        synchronized (observers){
            for(SetObserver<E> observer:observers){
                observer.added(this,element);
            }
        }
    }

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

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

@SpringBootApplication
public class RiskpriceApplication {



    public static void main(String[] args)  {

        ObservableSet<Integer> set=new ObservableSet<Integer>(new HashSet<Integer>());
        set.addObserver(new SetObserver<Integer>() {
            @Override
            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);
        }
    }
}
你觉得会打印0~23吗，实际上运行后就挂了，for循环遍历过程中，不允许修改枚举列表，我们可以考虑通过另外一个线程去移除这个观察者，也是下面过度得第二版了 通过 ExecutorService 

//第二版
package com.hra.riskprice;

import com.hra.riskprice.SysEnum.Factor_Type;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import javax.swing.text.html.HTMLDocument;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

 class ForwardingSet<E> implements  Set<E>{
    private final Set<E> s;
    public ForwardingSet(Set<E> s){this.s=s;}

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

    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);
    }

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

    @Override
    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();
    }

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

 interface  SetObserver<E>{

    void added(ObservableSet<E> set,E element);
}

 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);
        }
    }
    public void notifyElementAdded(E element){
        synchronized (observers){
            for(SetObserver<E> observer:observers){
                observer.added(this,element);
            }
        }
    }

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

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

@SpringBootApplication
public class RiskpriceApplication {



    public static void main(String[] args)  throws  InterruptedException{

        ObservableSet<Integer> set=new ObservableSet<Integer>(new HashSet<Integer>());
        set.addObserver(new SetObserver<Integer>() {
            @Override
            public void added(ObservableSet<Integer> s, Integer e) {
                System.out.println(e);
                if(e==23){
                    ExecutorService excutor= Executors.newSingleThreadExecutor();
                    final SetObserver<Integer> observer=this;
                    try{
                        excutor.submit(new Runnable() {
                            @Override
                            public void run() {
                                s.removeObserver(observer);
                            }
                        }).get();

                    }catch (ExecutionException ex){
                        throw new AssertionError(ex.getCause());
                    }catch (InterruptedException ex){
                        throw new AssertionError(ex.getCause());
                    }finally {
                        excutor.shutdown();
                    }
                }
            }
        });
        for(int i=0;i<100;i++){
            set.add(i);
        }
    }
}
第二版虽然会打印到23但是实际上并没有成功，
 public void run() {
                                s.removeObserver(observer);
                            }
进入
    public boolean removeObserver(SetObserver<E> observer){
        synchronized (observers){
            return  observers.remove(observer);
        }

    }
经过同步快synchronized 的时候将会遭遇死锁，因为主线程已经锁定了observers，只有等待子线程执行完成后才会释放锁，而子线程又在等待锁的释放，这样相互的等待就造成了死锁，但是由于Java设计的锁是可重入的，这种调用不会产生死锁，但会产生一个异常，因为调用线程正在该锁所保护的线程上进行着。这种失败可能是灾难性的，本质来说这个锁，没有尽到它的职责。可重入的锁简化了多线程的面向对象程序构造，但是它可能会将活性失败，变成安全性失败（参考自Effective java）
什么解决呢，来个2.1版本吧
我们建立个快照，而不使用原observers，这样每个通知都使用了自己的快照观察者列表引用就不会死锁了
 public void notifyElementAdded(E element){
        List<SetObserver<E>> snaphot=null;//快照
        synchronized (observers){
            snaphot=new ArrayList<SetObserver<E>>(observers);
        }
        for(SetObserver<E> observer:snaphot){
            observer.added(this,element);
        }
    }

//第三版 
事实上，要将外来方法的调用移出同步代码块还有更好的方法，从java1.5发行版以来，提供了并发集合 corrent collection ，称作 CopyOnWriteArrayList,
这是专门为此定制的，他是Arraylist的一种变体，通过重新拷贝整个底层数组，在这里实现所有的操作，由于内部数组永远不动（归功于重新拷贝），因此迭代不需要锁定，大量使用有性能影响，但对于观察者列表几乎不变来说却是很好的，因为他们几乎不改动，并且经常遍历
第三版较之前2.1版本更改如下：
  private final  List<SetObserver<E>> observers=new CopyOnWriteArrayList<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);
        //}

    }
    public void notifyElementAdded(E element){
        //List<SetObserver<E>> snaphot=null;//快照
        //synchronized (observers){
          //  snaphot=new ArrayList<SetObserver<E>>(observers);
        //}
        for(SetObserver<E> observer:observers){
            observer.added(this,element);
        }
    }

当然这个方法也可以改了，因为实际操作的时候底层是重新拷贝，所以也就不需要通过另外一个线程去移除引用了 修改如下：
 set.addObserver(new SetObserver<Integer>() {
            @Override
            public void added(ObservableSet<Integer> s, Integer e) {
                System.out.println(e);
                if(e==23){
                    s.removeObserver(this);
//                    ExecutorService excutor= Executors.newSingleThreadExecutor();
//                    final SetObserver<Integer> observer=this;
//                    try{
//                        excutor.submit(new Runnable() {
//                            @Override
//                            public void run() {
//                                s.removeObserver(observer);
//                            }
//                        }).get();
//
//                    }catch (ExecutionException ex){
//                        throw new AssertionError(ex.getCause());
//                    }catch (InterruptedException ex){
//                        throw new AssertionError(ex.getCause());
//                    }finally {
//                        excutor.shutdown();
//                    }
                }
            }
        });