JAVA多線程線程安全性基礎(chǔ)

更新時(shí)間：2021年08月12日 15:10:03 作者：聞人此生

這篇文章主要介紹了如何測試Java類的線程安全性,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

線程安全性

一個(gè)對象是否需要是線程安全的，取決于它是否被多個(gè)線程訪問，而不取決于對象要實(shí)現(xiàn)的功能

什么是線程安全的代碼

核心：對共享的和可變的狀態(tài)的訪問進(jìn)行管理。防止對數(shù)據(jù)發(fā)生不受控的并發(fā)訪問。

何為對象的狀態(tài)？

狀態(tài)是指存儲在對象的狀態(tài)變量（例如實(shí)例或靜態(tài)域）中的數(shù)據(jù)。還可能包括其他依賴對象的域。

eg：某個(gè)HashMap的狀態(tài)不僅存儲在HashMap對象本身，還存儲在許多Map.Entry對象中。

在這里插入圖片描述

總而言之，在對象的狀態(tài)中包含了任何可能影響其外部可見行為的數(shù)據(jù)。

何為共享的？

共享的是指變量可同時(shí)被多個(gè)線程訪問

何為可變的？

可變的是指變量的值在其生命周期內(nèi)可以發(fā)生變化。試想，如果一個(gè)共享變量的值在其生命周期內(nèi)不會發(fā)生變化，那么在多個(gè)

線程訪問它的時(shí)候，就不會出現(xiàn)數(shù)據(jù)不一致的現(xiàn)象，自然就不存在線程安全性問題了。

什么是線程安全性

當(dāng)多個(gè)線程訪問某個(gè)類時(shí)，不管運(yùn)行時(shí)環(huán)境采用何種調(diào)度方式或者這些線程將如何交替執(zhí)行，并且在主調(diào)代碼中不需要任何額外的同步或協(xié)同，這個(gè)類都能表現(xiàn)出正確的行為，達(dá)到預(yù)期的效果，那么就稱這個(gè)類是線程安全的。

如下啟動10個(gè)線程，每個(gè)線程對inc執(zhí)行1000次遞增，并添加一個(gè)計(jì)時(shí)線程，預(yù)期效果應(yīng)為10000，而實(shí)際輸出值為6880，是一個(gè)小于10000的值，并未達(dá)到預(yù)期效果，因此INS類不是線程安全的，整個(gè)程序也不是線程安全的。原因是遞增操作不是原子操作，并且沒有適當(dāng)?shù)耐綑C(jī)制

package hgh0808;
public class Test {
    public static void main(String[] args){
        for(int i = 0;i < 10;i++){
            Thread th = new Thread(new CThread());
            th.start();
        }
        TimeThread tt = new TimeThread();
        tt.start();
        try{
            Thread.sleep(21000);
        }catch(Exception e){
            e.printStackTrace();
        }
        System.out.println(INS.inc);
    }
}
---------------------------------------------------------------------
package hgh0808;
import java.util.concurrent.atomic.*;
public class TimeThread extends Thread{
    @Override
    public void run(){
        int count = 1;
        for(int i = 0;i < 20;i++){
            try{
                Thread.sleep(1000);
            }catch(Exception e){
                e.printStackTrace();
            }
            System.out.println(count++);
        }
    }
}
---------------------------------------------------------------------
package hgh0808;
public class CThread implements Runnable{
    @Override
    public void run(){
        for(int j = 0;j < 1000;j++){
            INS.increase();
        }
    }
}
---------------------------------------------------------------------
package hgh0808;
public class INS{
    public static volatile int inc = 0;
    public static void increase(){
            inc++;
    }
}
=====================================================================

執(zhí)行結(jié)果：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
6880

通過synchronized加鎖機(jī)制，對INS類實(shí)現(xiàn)同步，如下得到了正確的運(yùn)行結(jié)果，很容易可以看出，主調(diào)代碼中并沒有任何額外的同步或協(xié)同，此時(shí)的INS類是線程安全的，整個(gè)程序也是線程安全的

package hgh0808;
public class INS{
    public static volatile int inc = 0;
    public static void increase(){
        synchronized (INS.class){
            inc++;
        }
    }
}

執(zhí)行結(jié)果：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
10000

如何編寫線程安全的代碼
------------------------------------------------------------------------------------------------
如果當(dāng)多個(gè)線程訪問同一個(gè)可變的狀態(tài)變量時(shí)沒有使用合適的同步，那么程序就會出現(xiàn)錯(cuò)誤，像上文中進(jìn)行同步之前的代碼
有三種方式可以修復(fù)這個(gè)問題：
*不在線程之間共享該狀態(tài)變量
*將狀態(tài)變量修改為不可變的變量
*在訪問狀態(tài)變量時(shí)使用同步
前兩種方法是針對共享和不變這兩個(gè)屬性（見上文）解決問題，在有些情境下會違背程序設(shè)計(jì)的初衷（比如上文中INS類中的inc變量不可能不變，且在多核處理器的環(huán)境下為了提高程序性能，就需要多個(gè)線程同時(shí)處理，這樣變量就必然要被多個(gè)線程共享）。
基于此，我們針對第三種方式------ 在訪問狀態(tài)變量時(shí)使用同步展開討論
在討論第三種方式之前，我們先介紹幾個(gè)簡單的概念
原子性：一個(gè)操作序列的所有操作要么不間斷地全部被執(zhí)行，要么一個(gè)也沒有執(zhí)行
競態(tài)條件：當(dāng)某個(gè)計(jì)算的正確性取決于多個(gè)線程的的交替執(zhí)行時(shí)序時(shí)，就會發(fā)生競態(tài)條件。通俗的說，就是某個(gè)程序結(jié)果的正確性取決于運(yùn)氣時(shí)，就會發(fā)生競態(tài)條件。（競態(tài)條件并不總是會產(chǎn)生錯(cuò)誤，還需要某種不恰當(dāng)?shù)膱?zhí)行時(shí)序）
常見的競態(tài)條件類型：
*檢查–執(zhí)行（例如延遲初始化）
*讀取–修改–寫入（例如自增++操作）
針對以上兩種常見的競態(tài)條件類型，我們分別給出例子

延遲初始化（檢查--執(zhí)行）
--------------------------------------------------------------------
package hgh0808;
import java.util.ArrayList;
public class Test1 {
    public ArrayList<Ball> list;
    public ArrayList<Ball> getInstance(){
        if(list == null){
            list = new ArrayList<Ball>();
        }
        return list;
    }
}
class Ball{
}

大概邏輯是先判斷l(xiāng)ist是否為空，若為空，創(chuàng)建一個(gè)新的ArrayList對象，若不為空，則直接使用已存在的ArrayList對象，這樣可以保證在整個(gè)項(xiàng)目中l(wèi)ist始終指向同一個(gè)對象。這在單線程環(huán)境中是完全沒有問題的，但是如果在多線程環(huán)境中，list還未實(shí)例化時(shí)，A線程和B線程同時(shí)執(zhí)行if語句，A和B線程都會認(rèn)為list為null，A和B線程都會執(zhí)行實(shí)例化語句，造成混亂。

自增++操作（讀取--修改--寫入）
------------------------------------------------------------------------
參考上文中為改進(jìn)之前的代碼（對INS類中inc的自增）

以上兩個(gè)例子告訴我們，必須添加適當(dāng)?shù)耐讲呗?，保證復(fù)合操作的原子性，防止競態(tài)條件的出現(xiàn)

策略一：使用原子變量類,在java.util.concurrent.atomic包中包含了一些原子變量類

在這里插入圖片描述

package hgh0808;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public class INS{
    public static AtomicInteger inc = new AtomicInteger(0);
    public static void increase(){
        inc.incrementAndGet();
    }
}

執(zhí)行結(jié)果：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
10000

值得注意的是，只有一個(gè)狀態(tài)變量時(shí)，可以通過原子變量類實(shí)現(xiàn)線程安全。但是如果有多個(gè)狀態(tài)變量呢？

設(shè)想一個(gè)情景

多個(gè)線程不斷產(chǎn)生1到10000的隨機(jī)數(shù)并且發(fā)送到一個(gè)計(jì)算線程，計(jì)算線程每獲取一個(gè)數(shù)字n，就計(jì)算sinx在[0,n]上的積分并打印到控制臺上，為了提高程序性能，設(shè)計(jì)一個(gè)緩存機(jī)制，保存上次的數(shù)字n和積分結(jié)果（兩個(gè)狀態(tài)變量）。如果本次的數(shù)字和上次的數(shù)字相等，直接打印積分結(jié)果，避免重復(fù)計(jì)算。

看代碼：

package hgh0808;
import java.util.concurrent.atomic.AtomicReference;

public class Calculate extends Thread{
    private final AtomicReference<Double> lastNumber  = new AtomicReference<Double>();  //緩存機(jī)制，原子變量類
    private final AtomicReference<Double> lastRes = new AtomicReference<Double>();      //緩存機(jī)制，原子變量類
    private static final double N = 100000;    //將區(qū)間[0,e]分成100000份，方便定積分運(yùn)算
    public void service() throws Exception{
        getData();
        Thread.sleep(10000);   //等待MyQueue隊(duì)列中有一定數(shù)量的元素后，再開始從其中取元素
        while(true){
            Double e;
                if(!MyQueue.myIsEmpty()){
                     e = MyQueue.myRemove();
                }else{
                    return;
                }
            if(e.equals(lastNumber.get())){
                System.out.println(lastNumber.get()+" "+lastRes.get());
            }else{
                Double temp = integral(e);
                lastNumber.set(e);
                lastRes.set(temp);
                System.out.println(e+" "+temp);
            }
            Thread.sleep(2000);
        }
    }
    public void getData(){   //創(chuàng)建并啟動四個(gè)獲取隨機(jī)數(shù)的線程，這四個(gè)線程交替向MyQueue隊(duì)列中添加元素
        Thread1 th1 = new Thread1();
        Thread2 th2 = new Thread2();
        Thread3 th3 = new Thread3();
        Thread4 th4 = new Thread4();
        th1.start();
        th2.start();
        th3.start();
        th4.start();
    }
    public Double integral(double e){    //計(jì)算定積分
        double step = (e-0)/N;
        double left = 0,right = step;
        double sum = 0;
        while(right <= e){
            double mid = left+(right-left)/2;
            sum+=Math.sin(mid);
            left+=step;
            right+=step;
        }
        sum*=step;
        return sum;
    }
}
---------------------------------------------------------------------
package hgh0808;
import java.util.LinkedList;
public class MyQueue {      //由于LinkedList是線程不安全的，因此需要將其改寫為線程安全類
    private static LinkedList<Double> queue = new LinkedList<>();
    synchronized public static void myAdd(Double e){
        queue.addLast(e);
    }
    synchronized public static void myClear(){
        queue.clear();
    }
    synchronized public static int mySize(){
        return queue.size();
    }
    synchronized public static boolean myIsEmpty(){
        return queue.isEmpty();
    }
    synchronized public static double myRemove(){
        return queue.removeFirst();
    }
}
-----------------------------------------------------------------------
package hgh0808;
import java.util.Random;
public class Thread1 extends Thread{
    private double data;
    @Override
    public void run(){
        while(true){
            Random random = new Random();
            data = (double) (random.nextInt(10000)+1);
            if(MyQueue.mySize() > 10000){     //由于從隊(duì)列中取元素的速度低于四個(gè)線程向隊(duì)列中加元素的速度，因此隊(duì)列的長度是趨于擴(kuò)張的，當(dāng)達(dá)到一定程度時(shí)，清空隊(duì)列
                MyQueue.myClear();
            }
            MyQueue.myAdd(data);
            try {
                Thread.sleep(1000);
            }catch(Exception e){
                e.printStackTrace();
            }
        }
    }
}
------------------------------------------------------------------------
package hgh0808;
import java.util.Random;
public class Thread2 extends Thread{
    private double data;
    @Override
    public void run(){
        while(true){
            Random random = new Random();
            data = (double) (random.nextInt(10000)+1);
            if(MyQueue.mySize() > 10000){
                MyQueue.myClear();
            }
            MyQueue.myAdd(data);
            try {
                Thread.sleep(1000);
            }catch(Exception e){
                e.printStackTrace();
            }
        }
    }
}
-----------------------------------------------------------------------
package hgh0808;
import java.util.Random;
public class Thread3 extends Thread{
    private double data;
    @Override
    public void run(){
        while(true){
            Random random = new Random();
            data = (double) (random.nextInt(10000)+1);
            if(MyQueue.mySize() > 10000){
                MyQueue.myClear();
            }
            MyQueue.myAdd(data);
            try {
                Thread.sleep(1000);
            }catch(Exception e){
                e.printStackTrace();
            }
        }
    }
}
------------------------------------------------------------------------
package hgh0808;
import java.util.Random;
public class Thread4 extends Thread{
    private double data;
    @Override
    public void run(){
        while(true){
            Random random = new Random();
            data = (double) (random.nextInt(10000)+1);
            if(MyQueue.mySize() > 10000){
                MyQueue.myClear();
            }
            MyQueue.myAdd(data);
            try {
                Thread.sleep(1000);
            }catch(Exception e){
                e.printStackTrace();
            }
        }
    }
}

只看Calculate線程，不看其他線程和MyQueue中的鎖機(jī)制，本問題的焦點(diǎn)在于Calculate線程中對多個(gè)狀態(tài)變量的同步策略

存在問題：

盡管對lastNumber和lastRes的set方法的每次調(diào)用都是原子的，但仍然無法同時(shí)更新lastNumber和lastRes；如果只修改了其中一個(gè)變量，那么在這兩次修改操作之間，其它線程將發(fā)現(xiàn)不變性條件被破壞了。換句話說，就是沒有足夠的原子性

**當(dāng)在不變性條件中涉及多個(gè)變量時(shí)，各個(gè)變量間并不是彼此獨(dú)立的，而是某個(gè)變量的值會對其它變量的值產(chǎn)生約束。因此當(dāng)更新某一個(gè)變量時(shí)，需要在同一個(gè)原子操作中對其他變量同時(shí)進(jìn)行更新。

改進(jìn) ================>加鎖機(jī)制內(nèi)置鎖 synchronized

之所以每個(gè)對象都有一個(gè)內(nèi)置鎖，只是為了免去顯式地創(chuàng)建鎖對象

synchronized修飾方法就是橫跨整個(gè)方法體的同步代碼塊

非靜態(tài)方法的鎖-----方法調(diào)用所在的對象

靜態(tài)方法的鎖-----方法所在類的class對象

public class Calculate extends Thread{
    private final AtomicReference<Double> lastNumber  = new AtomicReference<Double>();  //緩存機(jī)制，原子變量類
    private final AtomicReference<Double> lastRes = new AtomicReference<Double>();      //緩存機(jī)制，原子變量類
    private static final double N = 100000;    //將區(qū)間[0,e]分成100000份，方便定積分運(yùn)算
    public void service() throws Exception{
        getData();
        Thread.sleep(10000);   //等待MyQueue隊(duì)列中有一定數(shù)量的元素后，再開始從其中取元素
        while(true){
            Double e;
            synchronized (this){    //檢查--執(zhí)行 使用synchronized同步，防止出現(xiàn)競態(tài)條件
                if(!MyQueue.myIsEmpty()){
                     e = MyQueue.myRemove();
                }else{
                    return;
                }
            }
            if(e.equals(lastNumber.get())){
                System.out.println(lastNumber.get()+" "+lastRes.get());
            }else{
                Double temp = integral(e);
                synchronized (this) {     //兩個(gè)狀態(tài)變量在同一個(gè)原子操作中更新
                    lastNumber.set(e);
                    lastRes.set(temp);
                }
                System.out.println(e+" "+temp);
            }
            Thread.sleep(2000);
        }
    }
    public void getData(){   //創(chuàng)建并啟動四個(gè)獲取隨機(jī)數(shù)的線程，這四個(gè)線程交替向MyQueue隊(duì)列中添加元素
        Thread1 th1 = new Thread1();
        Thread2 th2 = new Thread2();
        Thread3 th3 = new Thread3();
        Thread4 th4 = new Thread4();
        th1.start();
        th2.start();
        th3.start();
        th4.start();
    }
    public Double integral(double e){    //計(jì)算定積分
        double step = (e-0)/N;
        double left = 0,right = step;
        double sum = 0;
        while(right <= e){
            double mid = left+(right-left)/2;
            sum+=Math.sin(mid);
            left+=step;
            right+=step;
        }
        sum*=step;
        return sum;
    }
}

對于包含多個(gè)變量的不變性條件中，其中涉及的所有變量都需要由同一個(gè)鎖來保護(hù)

synchronized (this) {     //兩個(gè)狀態(tài)變量在同一個(gè)原子操作中更新
                    lastNumber.set(e);
                    lastRes.set(temp);
                }

鎖的重入

如果某個(gè)線程試圖獲得一個(gè)已經(jīng)由它自己持有的鎖，那么這個(gè)請求就會成功，“重入”意味著獲取鎖的操作的粒度是‘線程'，而不是‘調(diào)用'。

重入的一種實(shí)現(xiàn)方式：

為每個(gè)鎖關(guān)聯(lián)一個(gè)獲取計(jì)數(shù)值和一個(gè)所有者線程。當(dāng)計(jì)數(shù)值為0時(shí)，這個(gè)鎖就被認(rèn)為是沒有被任何線程持有。當(dāng)線程請求一個(gè)未被持有的鎖時(shí)，JVM將記下鎖的持有者，并且將獲取計(jì)數(shù)值置為1。如果同一個(gè)線程再次獲取這個(gè)鎖，計(jì)數(shù)值將遞增，當(dāng)線程退出同步代碼塊時(shí)，計(jì)數(shù)器會相應(yīng)地遞減。當(dāng)計(jì)數(shù)值為0時(shí)，這個(gè)鎖將被釋放。

如果內(nèi)置鎖不可重入，那么以下這段代碼將發(fā)生死鎖(每個(gè)doSomething方法在執(zhí)行前都會獲取Father上的內(nèi)置鎖)
----------------------------------------------------------------------
public class Father{
  public synchronized void doSomething(){
  }
}

public class Son extends Father{
   @Override
   public synchronized void doSomething(){
       System.out.println("重寫");
       super.doSomething();
   }
}

線程安全性與性能和活躍性之間的平衡

活躍性：是否會發(fā)生死鎖饑餓等現(xiàn)象
性能：線程的并發(fā)度
不良并發(fā)的應(yīng)用程序：可同時(shí)調(diào)用的線程數(shù)量，不僅受到可用處理資源的限制，還受到應(yīng)用程序本身結(jié)構(gòu)的限制。幸運(yùn)的是，通過縮小同步代碼塊的作用范圍，可以平衡這個(gè)問題。
縮小作用范圍的原則====>當(dāng)執(zhí)行時(shí)間較長的計(jì)算或者可能無法快速完成的操作時(shí)，一定不能持有鎖?。。?/p>