Java並發編程系列-(7) Java線程安全

目前已經更新完《Java並發編程》,《Docker教程》和《JVM性能優化》，歡迎關注【后端精進之路】，輕松閱讀全部文章。

Java並發編程：

• Java並發編程系列-(1) 並發編程基礎
• Java並發編程系列-(2) 線程的並發工具類
• Java並發編程系列-(3) 原子操作與CAS
• Java並發編程系列-(4) 顯式鎖與AQS
• Java並發編程系列-(5) Java並發容器
• Java並發編程系列-(6) Java線程池
• Java並發編程系列-(7) Java線程安全
• Java並發編程系列-(8) JMM和底層實現原理
• Java並發編程系列-(9) JDK 8/9/10中的並發

Docker教程:

• Docker系列-(1) 原理與基本操作
• Docker系列-(2) 鏡像制作與發布
• Docker系列-(3) Docker-compose使用與負載均衡

JVM性能優化:

• JVM性能優化系列-(1) Java內存區域
• JVM性能優化系列-(2) 垃圾收集器與內存分配策略
• JVM性能優化系列-(3) 虛擬機執行子系統
• JVM性能優化系列-(4) 編寫高效Java程序
• JVM性能優化系列-(5) 早期編譯優化
• JVM性能優化系列-(6) 晚期編譯優化
• JVM性能優化系列-(7) 深入了解性能優化

7. 線程安全

7.1 線程安全的定義

如果多線程下使用這個類，不過多線程如何使用和調度這個類，這個類總是表示出正確的行為，這個類就是線程安全的。

類的線程安全表現為：

• 操作的原子性
• 內存的可見性

不做正確的同步，在多個線程之間共享狀態的時候，就會出現線程不安全。

7.2 如何保證線程安全

棧封閉

所有的變量都是在方法內部聲明的，這些變量都處於棧封閉狀態。

比如下面的例子,a和b都是在方法內部定義的，無法被外部線程所訪問，當方法結束后，棧內存被回收，所以是線程安全的。

void fun（）{
    int a = 1；
    int b= 2；
    // do something
}

無狀態

沒有任何成員變量的類，就叫無狀態的類，這種類不存在共享的資源，顯然是安全的。

public class StatelessClass {
	
	public int service(int a,int b) {
		return a*b;
	}
}

不可變的類

讓狀態不可變，兩種方式：

1. 加final關鍵字。對於一個類，所有的成員變量應該是私有的，並且可能的情況下，所有的成員變量應該加上final關鍵字。需要注意如果成員變量又是一個對象時，這個對象所對應的類也要是不可變，才能保證整個類是不可變的。
2. 根本就不提供任何可供修改成員變量的地方，同時成員變量也不作為方法的返回值。

下面例子中的，成員變量都是final並且也沒有提供給外部修改變量的地方，因此是線程安全的。

public class ImmutableFinal {
	
	private final int a;
	private final int b;
	
	public ImmutableFinal(int a, int b) {
		super();
		this.a = a;
		this.b = b;
	}

	public int getA() {
		return a;
	}

	public int getB() {
		return b;
	}
}

下面的例子中，雖然User成員變量是final的，無法修改引用。但是外部依然可以通過getUser獲取到User的引用之后，修改User對象。

public class ImmutableFinalRef {
	
	private final int a;
	private final int b;
	private final User user;//這里就不能保證線程安全了
	
	public ImmutableFinalRef(int a, int b) {
		super();
		this.a = a;
		this.b = b;
		this.user = new User();
	}

	public int getA() {
		return a;
	}

	public int getB() {
		return b;
	}
	
	public User getUser() {
		return user;
	}

	public static class User{
		private int age;

		public User(int age) {
			super();
			this.age = age;
		}

		public int getAge() {
			return age;
		}

		public void setAge(int age) {
			this.age = age;
		}
		
	}
	
	public static void main(String[] args) {
		ImmutableFinalRef ref = new ImmutableFinalRef(12,23);
		User u = ref.getUser();
		//u.setAge(35);
	}
}

volatile

volitile在ConcurrentHashMap等並發容器中都有使用，用於保證變量的可見性。最適合一個線程寫，多個線程讀的情景。

加鎖和CAS

加鎖可以顯示地控制線程對類的訪問，使用正確可以保證線程安全。

CAS操作通過不斷的循環對比，試圖對目標對象進行修改，也能保證線程安全。廣泛用於JDK並發容器的實現中。

安全的發布

類中持有的成員變量，特別是對象的引用，如果這個成員對象不是線程安全的，通過get等方法發布出去，會造成這個成員對象本身持有的數據在多線程下不正確的修改，從而造成整個類線程不安全的問題。

ThreadLocal

這個類能使線程中的某個值與保存值的對象關聯起來。ThreadLocal提供了get與set等訪問接口與方法，這些方法為使用該變量的每個線程都存有一份獨立的副本，因此get總是返回由當前執行線程在調用set時設置的最新值。

當某個線程初次調用ThreadLocal.get方法時，就會調用initialValue來獲取初始值。從概念上講，你可以將ThreadLocal 視為包含了Map<Thread, T>對象，其中保存了特定於該線程的值，但ThreadLocal的實現並非如此，這些特定的值保存在Thread對象中，當線程終止后，這些值會作為垃圾回收。

7.3 死鎖

定義

死鎖是指兩個或兩個以上的進程在執行過程中，由於競爭資源或者由於彼此通信而造成的一種阻塞的現象，若無外力作用，它們都將無法推進下去。此時稱系統處於死鎖狀態或系統產生了死鎖。

死鎖的根本成因：獲取鎖的順序不一致導致。

可以利用下面的示意圖幫助理解：

死鎖范例

下面的程序中，兩個線程分別獲取到了first和second，然后相互等待，產生了死鎖。

public class DeadLockSample extends Thread {
    private String first;
    private String second;
    public DeadLockSample(String name, String first, String second) {
        super(name);
        this.first = first;
        this.second = second;
    }
    public void run() {
        synchronized (first) {
            System.out.println(this.getName() + " obtained: " + first);
            try {
                Thread.sleep(1000L);
                synchronized(second) {
                    System.out.println(this.getName() + " obtained: " + second);
                }
            } catch (InterruptedException e) {
                // Do nothing
            }
        }
    }
    public static void main(String[] args) throws InterruptedException {
        String lockA = "lockA";
        String lockB = "lockB";
        DeadLockSample t1 = new DeadLockSample("Thread1", lockA, lockB);
        DeadLockSample t2 = new DeadLockSample("Thread2", lockB, lockA);
        t1.start();
        t2.start();
        t1.join();
        t2.join();
    }
}

定位和解決死鎖

Debug時可以使用 jps 或者系統的 ps 命令、任務管理器等工具，確定進程 ID。其次，調用 jstack 獲取線程棧，jstack your_pid. jstack 本身也會把類似的簡單死鎖抽取出來，直接打印出來。

如果是開發自己的管理工具，需要用更加程序化的方式掃描服務進程、定位死鎖，可以考慮使用 Java 提供的標准管理 API，ThreadMXBean，其直接就提供 findDeadlockedThreads() 方法用於定位，上面的例子中用到了這個方法。

怎么預防死鎖？

1. 如果可能的話，盡量避免使用多個鎖，並且只有需要時才持有鎖。

2. 如果必須使用多個鎖，盡量設計好鎖的獲取順序。如果對於兩個線程的情況，可以參考如下的實現：

在實現轉賬的類時，為了防止由於相互轉賬導致的死鎖，下面的實現中，通過對比賬戶的hash值來確定獲取鎖的順序。當兩者的hash值相等時，雖然這種情況非常少見，使用了單獨的鎖，來控制兩個線程的訪問順序。

注意System.identityHashCode()是JDK自帶的hash實現，在絕大部分情況下，保證了對象hash值的唯一性。

public class SafeOperate implements ITransfer {
	private static Object tieLock = new Object();//加時賽鎖

    @Override
    public void transfer(UserAccount from, UserAccount to, int amount)
            throws InterruptedException {
    	
    	int fromHash = System.identityHashCode(from);
    	int toHash = System.identityHashCode(to);
    	//先鎖hash小的那個
    	if(fromHash<toHash) {
            synchronized (from){
                synchronized (to){
                    System.out.println(Thread.currentThread().getName()
                    		+" get"+to.getName());
                    from.flyMoney(amount);
                    to.addMoney(amount);
                }
            }    		
    	}else if(toHash<fromHash) {
            synchronized (to){
                Thread.sleep(100);
                synchronized (from){
                    from.flyMoney(amount);
                    to.addMoney(amount);
                }
            }    		
    	}else {//解決hash沖突的方法
    		synchronized (tieLock) {
				synchronized (from) {
					synchronized (to) {
	                    from.flyMoney(amount);
	                    to.addMoney(amount);						
					}
				}
			}
    	}
    	
    }
}

3. 使用帶超時的方法，為程序帶來更多可控性。

類似 Object.wait(…) 或者 CountDownLatch.await(…)，都支持所謂的 timed_wait，我們完全可以就不假定該鎖一定會獲得，指定超時時間，並為無法得到鎖時准備退出邏輯。

4. 使用Lock實現（推薦）

並發 Lock 實現，如 ReentrantLock 還支持非阻塞式的獲取鎖操作 tryLock()，這是一個插隊行為（barging），並不在乎等待的公平性，如果執行時對象恰好沒有被獨占，則直接獲取鎖。

標准的使用流程如下：

while(true) {
   if(A.getLock().tryLock()) {
    try {
    	if(B.getLock().tryLock()) {
    		try {
    		  //兩把鎖都拿到了，開始執行業務代碼
    	           break;
    		}finally {
    		  B.getLock().unlock();
    	    }
       }
    }finally {
    	A.getLock().unlock();
    }
  }
  // 非常重要，sleep隨機的時間，以防兩個線程謙讓，產生長時間的等待，也就是活鎖
  SleepTools.ms(r.nextInt(10));
}

7.4 活鎖/線程飢餓/無鎖

活鎖

活鎖恰恰與死鎖相反，死鎖是大家都拿不到資源都占用着對方的資源，而活鎖是拿到資源卻又相互釋放不執行。當多線程中出現了相互謙讓，都主動將資源釋放給別的線程使用，這樣這個資源在多個線程之間跳動而又得不到執行，這就是活鎖。

在上面解決死鎖的第四個方案中，為了避免活鎖，采用了隨機休眠的機制。

線程飢餓

線程執行中有線程優先級，優先級高的線程能夠插隊並優先執行，這樣如果優先級高的線程一直搶占優先級低線程的資源，導致低優先級線程無法得到執行，這就是飢餓。當然還有一種飢餓的情況，一個線程一直占着一個資源不放而導致其他線程得不到執行，與死鎖不同的是飢餓在以后一段時間內還是能夠得到執行的，如那個占用資源的線程結束了並釋放了資源。

無鎖

對於並發控制而言，鎖是一種悲觀的策略，它總是假設每一次的臨界區操作會產生沖突，由此，如果有多個線程同時需要訪問臨界區資源，則寧可犧牲資源讓線程進行等待。

無鎖是一種樂觀的策略，它假設對資源的訪問是沒有沖突的。既然沒有沖突，自然不需要等待，所以所有的線程都可以在不停頓地狀態下持續執行。當遇到沖突，則使用CAS來檢測線程沖突，如果發現沖突，則重試直到沒有沖突為止。

CAS算法的過程是，它包含三個參數CAS（V，E，N），V表示要更新的變量，E表示預期值，N表示新值。僅當V值等於E值時，才將V的值設置為N，如果V值和E值不同，說明已經有其他線程做了更新，則當前線程什么都不做。使用CAS操作一個變量時，只有一個會勝出，並成功更新，其余均會失敗。

7.5 影響性能的因素

• 上下文切換：一般花費5000-10000個時鍾周期，幾微秒
• 內存同步：加鎖等操作，增加額外的指令執行時間
• 阻塞：掛起線程，包括額外的上下文切換

7.6 鎖性能優化

減少鎖的持有時間

減少鎖的持有時間有助於降低鎖沖突的可能性,進而提升系統的並發能力。

減小鎖粒度

這種技術的典型使用場景就是ConcurrentHashMap。

對於HashMap來說,最重要的兩個方法就是get() 和put(),一種最自然的想法就是對整個HashMap加鎖,必然可以得到一個線程安全的對象.但是這樣做,我們就認為加鎖粒度太大.對於ConcurrentHashMap,它內部進一步細分了若干個小的hashMap,稱之為段(SEGMENT).默認的情況下,一個ConcurrentHashMap被進一步細分為16個段

如果需要在ConcurrentHashMap中增加一個新的表項,並不是整個HashMap加鎖,而是首先根據hashcode得到該表項應該被存放到哪個段中,然后對該段加鎖,並完成put()操作.在多線程環境中,如果多個線程同時進行put()操作,只要被加入的表項不存放在同一個段中,則線程間便可以做到真正的並行。

讀寫分離鎖來替換獨占鎖

在讀多寫少的場合,使用讀寫鎖可以有效提升系統的並發能力

鎖分離

如果將讀寫鎖的思想進一步的延伸,就是鎖分離.讀寫鎖根據讀寫鎖操作功能上的不同,進行了有效的鎖分離.使用類似的思想,也可以對獨占鎖進行分離.

以LinkedBlockingQueue為例,take函數和put函數分別實現了沖隊列取和往隊列加數據，雖然兩個方法都對隊列進項了修改，但是LinkedBlockingQueue是基於鏈表的所以一個操作的是頭，一個是隊列尾端，從理論情況下將並不沖突

如果使用獨占鎖則take和put就不能完成真正的並發，所以jdk並沒有才用這種方式取而代之的是兩把不同的鎖分離了put和take的操作

鎖粗化

凡事都有一個度,如果對同一個鎖不停地進行請求,同步和釋放,其本身也會消耗系統寶貴的資源,反而不利於性能的優化。

為此,虛擬機在遇到一連串連續地對同一鎖不斷進行請求和釋放的操作時,便會把所有的鎖操作整合成對鎖的一次請求,從而減少對鎖的請求同步次數,這個操作叫做鎖的粗化.

7.7 實現線程安全的單例模式

懶漢式

public static synchronized Singleton getInstance() {
    if (instance == null) {
        instance = new Singleton();
    }
    return instance;
}

線程安全，並且解決了多實例的問題，但是它並不高效。因為在任何時候只能有一個線程調用 getInstance() 方法。

雙重檢驗鎖

public class Singleton {
    private static volatile Singleton singleton = null;
    private Singleton() {
    }
    public static Singleton getSingleton() {
        if (singleton == null) { // 盡量避免重復進入同步塊
            synchronized (Singleton.class) { // 同步.class，意味着對同步類方法調用
                if (singleton == null) {
                    singleton = new Singleton();
                }
            }
        }
        return singleton;
    }
}

• volatile 能夠提供可見性，以及保證 getInstance 返回的是初始化完全的對象。
• 在同步之前進行 null 檢查，以盡量避免進入相對昂貴的同步塊。
• 直接在 class 級別進行同步，保證線程安全的類方法調用。

在這段代碼中，爭論較多的是 volatile 修飾靜態變量，當 Singleton 類本身有多個成員變量時，需要保證初始化過程完成后，才能被 get 到。 在現代 Java 中，內存排序模型（JMM）已經非常完善，通過 volatile 的 write 或者 read，能保證所謂的 happen-before，也就是避免常被提到的指令重排。換句話說，構造對象的 store 指令能夠被保證一定在 volatile read 之前。

餓漢式

這種方法非常簡單，因為單例的實例被聲明成 static 和 final 變量了，在第一次加載類到內存中時就會初始化，所以創建實例本身是線程安全的。

public class Singleton{
    //類加載時就初始化
    private static final Singleton instance = new Singleton();
    
    private Singleton(){}

    public static Singleton getInstance(){
        return instance;
    }
}

靜態內部類(推薦)

public class Singleton {  
    private static class SingletonHolder {  
        private static final Singleton INSTANCE = new Singleton();  
    }  
    private Singleton (){}  
    public static final Singleton getInstance() {  
        return SingletonHolder.INSTANCE; 
    }  
}

靜態內部類是在被調用時才會被加載，因此它是懶漢式的。

---

本文由『后端精進之路』原創，首發於博客 http://teckee.github.io/ , 轉載請注明出處

搜索『后端精進之路』關注公眾號，立刻獲取最新文章和價值2000元的BATJ精品面試課程。