多線程安全的案例展示與解決方案

2023-06-03 12:25:04 來源：博客園

一、概念

1. 什么是線程安全

當(dāng)多個(gè)線程訪問一個(gè)對(duì)象時(shí)，如果不用考慮這些線程在運(yùn)行時(shí)環(huán)境下的調(diào)度和交替執(zhí)行，也不需要進(jìn)行額外的同步，或者在調(diào)用方進(jìn)行任何其他的協(xié)調(diào)操作，調(diào)用這個(gè)對(duì)象的行為都可以獲得正確的結(jié)果，那這個(gè)對(duì)象是線程安全的。

通俗來說就是：不管業(yè)務(wù)中遇到怎么的多個(gè)線程訪問某個(gè)對(duì)象或某個(gè)方法的情況，而在編寫這個(gè)業(yè)務(wù)邏輯的時(shí)候，都不需要額外做任何額外的處理（也就是可以像單線程編程一樣），程序也可以正常運(yùn)行（不會(huì)因?yàn)槎嗑€程而出錯(cuò)），就可以稱為線程安全。

2. 什么是線程不安全：

多個(gè)線程同時(shí)做一個(gè)操作時(shí)，如使用 set 設(shè)置一個(gè)對(duì)象的值時(shí)，如果同時(shí)有另一個(gè)線程使用 get 方法取該對(duì)象的值，就有可能取到不正確的值，這種情況就需要我們進(jìn)行額外的操作保證結(jié)果正確，如Synchronized關(guān)鍵詞修飾做同步

(相關(guān)資料圖)

3. 那為什么不全部設(shè)計(jì)為線程安全：

主要考慮到運(yùn)行速度、設(shè)計(jì)成本等因素。

二、出現(xiàn)線程安全的案例

什么情況下會(huì)出現(xiàn)線程安全問題，怎么避免？

運(yùn)行結(jié)果錯(cuò)誤：a++多線程下出現(xiàn)消失的請求現(xiàn)象。
線程的活躍性問題：死鎖、活鎖、饑餓
對(duì)象發(fā)布和初始化的時(shí)候的安全問題

1. a++ 在多線程下出現(xiàn)的數(shù)據(jù)運(yùn)算出錯(cuò)

代碼演示：

public class MultiThreadsError implements Runnable {    static MultiThreadsError multiThreadsError = new MultiThreadsError();    int index = 0;    public static void main(String[] args) throws InterruptedException {        Thread thread1 = new Thread(multiThreadsError);        Thread thread2 = new Thread(multiThreadsError);        thread1.start();        thread2.start();        thread1.join();        thread2.join();        System.out.println(multiThreadsError.index);    }    @Override    public void run() {        for (int i = 0; i < 10000; i++) {            index++;        }    }}

結(jié)果如下：

兩個(gè)線換各執(zhí)行 10000 次 +1，結(jié)果卻不是20000，運(yùn)行過程中出現(xiàn)了什么問題呢？分析如下：

如上圖，對(duì)于i++這一步，兩個(gè)線程之間會(huì)獲取 i的值然后去執(zhí)行 ++，比如線程1 拿到i時(shí)i=1，在執(zhí)行 i+1時(shí)，線程2又拿到了i,這時(shí)依然是i=1（因?yàn)榫€程1還沒有運(yùn)算結(jié)束），線程2也進(jìn)行i+1,這就導(dǎo)致兩個(gè)線程都計(jì)算的結(jié)果都是2，然后都給i賦值，最終i=2,但其實(shí)是i+1執(zhí)行了兩次，結(jié)果應(yīng)該是i=3，運(yùn)算結(jié)果不符合預(yù)期；結(jié)果也就導(dǎo)致上面代碼中的打印結(jié)果。

那么 a++ 具體在哪里沖突了？又沖突了幾次？？我們可以嘗試作如下改進(jìn)，打印出出錯(cuò)的地方

public class MultiThreadsError implements Runnable {    static MultiThreadsError multiThreadsError = new MultiThreadsError();    // 通過一個(gè)boolean數(shù)組來標(biāo)記已經(jīng)++了的下標(biāo)為true值，若如果已經(jīng)為true，那么打印下標(biāo)為越界線程錯(cuò)誤的話；    final boolean[] marked = new boolean[100000];    int index = 0;    // AtomicInteger為細(xì)化步驟，就會(huì)讓這次操作不會(huì)出現(xiàn)線程不安全操作，這里用他記錄錯(cuò)誤/正確次數(shù)    static AtomicInteger realInt = new AtomicInteger();//正確次數(shù)    static AtomicInteger errorInt = new AtomicInteger();//錯(cuò)誤次數(shù)        //有參構(gòu)造參數(shù)為2，代表需要等待兩個(gè)線程經(jīng)過，再放行    static volatile CyclicBarrier cyclicBarrier1 = new CyclicBarrier(2);    static volatile CyclicBarrier cyclicBarrier2 = new CyclicBarrier(2);        public static void main(String[] args) throws InterruptedException {        Thread thread1 = new Thread(multiThreadsError);        Thread thread2 = new Thread(multiThreadsError);        thread1.start();        thread2.start();        thread1.join();        thread2.join();        System.out.println(multiThreadsError.index);        System.out.println("正確次數(shù)為："+realInt);        System.out.println("錯(cuò)誤次數(shù)為："+errorInt);    }    @Override    public void run() {        marked[0] = true;        for (int i = 0; i < 10000; i++) {            //設(shè)置柵欄            try {                cyclicBarrier2.reset();                cyclicBarrier1.await();            } catch (InterruptedException e) {                e.printStackTrace();            } catch (BrokenBarrierException e) {                e.printStackTrace();            }            index++;            try {                cyclicBarrier1.reset();                cyclicBarrier2.await();            } catch (InterruptedException e) {                e.printStackTrace();            } catch (BrokenBarrierException e) {                e.printStackTrace();            }            // 原子整數(shù)，是線程安全的，用來計(jì)數(shù) ++ 執(zhí)行的次數(shù)            realInt.incrementAndGet();            synchronized (multiThreadsError){                // 如果已經(jīng)為true，說明此時(shí)的index所在的位置已經(jīng)執(zhí)行過一次 ++ 了，那么打印下標(biāo)為越界線程錯(cuò)誤的話；                // if (marked[index]){                if (marked[index] && marked[index-1]){                    // 原子整數(shù)，是線程安全的，用來失敗的次數(shù)                    errorInt.incrementAndGet();                    System.out.println("該下標(biāo)【"+index+"】越界被標(biāo)記過了");                }                marked[index] = true;            }        }    }}

（1）定義 boolean[] marked 數(shù)組和 marked[index] = true 的作用？

通過一個(gè)boolean數(shù)組來標(biāo)記已經(jīng)++了的下標(biāo)為true值，若如果已經(jīng)為true，那么說明此時(shí)的 index 的當(dāng)前值已經(jīng)執(zhí)行過一次 ++ 了，那就說明遇到了 index++ 沖突；

(1) 這里使用到了 CyclicBarrier 類，原因如下：

假設(shè)i=4時(shí)，線程1和2遇到?jīng)_突，計(jì)算之后都給i賦值了5，本該是一次錯(cuò)誤運(yùn)算，但是如果遇到下面的情景會(huì)將其認(rèn)為是正確運(yùn)算：此時(shí)的i=5,線程1先拿到鎖后設(shè)置 marked[5]=true，然后線程1釋放鎖，此時(shí)線程1執(zhí)行特別快，在線程2拿到鎖時(shí)，線程1已經(jīng)又執(zhí)行了一次index++，此時(shí)的 index=6，線程2運(yùn)行 if (marked[index])時(shí)，即 marked[6]明顯還沒設(shè)置，就不會(huì)認(rèn)為是已經(jīng)失敗了。所以為了避免這種場景，保證線程1和2在交換鎖期間，兩個(gè)線程都只有一次 index++ 運(yùn)算，就用到了 CyclicBarrier 類。

(2) 上面代碼中為什么使用 `if (marked[index] && marked[index-1])`，而不是使用`if (marked[index])`,來作為失敗運(yùn)算的標(biāo)記呢？

因?yàn)槿绻麅蓚€(gè)線程的 index++ 如果沒有沖突的話，上個(gè)循環(huán)中的 index，和本次循環(huán)中的 index 應(yīng)該是相差2，也就表示中間會(huì)少設(shè)置一個(gè) marked[index]，但是如果 marked[index-1] 已經(jīng)被設(shè)置了，那就說明本次循環(huán)，兩個(gè)線程的 index++ 沖突了，但是有一個(gè)特殊 marked[0]，該值無論成功與否，都不會(huì)設(shè)置，所以需要在 run() 方法開頭加上 marked[0] = true;

打印結(jié)果如下：

可以看到錯(cuò)誤的次數(shù)和表面上結(jié)果相加剛好是20000，同時(shí)也打印了發(fā)生錯(cuò)誤的位置是19143，從而更清晰地知道哪里發(fā)生了 index++ 沖突。

2. 線程的`活躍性`問題：死鎖、活鎖、饑餓

這里以死鎖為例，代碼展示如下：

public class MultiThreadError implements Runnable {    int flag = 1;    static Object o1 = new Object();    static Object o2 = new Object();    public static void main(String[] args) {        MultiThreadError r1 = new MultiThreadError();        MultiThreadError r2 = new MultiThreadError();        r1.flag=1;        r2.flag=0;        Thread thread1 = new Thread(r1);        Thread thread2 = new Thread(r2);        thread1.start();        thread2.start();    }    @Override    public void run() {        System.out.println("flag： "+flag);        if (flag==1){            synchronized (o1){                try {                    Thread.sleep(500);                } catch (InterruptedException e) {                    e.printStackTrace();                }                synchronized (o2){                    System.out.println("1");                }            }        }        if (flag==0){            synchronized (o2){                try {                    Thread.sleep(500);                } catch (InterruptedException e) {                    e.printStackTrace();                }                synchronized (o1){                    System.out.println("1");                }            }        }    }}

結(jié)果如下：

在打印出上面兩行之后，便不會(huì)再進(jìn)行打印，而且程序也不會(huì)終止，這就是死鎖。兩個(gè)線程首先都各自持有一個(gè)鎖，然后去搶奪另一把鎖，但是要搶奪的鎖都已經(jīng)心有所屬，分別就是所屬于對(duì)方，于是兩個(gè)線程就一直干耗著，進(jìn)退兩難，成了死局。

3. 對(duì)象發(fā)布和初始化的時(shí)候的安全問題

什么是對(duì)象發(fā)布：

讓這個(gè)對(duì)象在超過這個(gè)類的范圍去使用。比如先使用 public 聲明這個(gè)類，那么這個(gè)類就是被發(fā)布出去了，那怎么超過這個(gè)類的范圍去使用呢，如下：

如果一個(gè)方法內(nèi) return 返回了一個(gè)對(duì)象的話，任何調(diào)用這個(gè)方法的類，都會(huì)獲取到這個(gè)對(duì)象
將某類的對(duì)象作為參數(shù)傳遞到其他類中，也是該類的對(duì)象脫離了本類，進(jìn)入其他對(duì)象中

什么是逸出：

某個(gè)被發(fā)布到不該發(fā)布的地方，比如：

方法返回一個(gè)private對(duì)象（private對(duì)象本身是不讓外部訪問）
還未完成初始化（構(gòu)造函數(shù)沒完全執(zhí)行完畢）就把對(duì)象提供給外界，比如以下幾種情況：
- 在構(gòu)造函數(shù)中未初始化完畢就給外部對(duì)象賦值this實(shí)例
- 隱式逸出——注冊監(jiān)聽事件
- 在構(gòu)造函數(shù)中運(yùn)行子線程

3.1 方法返回一個(gè)private對(duì)象

(1) 代碼展示：

public class MultiThreadError3 {    private Map states;    public MultiThreadError3(){        states=new HashMap<>();        states.put("1","周一");        states.put("2","周二");        states.put("3","周三");        states.put("4","周四");        states.put("5","周五");        states.put("6","周六");        states.put("7","周七");    }    //這里逸出了    public Map getStates(){        return states;    }    //導(dǎo)致下面可以獲取修改states對(duì)象的內(nèi)容    public static void main(String[] args) {        MultiThreadError3 multiThreadError3 = new MultiThreadError3();        Map states = multiThreadError3.getStates();        System.out.println(states.get("1"));        states.remove("1");        System.out.println(states.get("1"));    }}

打印結(jié)果如下：

states這個(gè)Map 對(duì)象本來是 MultiThreadError3類私有的，但是在 getStates()方法中被 return 出去了，那么外部就能拿到這個(gè)states ,而且甚至能對(duì)它進(jìn)行操作，修改里面的值，這就可能造成很嚴(yán)重的安全問題。

解決方案：

通過返回副本的方式，避免直接讓這個(gè)對(duì)象暴露給外界。

/** * 描述：     返回副本，解決逸出 */public class MultiThreadsError3 {    private Map states;    public MultiThreadsError3() {        states = new HashMap<>();        states.put("1", "周一");        states.put("2", "周二");        states.put("3", "周三");        states.put("4", "周四");    }    public Map getStates() {        return states;    }    public Map getStatesImproved() {        return new HashMap<>(states);    }    public static void main(String[] args) {        MultiThreadsError3 multiThreadsError3 = new MultiThreadsError3();        Map states = multiThreadsError3.getStates();//        System.out.println(states.get("1"));//        states.remove("1");//        System.out.println(states.get("1"));        System.out.println(multiThreadsError3.getStatesImproved().get("1"));        multiThreadsError3.getStatesImproved().remove("1");        System.out.println(multiThreadsError3.getStatesImproved().get("1"));    }}

打印結(jié)果：

3.2 還未完成初始化（構(gòu)造函數(shù)沒完全執(zhí)行完畢）就把this對(duì)象提供給外界

（1）代碼演示：在構(gòu)造函數(shù)中未初始化完畢就給外界對(duì)象賦值

public class MultiThreadsError4 {    static Point point;    public static void main(String[] args) throws InterruptedException {        new PointMaker().start();        Thread.sleep(10);        if (point != null) {            System.out.println(point);        }        Thread.sleep(105);        if (point != null) {            System.out.println(point);        }    }}class Point {    private final int x, y;    public Point(int x, int y) throws InterruptedException {        this.x = x;        MultiThreadsError4.point = this;        Thread.sleep(100);        this.y = y;    }    @Override    public String toString() {        return x + "," + y;    }}class PointMaker extends Thread {    @Override    public void run() {        try {            new Point(1, 1);        } catch (InterruptedException e) {            e.printStackTrace();        }    }}

打印結(jié)果：

因?yàn)閤的初始化比 y 要早一點(diǎn)，并且在構(gòu)造函數(shù)中有線程睡眠，就可能導(dǎo)致在 main 函數(shù)中不同的時(shí)間輸出的結(jié)果不一樣，比如上圖在main 函數(shù)中 Thread.sleep(10) 之后打印出的結(jié)果，和 Thread.sleep(105)之后打印出的結(jié)果不一樣

(2) 代碼演示：隱式逸出——注冊監(jiān)聽事件

/** * 觀察者模式 */public class MultiThreadsError5 {    private int count;    public MultiThreadsError5(MySource source) {        source.registerListener(new EventListener() {            @Override            public void onEvent(Event e) {                System.out.println("\n我得到的數(shù)字是" + count);            }        });        //模擬業(yè)務(wù)操作        for (int i = 0; i < 10000; i++) {            System.out.print(i);        }        count = 100;    }    public static void main(String[] args) {        MySource mySource = new MySource();        new Thread(() -> {            try {                Thread.sleep(10);            } catch (InterruptedException e) {                e.printStackTrace();            }            mySource.eventCome(new Event() {            });        }).start();        new MultiThreadsError5(mySource);    }    static class MySource {        private EventListener listener;        void registerListener(EventListener eventListener) {            this.listener = eventListener;        }        void eventCome(Event e) {            if (listener != null) {                listener.onEvent(e);            } else {                System.out.println("還未初始化完畢");            }        }    }    interface EventListener {        void onEvent(Event e);    }    interface Event {    }}

結(jié)果如下：

結(jié)果為什么是0而不是100呢？

在 new EventListener()這個(gè)匿名內(nèi)部類中，引用了外部類的 count變量，這個(gè)匿名內(nèi)部類就可以對(duì)它進(jìn)行操作，如果 count 的值在構(gòu)造函數(shù)中還沒有初始化完成，就對(duì)該 count 進(jìn)行操作，就導(dǎo)致count的值不準(zhǔn)確。、

解決方案：

使用工廠模式，將構(gòu)造器私有化不對(duì)外暴露，對(duì)外暴露一個(gè)方法：等做完所需的操作之后再 return 發(fā)布出去，就不會(huì)有實(shí)例過早被暴露的問題了。

/** * 描述：     用工廠模式修復(fù)剛才的初始化問題 */public class MultiThreadsError7 {    int count;    private EventListener listener;    private MultiThreadsError7(MySource source) {        listener = new EventListener() {            @Override            public void onEvent(MultiThreadsError5.Event e) {                System.out.println("\n我得到的數(shù)字是" + count);            }        };        for (int i = 0; i < 10000; i++) {            System.out.print(i);        }        count = 100;    }    public static MultiThreadsError7 getInstance(MySource source) {        MultiThreadsError7 safeListener = new MultiThreadsError7(source);        source.registerListener(safeListener.listener);        return safeListener;    }    public static void main(String[] args) {        MySource mySource = new MySource();        new Thread(new Runnable() {            @Override            public void run() {                try {                    Thread.sleep(10);                } catch (InterruptedException e) {                    e.printStackTrace();                }                mySource.eventCome(new MultiThreadsError5.Event() {                });            }        }).start();        MultiThreadsError7 multiThreadsError7 = new MultiThreadsError7(mySource);    }    static class MySource {        private EventListener listener;        void registerListener(EventListener eventListener) {            this.listener = eventListener;        }        void eventCome(MultiThreadsError5.Event e) {            if (listener != null) {                listener.onEvent(e);            } else {                System.out.println("還未初始化完畢");            }        }    }    interface EventListener {        void onEvent(MultiThreadsError5.Event e);    }    interface Event {    }}

打印結(jié)果：

(3) 在構(gòu)造函數(shù)中新建線程

/** *  構(gòu)造函數(shù)中新建線程 */public class MultiThreadError6 {    private Map states;    public MultiThreadError6(){        new Thread(new Runnable() {            @Override            public void run() {                states=new HashMap<>();                states.put("1","周一");                states.put("2","周二");                states.put("3","周三");                states.put("4","周四");                states.put("5","周五");                states.put("6","周六");                states.put("7","周七");            }        }).start();    }    public Map getStates(){        return states;    }    public static void main(String[] args) {        MultiThreadError6 multiThreadError6 = new MultiThreadError6();        System.out.println(multiThreadError6.states.get("1"));    }}

打印結(jié)果：

上圖所示：出現(xiàn)了空指針的情況，因?yàn)槌跏蓟牟僮髟诹硗庖粋€(gè)線程中，可能那個(gè)線程沒有執(zhí)行完畢，就會(huì)出現(xiàn)空指針，假如在 System.out.println(multiThreadError6.states.get("1"));之前加入 Thread.sleep(1000)休眠一段時(shí)間后等另外一個(gè)線程執(zhí)行完，就不會(huì)出現(xiàn)這個(gè)問題了。

三、總結(jié)

1. 各種需要考慮線程安全的情況，如下：

訪問共享的變量或資源，會(huì)有并發(fā)風(fēng)險(xiǎn)，這里的共享變量或資源指的是：對(duì)象的屬性，靜態(tài)變量，共享緩存，數(shù)據(jù)庫等等。
所有依賴時(shí)序的操作，即可以拆分成多個(gè)步驟的操作，即使每一步操作都是線程安全的，但是如果存在操作時(shí)序不對(duì)，還是存在并發(fā)問題，比如：read-modify-write（先讀取再修改最后寫入）、 check-then-act（先檢查再操作）
不同的數(shù)據(jù)之間存在捆綁關(guān)系的時(shí)候，那就要么把這些捆綁的數(shù)據(jù)全部修改，要么都不修改
在使用其他類的時(shí)候，如果該類沒有聲明自己是線程安全的，那就要注意該類可能是線程不安全的

2. 多線程除了安全問題，還可能會(huì)導(dǎo)致性能問題：

從某種程度上來講，多線程可以提高復(fù)雜的運(yùn)算效率，但是一定程度上多線程可能會(huì)帶來性能提交，比如多線程間的調(diào)度和協(xié)作帶來的性能開銷。

（1）調(diào)度：上下文切換

線程運(yùn)行個(gè)數(shù)超過CPU核心數(shù)的時(shí)候，CPU就需要對(duì)線程進(jìn)行調(diào)度，線程調(diào)度中就涉及線程切換，線程的切換的開銷是很大的，CPU需要保存當(dāng)前線程的運(yùn)行場景，將當(dāng)前線程的當(dāng)前運(yùn)行狀態(tài)保存好，為載入新的運(yùn)行線程做準(zhǔn)備。這樣來來回回其實(shí)是很耗費(fèi)性能的。而引起密集的上下文切換的操作就包括搶鎖和IO操作。