關于并發(fā)編程與線程安全的思考與實踐

來源| OSCHINA 社區(qū)

作者 | 京東云開發(fā)者-京東健康 張娜

一、并發(fā)編程的意義與挑戰(zhàn)

并發(fā)編程的意義是充分的利用處理器的每一個核，以達到最高的處理性能，可以讓程序運行的更快。而處理器也為了提高計算速率，作出了一系列優(yōu)化，比如：

1、硬件升級：為平衡 CPU 內高速存儲器和內存之間數量級的速率差，提升整體性能，引入了多級高速緩存的傳統硬件內存架構來解決，帶來的問題是，數據同時存在于高速緩存和主內存中，需要解決緩存一致性問題。

2、處理器優(yōu)化：主要包含，編譯器重排序、指令級重排序、內存系統重排序。通過單線程語義、指令級并行重疊執(zhí)行、緩存區(qū)加載存儲 3 種級別的重排序，減少執(zhí)行指令，從而提高整體運行速度。帶來的問題是，多線程環(huán)境里，編譯器和 CPU 指令無法識別多個線程之間存在的數據依賴性，影響程序執(zhí)行結果。

并發(fā)編程的好處是巨大的，然而要編寫一個線程安全并且執(zhí)行高效的代碼，需要管理可變共享狀態(tài)的操作訪問，考慮內存一致性、處理器優(yōu)化、指令重排序問題。比如我們使用多線程對同一個對象的值進行操作時會出現值被更改、值不同步的情況，得到的結果和理論值可能會天差地別，此時該對象就不是線程安全的。而當多個線程訪問某個數據時，不管運行時環(huán)境采用何種調度方式或者這些線程如何交替執(zhí)行，這個計算邏輯始終都表現出正確的行為，那么稱這個對象是線程安全的。因此如何在并發(fā)編程中保證線程安全是一個容易忽略的問題，也是一個不小的挑戰(zhàn)。

所以，為什么會有線程安全的問題，首先要明白兩個關鍵問題：

1、線程之間是如何通信的，即線程之間以何種機制來交換信息。

2、線程之間是如何同步的，即程序如何控制不同線程間的發(fā)生順序。

二、Java 并發(fā)編程

Java 并發(fā)采用了共享內存模型，Java 線程之間的通信總是隱式進行的，整個通信過程對程序員完全透明。

2.1 Java 內存模型

為了平衡程序員對內存可見性盡可能高（對編譯器和處理的約束就多）和提高計算性能（盡可能少約束編譯器處理器）之間的關系，JAVA 定義了Java 內存模型（Java Memory Model，JMM），約定只要不改變程序執(zhí)行結果，編譯器和處理器怎么優(yōu)化都行。所以，JMM 主要解決的問題是，通過制定線程間通信規(guī)范，提供內存可見性保證。

JMM 結構如下圖所示：


以此看來，線程內創(chuàng)建的局部變量、方法定義參數等只在線程內使用不會有并發(fā)問題，對于共享變量，JMM 規(guī)定了一個線程如何和何時可以看到由其他線程修改過后的共享變量的值，以及在必須時如何同步的訪問共享變量。

為控制工作內存和主內存的交互，定義了以下規(guī)范：

?所有的變量都存儲在主內存 (Main Memory) 中。

?每個線程都有一個私有的本地內存 (Local Memory)，本地內存中存儲了該線程以讀 / 寫共享變量的拷貝副本。

?線程對變量的所有操作都必須在本地內存中進行，而不能直接讀寫主內存。
?不同的線程之間無法直接訪問對方本地內存中的變量。

具體實現上定義了八種操作：

1.lock：作用于主內存，把變量標識為線程獨占狀態(tài)。

2.unlock：作用于主內存，解除獨占狀態(tài)。

3.read：作用主內存，把一個變量的值從主內存?zhèn)鬏數骄€程的工作內存。

4.load：作用于工作內存，把 read 操作傳過來的變量值放入工作內存的變量副本中。

5.use：作用工作內存，把工作內存當中的一個變量值傳給執(zhí)行引擎。

6.assign：作用工作內存，把一個從執(zhí)行引擎接收到的值賦值給工作內存的變量。

7.store：作用于工作內存的變量，把工作內存的一個變量的值傳送到主內存中。

8.write：作用于主內存的變量，把 store 操作傳來的變量的值放入主內存的變量中。

這些操作都滿足以下原則：

?不允許 read 和 load、store 和 write 操作之一單獨出現。

?對一個變量執(zhí)行 unlock 操作之前，必須先把此變量同步到主內存中（執(zhí)行 store 和 write 操作）。

2.2 Java 中的并發(fā)關鍵字

Java 基于以上規(guī)則提供了 volatile、synchronized 等關鍵字來保證線程安全，基本原理是從限制處理器優(yōu)化和使用內存屏障兩方面解決并發(fā)問題。如果是變量級別，使用 volatile 聲明任何類型變量，同基本數據類型變量、引用類型變量一樣具備原子性；如果應用場景需要一個更大范圍的原子性保證，需要使用同步塊技術。Java 內存模型提供了 lock 和 unlock 操作來滿足這種需求。虛擬機提供了字節(jié)碼指令 monitorenter 和 monitorexist 來隱式地使用這兩個操作，這兩個字節(jié)碼指令反映到 Java 代碼中就是同步塊 - synchronized 關鍵字。

這兩個字的作用：volatile 僅保證對單個 volatile 變量的讀 / 寫具有原子性，而鎖的互斥執(zhí)行的特性可以確保整個臨界區(qū)代碼的執(zhí)行具有原子性。在功能上，鎖比 volatile 更強大，在可伸縮性和執(zhí)行性能上，volatile 更有優(yōu)勢。

2.3 Java 中的并發(fā)容器與工具類

2.3.1 CopyOnWriteArrayList

CopyOnWriteArrayList 在操作元素時會加可重入鎖，一次來保證寫操作是線程安全的，但是每次添加刪除元素就需要復制一份新數組，對空間有較大的浪費。

publicEget(int index){
        returnget(getArray(), index);
    }

    publicbooleanadd(E e){
        finalReentrantLock lock =this.lock;
        lock.lock();
        try{
            Object[] elements =getArray();
            int len = elements.length;
            Object[] newElements =Arrays.copyOf(elements, len +1);
            newElements[len]= e;
            setArray(newElements);
            returntrue;
        }finally{
            lock.unlock();
        }
    }

2.3.2 Collections.synchronizedList(new ArrayList<>());

這種方式是在 List 的操作外包加了一層 synchronize 同步控制。需要注意的是在遍歷 List 是還得再手動做整體的同步控制。

publicvoidadd(int index,E element){
        // SynchronizedList 就是在 List的操作外包加了一層synchronize同步控制
        synchronized(mutex){list.add(index, element);}
    }
    publicEremove(int index){
        synchronized(mutex){return list.remove(index);}
    }

2.3.3 ConcurrentLinkedQueue

通過循環(huán) CAS 操作非阻塞的給隊列添加節(jié)點，

publicbooleanoffer(E e){
        checkNotNull(e);
        finalNode newNode =newNode(e);

        for(Node t = tail, p = t;;){
            Node q = p.next;
            if(q ==null){
                // p是尾節(jié)點，CAS 將p的next指向newNode.
                if(p.casNext(null, newNode)){
                    if(p != t) 
                        //tail指向真正尾節(jié)點
                        casTail(t, newNode);
                    returntrue;
                }
            }
            elseif(p == q)
                // 說明p節(jié)點和p的next節(jié)點都等于空，表示這個隊列剛初始化，正準備添加節(jié)點，所以返回head節(jié)點
                p =(t !=(t = tail))? t : head;
            else
                // 向后查找尾節(jié)點
                p =(p != t && t !=(t = tail))? t : q;
        }
    }

三、線上案例

3.1 問題發(fā)現

在互聯網醫(yī)院醫(yī)生端，醫(yī)生打開問診 IM 聊天頁，需要加載幾十個功能按鈕。在 2022 年 12 月抗疫期間，QPS 全天都很高，高峰時是平日的 12 倍，偶現報警提示按鈕顯示不全，問題出現概率大概在百萬分之一。

3.2 排查問題的詳細過程

醫(yī)生問診 IM 頁面的加載屬于業(yè)務黃金流程，上面的每一個按鈕就是一個業(yè)務線的入口，所以處在核心邏輯的上的報警均使用自定義報警，該類報警不設置收斂，無論何種異常包括按鈕個數異常就會立即報警。

1. 根據報警信息，開始排查，卻發(fā)現以下問題：

（1）沒有異常日志：順著異常日志的 logId 排查，過程中竟然沒有異常日志，按鈕莫名其妙的變少了。

（2）不能復現：在預發(fā)環(huán)境，使用相同入參，接口正常返回，無法復現。

2. 代碼分析，縮小異常范圍：

醫(yī)生問診 IM 按鈕處理分組進行：

// 多個線程結果集合
    List multiButtonList =newArrayList<>();

// 多線程并行處理
    Future multiButtonFuture = joyThreadPoolTaskExecutor.submit(()->{
        List multiButtonListTemp =newArrayList<>();
        buttonTypes.forEach(buttonType ->{
            multiButtonListTemp.add(appButtonInfoMap.get(buttonType));
        });
        multiButtonList.addAll(multiButtonListTemp);
        return multiButtonListTemp;
    });

     publicbooleanaddAll(Collection  c){
         Object[] a = c.toArray();
         int numNew = a.length;
         ensureCapacityInternal(size + numNew);// Increments modCount
 
         //以當前size為起點，向數組中追加本次新增對象
         System.arraycopy(a,0, elementData, size, numNew);
 
         //更新全局變量size的值，和上一步是非原子操作，引發(fā)并發(fā)問題的根源
         size += numNew;
         return numNew !=0;
     }
 
     privatevoidensureCapacityInternal(int minCapacity){
         if(elementData == DEFAULTCAPACITY_EMPTY_ELEMENTDATA){
             minCapacity =Math.max(DEFAULT_CAPACITY, minCapacity);
         }
 
         ensureExplicitCapacity(minCapacity);
     }
 
     privatevoidensureExplicitCapacity(int minCapacity){
         modCount++;
 
         // overflow-conscious code
         if(minCapacity - elementData.length >0)
             grow(minCapacity);
     }
 
     privatevoidgrow(int minCapacity){
         // overflow-conscious code
         int oldCapacity = elementData.length;
         int newCapacity = oldCapacity +(oldCapacity >>1);
         if(newCapacity - minCapacity <0)
             newCapacity = minCapacity;
         if(newCapacity - MAX_ARRAY_SIZE >0)
             newCapacity =hugeCapacity(minCapacity);
         // minCapacity is usually close to size, so this is a win:
         elementData =Arrays.copyOf(elementData, newCapacity);
     }
 

publicbooleanaddAll(Collection  c){
         Object[] a = c.toArray();
         int numNew = a.length;
         //第1次停頓，獲取當前size
         try{
             Thread.sleep(1000*timeout1);
         }catch(InterruptedException e){
             e.printStackTrace();
         }
         ensureCapacityInternal(size + numNew);// Increments modCount
 
         //第2次停頓，等待copy
         try{
             Thread.sleep(1000*timeout2);
         }catch(InterruptedException e){
             e.printStackTrace();
         }
         System.arraycopy(a,0, elementData, size, numNew);
 
         //第3次停頓，等待size+=
         try{
             Thread.sleep(1000*timeout3);
         }catch(InterruptedException e){
             e.printStackTrace();
         }
         size += numNew;
         return numNew !=0;
     }

3.3 解決問題

使用線程安全工具 Collections.synchronizedList 創(chuàng)建 ArrayList ：

List multiButtonList =Collections.synchronizedList(newArrayList<>());

3.4 總結反思

使用多線程處理問題已經變得很普遍，但是對于多線程共同操作的對象必須使用線程安全的類。

另外，還要搞清楚幾個靈魂問題：

（1）JMM 的靈魂：Happens-before 原則

（2）并發(fā)工具類的靈魂：volatile 變量的讀 / 寫 和 CAS

審核編輯：湯梓紅