java 多執行緒同步觀察者併發集合的一個例子

阿新 • • 發佈：2018-12-22

//第一版
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();
//                    }
                }
            }
        });

java 多執行緒同步觀察者併發集合的一個例子

//第一版 package com.hra.riskprice; import com.hra.riskprice.SysEnum.Factor_Type; import org.springframework.boot.SpringApplication; import org.springfram

Java多執行緒阻塞佇列和併發集合

本章主要探討在多執行緒程式中與集合相關的內容。在多執行緒程式中，如果使用普通集合往往會造成資料錯誤，甚至造成程式崩潰。Java為多執行緒專門提供了特有的執行緒安全的集合類，通過下面的學習，您需要掌握這些集合的特點是什麼，底層實現如何、在何時使用等問題。 3.1Blockin

【轉】Java多執行緒-同步集合和併發集合

同步集合可以簡單地理解為通過synchronized來實現同步的集合。如果有多個執行緒呼叫同步集合的方法，它們將會序列執行。 arrayList和vector、stack Vector是執行緒安全的，原始碼中有很多的synchronized可以看出，而

java 多執行緒同步

執行緒屬性： 1.執行緒優先順序，　　優先順序高度依賴於系統，不要讓程式的正確性依賴於優先順序。預設情況子執行緒會繼承父執行緒的優先順序。 2.守護執行緒，　　唯一用途是為其他執行緒提供服務，當只剩

java多執行緒同步方式

1 同步程式碼塊自定義鎖 class Thread33 implements Runnable{ int a = 100; Object obj = new Object(); @Override public void run() { while(a > 0)

Java多執行緒同步---以銀行存取錢的過程的簡單程式碼例項

首先存錢取錢的這個操作，應該是執行緒操作的，可以有很多的顧客，這意思就是得有多個執行緒，多個執行緒之間共同操作一個銀行，銀行的金額就需要同步。才能保證執行緒安全。所以，下面就把這個程式碼的例項放這，有不對的地方，還請指出來哈。因為有個老鐵問這個多執行緒的程式碼。首先是

Java多執行緒學習---------超詳細總結(java 多執行緒同步資料傳遞 )

平時專案中多執行緒的應用比較少，今天網上找了找相關的內容學習了下。看到下面的文章感覺比較好。轉自：http://www.cnblogs.com/1020182600HENG/p/5939933.html 目錄一擴充套件javalangThread類二實現javalan

java多執行緒同步的例子

多執行緒訪問共享的資源物件,為了避免錯誤,java提供三種解決機制同步程式碼塊 synchronized code block同步方法 synchronized method同步鎖 ReentrantLockpackage com.linchengshen._01.threa

Java多執行緒同步和非同步詳解

1. 多執行緒併發時，多個執行緒同時請求同一資源，必然導致此資源的資料不安全。 2. 執行緒池在WEB服務中，對於web伺服器的響應速度必須儘可能的快，這就容不得在使用者提交請求按鈕後，再建立執行緒提供服務。為了減少使用者的等待時間，執行緒必須預先建立，放線上程池中，執行

Java多執行緒--同步與死鎖：synchronized；等待與喚醒：wait、notify、notifyAll；生命週期

class Info{ // 定義資訊類 private String name = "李興華"; // 定義name屬性 private String content = "JAVA講師" ; // 定義content屬性 private boolean flag = false ; // 設

Java多執行緒同步的五種方法

一、引言前幾天面試，被大師虐殘了，好多基礎知識必須得重新拿起來啊。閒話不多說，進入正題。二、為什麼要執行緒同步因為當我們有多個執行緒要同時訪問一個變數或物件時，如果這些執行緒中既有讀又有寫操作時，就會導致變數值或物件的狀態出現混亂，從而導致程式異常。舉個例

Java 多執行緒同步的五種方法 [轉]

程式碼很簡單，我就不解釋了，看看執行結果怎樣呢？截取了其中的一部分，是不是很亂，有寫看不懂。餘額不足賬戶餘額：0 餘額不足賬戶餘額：100 1441790503354存進：100 賬戶餘額：100 1441790504354存進：100 賬戶餘額：100 144179050

Java 多執行緒同步的五種方法

一、引言前幾天面試，被大師虐殘了，好多基礎知識必須得重新拿起來啊。閒話不多說，進入正題。二、為什麼要執行緒同步因為當我們有多個執行緒要同時訪問一個變數或物件時，如果這些執行緒中既有讀又有寫操作時，就會導致變數值或物件的狀態出現混亂，從而導致程式異常。舉個例子，

Java多執行緒同步鎖條件物件

每個程序都有它自己的變數的完備集，執行緒共享相同的資料。---這也是危險之處。執行緒建立和銷燬的開銷都比程序小。---這是好處。簡單的，你只要實現Runnable介面，實現過載run()方法，來做自己想做的事情。然後就是開啟一個執行緒，把你實現的類的物件傳入：Runnable

java多執行緒之物件的併發訪問

1.synchronized同步方法 --1.1方法內的變數是執行緒安全的解釋：由於方法內的變數是私有的，本體訪問的同時別人訪問不了，所以是執行緒安全的。 --1.2例項變數是非執行緒安全的解釋：

java多執行緒同步案列---按照固定格式輸出數字和字母

/*2. 三個執行緒 1個執行緒從1~26輸出 1個執行緒從a~z輸出 1個執行緒從A~Z輸出輸出格式為： 1 a A 2 b B 3 c C 4 d D ...26 z Z*/ //首先建立一個公用類PrintAll public class PrintAll {

Java多執行緒同步中同步程式碼塊、wait、notify與notifyAll的真正含義與工作原理

今天在和導師討論Java多執行緒程式設計的同步問題時，發現自己對同步程式碼塊、wait()方法、notify()方法和notifyAll()方法的理解不太清晰，於是在網上查閱資料，可是結果眾說紛紜，又在

Java多執行緒同步之Lock用法

從JDK5.0開始，有兩種機制來保護程式碼塊不受到並行訪問的干擾。舊版本的Java使用synchronized關鍵字來達到這個目的，而JDK5.0引進了ReentrantLock()類。synchronize關鍵字自動提供一個鎖和相關的條件。用Reentra

Java多執行緒同步 synchronized 用法

在使用MongoDB的時候 (基於spring-mongo) ,我想在插入物件時獲取有序自增的主鍵 ,但是MongoDB的預設規則是生成一串無序 (大致有序) 的字串 .而Spring Data提供的主鍵生成方法也是隨機的 String/BigInteger

Java多執行緒同步的幾種方式

當多個執行緒同時訪問一個資源時，非常容易出現安全問題。因此需要採用同步機制來解決這種問題。java主要提供了3種實現同步機制的方法： 1）、synchronized關鍵字 synchronized有兩種用法（synchronized方法和synchronized塊） syn

java 多執行緒 同步 觀察者 併發集合的一個例子

相關推薦

java 多執行緒同步觀察者併發集合的一個例子