關于并發編程與線程安全的思考與實踐

來源| OSCHINA 社區

作者 | 京東云開發者-京東健康 張娜

一、并發編程的意義與挑戰

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

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

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

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

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

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

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

二、Java 并發編程

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

2.1 Java 內存模型

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

JMM 結構如下圖所示：


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

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

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

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

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

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

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

2.unlock：作用于主內存，解除獨占狀態。

3.read：作用主內存，把一個變量的值從主內存傳輸到線程的工作內存。

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

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

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

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

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

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

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

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

2.2 Java 中的并發關鍵字

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

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

2.3 Java 中的并發容器與工具類

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

通過循環 CAS 操作非阻塞的給隊列添加節點，

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

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

三、線上案例

3.1 問題發現

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

3.2 排查問題的詳細過程

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

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

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

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

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

醫生問診 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的值，和上一步是非原子操作，引發并發問題的根源
         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 創建 ArrayList ：

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

3.4 總結反思

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

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

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

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

審核編輯：湯梓紅