多線(xiàn)程并發(fā)問(wèn)題，基本是面試必問(wèn)的。

大部分同學(xué)應該都知道Synchronized，Lock，部分同學(xué)能說(shuō)到volatile、并發(fā)包，優(yōu)秀的同學(xué)則能在前面的基礎上，說(shuō)出Synchronized、volatile的原理，以及并發(fā)包中常用的數據結構，例如ConcurrentHashMap的原理。

這篇文章將總結多線(xiàn)程并發(fā)的各種處理方式，希望對大家有所幫助。

為什么多線(xiàn)程同時(shí)訪(fǎng)問(wèn)（讀寫(xiě)）同個(gè)變量，會(huì )有并發(fā)問(wèn)題？

1. Java 內存模型規定了所有的變量都存儲在主內存中，每條線(xiàn)程有自己的工作內存。

2. 線(xiàn)程的工作內存中保存了該線(xiàn)程中用到的變量的主內存副本拷貝，線(xiàn)程對變量的所有操作都必須在工作內存中進(jìn)行，而不能直接讀寫(xiě)主內存。

3. 線(xiàn)程訪(fǎng)問(wèn)一個(gè)變量，首先將變量從主內存拷貝到工作內存，對變量的寫(xiě)操作，不會(huì )馬上同步到主內存。

4. 不同的線(xiàn)程之間也無(wú)法直接訪(fǎng)問(wèn)對方工作內存中的變量，線(xiàn)程間變量的傳遞均需要自己的工作內存和主存之間進(jìn)行數據同步進(jìn)行。

Java 內存模型(JMM) 作用于工作內存（本地內存）和主存之間數據同步過(guò)程，它規定了如何做數據同步以及什么時(shí)候做數據同步，如下圖。

原子性：在一個(gè)操作中，CPU 不可以在中途暫停然后再調度，即不被中斷操作，要么執行完成，要么就不執行。

可見(jiàn)性：多個(gè)線(xiàn)程訪(fǎng)問(wèn)同一個(gè)變量時(shí)，一個(gè)線(xiàn)程修改了這個(gè)變量的值，其他線(xiàn)程能夠立即看得到修改的值。

有序性：程序執行的順序按照代碼的先后順序執行。

下面結合不同場(chǎng)景分析解決并發(fā)問(wèn)題的處理方式。

一、volatile

1.1 volatile 特性

保證可見(jiàn)性，不保證原子性。

1. 當寫(xiě)一個(gè)volatile變量時(shí)，JVM會(huì )把本地內存的變量強制刷新到主內存中。

2. 這個(gè)寫(xiě)操作導致其他線(xiàn)程中的緩存無(wú)效，其他線(xiàn)程讀，會(huì )從主內存讀。volatile的寫(xiě)操作對其它線(xiàn)程實(shí)時(shí)可見(jiàn)。

禁止指令重排序 指令重排序是指編譯器和處理器為了優(yōu)化程序性能對指令進(jìn)行排序的一種手段，需要遵守一定規則：

1. 不會(huì )對存在依賴(lài)關(guān)系的指令重排序，例如 a = 1;b = a; a 和b存在依賴(lài)關(guān)系，不會(huì )被重排序。

2. 不能影響單線(xiàn)程下的執行結果。比如：a=1;b=2;c=a+b這三個(gè)操作,前兩個(gè)操作可以重排序，但是c=a+b不會(huì )被重排序，因為要保證結果是3。

1.2 使用場(chǎng)景

對于一個(gè)變量，只有一個(gè)線(xiàn)程執行寫(xiě)操作，其它線(xiàn)程都是讀操作，這時(shí)候可以用 volatile 修飾這個(gè)變量。

1.3 單例雙重鎖為什么要用到volatile？

public class TestInstance {

private static volatile TestInstance mInstance;

public static TestInstance getInstance(){ //1
  if (mInstance == null){ //2
  synchronized (TestInstance.class){ //3
   if (mInstance == null){ //4
     mInstance = new TestInstance(); //5
   }
  }
 }
 return mInstance;
}

假如沒(méi)有用volatile，并發(fā)情況下會(huì )出現問(wèn)題，線(xiàn)程A執行到注釋5 new TestInstance() 的時(shí)候，分為如下幾個(gè)幾步操作：

1. 分配內存

2. 初始化對象

3. mInstance 指向內存

這時(shí)候如果發(fā)生指令重排，執行順序是132，執行到第3的時(shí)候，線(xiàn)程B剛好進(jìn)來(lái)了，并且執行到注釋2，這時(shí)候判斷mInstance 不為空，直接使用一個(gè)未初始化的對象。所以使用volatile關(guān)鍵字來(lái)禁止指令重排序。

1.4 volatile 原理

在JVM底層volatile是采用內存屏障來(lái)實(shí)現的，內存屏障會(huì )提供3個(gè)功能：

1. 它確保指令重排序時(shí)不會(huì )把其后面的指令排到內存屏障之前的位置，也不會(huì )把前面的指令排到內存屏障的后面；即在執行到內存屏障這句指令時(shí)，在它前面的操作已經(jīng)全部完成；

2. 它會(huì )強制將緩存的修改操作立即寫(xiě)到主內存

3. 寫(xiě)操作會(huì )導致其它CPU中的緩存行失效，寫(xiě)之后，其它線(xiàn)程的讀操作會(huì )從主內存讀。

1.5 volatile 的局限性

volatile 只能保證可見(jiàn)性，不能保證原子性寫(xiě)操作對其它線(xiàn)程可見(jiàn)，但是不能解決多個(gè)線(xiàn)程同時(shí)寫(xiě)的問(wèn)題。

二、Synchronized

2.1 Synchronized 使用場(chǎng)景

多個(gè)線(xiàn)程同時(shí)寫(xiě)一個(gè)變量。

例如售票，余票是100張，窗口A(yíng)和窗口B同時(shí)各賣(mài)出一張票， 假如余票變量用 volatile 修飾，是有問(wèn)題的。

A窗口獲取余票是100，B窗口獲取余票也是100，A賣(mài)出一張變成99，刷新回主內存，同時(shí)B賣(mài)出一張變成99，也刷新回主內存，會(huì )導致最終主內存余票是99而不是98。

前面說(shuō)到 volatile 的局限性，就是多個(gè)線(xiàn)程同時(shí)寫(xiě)的情況，這種情況一般可以使用Synchronized。

Synchronized 可以保證同一時(shí)刻，只有一個(gè)線(xiàn)程可執行某個(gè)方法或某個(gè)代碼塊。

2.2 Synchronized 原理

public class SynchronizedTest {

public static void main(String[] args) {
 synchronized (SynchronizedTest.class) {
   System.out.println("123");
 }
 method();
}

private static void method() {
}
}

將這段代碼先用javac命令編譯，再java p -v SynchronizedTest.class命令查看字節碼，部分字節碼如下

public static void main(java.lang.String[]);
descriptor: ([Ljava/lang/String;)V
flags: ACC_PUBLIC, ACC_STATIC
Code:
 stack=2, locals=3, args_size=1
  0: ldc #2   // class com/lanshifu/opengldemo/test/SynchronizedTest
  2: dup
  3: astore_1
  4: monitorenter
  5: getstatic #3   // Field java/lang/System.out:Ljava/io/PrintStream;
  8: ldc #4      // String 123
 10: invokevirtual #5 // Method java/io/PrintStream.println:(Ljava/lang/String;)V
 13: aload_1
 14: monitorexit
 15: goto          23
 18: astore_2
 19: aload_1
 20: monitorexit
 21: aload_2
 22: athrow
 23: invokestatic #6   // Method method:()V
 26: return

可以看到 4: monitorenter 和 14: monitorexit，中間是打印的語(yǔ)句。

執行同步代碼塊，首先會(huì )執行monitorenter指令，然后執行同步代碼塊中的代碼，退出同步代碼塊的時(shí)候會(huì )執行monitorexit指令 。

使用Synchronized進(jìn)行同步，其關(guān)鍵就是必須要對對象的監視器monitor進(jìn)行獲取，當線(xiàn)程獲取monitor后才能繼續往下執行，否則就進(jìn)入同步隊列，線(xiàn)程狀態(tài)變成BLOCK，同一時(shí)刻只有一個(gè)線(xiàn)程能夠獲取到monitor，當監聽(tīng)到monitorexit被調用，隊列里就有一個(gè)線(xiàn)程出隊，獲取monitor。詳情參考：

https://www.jianshu.com/p/d53bf830fa09

每個(gè)對象擁有一個(gè)計數器，當線(xiàn)程獲取該對象鎖后，計數器就會(huì )加一，釋放鎖后就會(huì )將計數器減一，所以只要這個(gè)鎖的計數器大于0，其它線(xiàn)程訪(fǎng)問(wèn)就只能等待。

2.3 Synchronized 鎖的升級

大家對Synchronized的理解可能就是重量級鎖，但是Java1.6對 Synchronized 進(jìn)行了各種優(yōu)化之后，有些情況下它就并不那么重，Java1.6 中為了減少獲得鎖和釋放鎖帶來(lái)的性能消耗而引入的偏向鎖和輕量級鎖。

偏向鎖： 大多數情況下，鎖不僅不存在多線(xiàn)程競爭，而且總是由同一線(xiàn)程多次獲得，為了讓線(xiàn)程獲得鎖的代價(jià)更低而引入了偏向鎖。

當一個(gè)線(xiàn)程A訪(fǎng)問(wèn)加了同步鎖的代碼塊時(shí)，會(huì )在對象頭中存 儲當前線(xiàn)程的id，后續這個(gè)線(xiàn)程進(jìn)入和退出這段加了同步鎖的代碼塊時(shí)，不需要再次加鎖和釋放鎖。

輕量級鎖： 在偏向鎖情況下，如果線(xiàn)程B也訪(fǎng)問(wèn)了同步代碼塊，比較對象頭的線(xiàn)程id不一樣，會(huì )升級為輕量級鎖，并且通過(guò)自旋的方式來(lái)獲取輕量級鎖。

重量級鎖： 如果線(xiàn)程A和線(xiàn)程B同時(shí)訪(fǎng)問(wèn)同步代碼塊，則輕量級鎖會(huì )升級為重量級鎖，線(xiàn)程A獲取到重量級鎖的情況下，線(xiàn)程B只能入隊等待，進(jìn)入BLOCK狀態(tài)。

2.4 Synchronized 缺點(diǎn)

1. 不能設置鎖超時(shí)時(shí)間

2. 不能通過(guò)代碼釋放鎖

3. 容易造成死鎖

三、ReentrantLock

上面說(shuō)到Synchronized的缺點(diǎn)，不能設置鎖超時(shí)時(shí)間和不能通過(guò)代碼釋放鎖，ReentranLock就可以解決這個(gè)問(wèn)題。

在多個(gè)條件變量和高度競爭鎖的地方，用ReentrantLock更合適，ReentrantLock還提供了Condition，對線(xiàn)程的等待和喚醒等操作更加靈活，一個(gè)ReentrantLock可以有多個(gè)Condition實(shí)例，所以更有擴展性。

3.1 ReentrantLock 的使用

lock 和 unlock

ReentrantLock reentrantLock = new ReentrantLock();
 System.out.println("reentrantLock->lock");
 reentrantLock.lock();
  try {    
   System.out.println("睡眠2秒...");
   Thread.sleep(2000);
  } catch (InterruptedException e) {
   e.printStackTrace();
  }finally {
   reentrantLock.unlock();
   System.out.println("reentrantLock->unlock");
}

實(shí)現可定時(shí)的鎖請求：tryLock

public static void main(String[] args) {
  ReentrantLock reentrantLock = new ReentrantLock();
  Thread thread1 = new Thread_tryLock(reentrantLock);
  thread1.setName("thread1");
  thread1.start();
  Thread thread2 = new Thread_tryLock(reentrantLock);
  thread2.setName("thread2");
  thread2.start();
}


static class Thread_tryLock extends Thread {
  ReentrantLock reentrantLock;
  public Thread_tryLock(ReentrantLock reentrantLock) {
    this.reentrantLock = reentrantLock;
  }

  @Override
  public void run() {
    try {
      System.out.println("try lock:" + Thread.currentThread().getName());
     boolean tryLock = reentrantLock.tryLock(3, TimeUnit.SECONDS);
     if (tryLock) {
      System.out.println("try lock success :" + Thread.currentThread().getName());
        System.out.println("睡眠一下：" + Thread.currentThread().getName());
      Thread.sleep(5000);
        System.out.println("醒了：" + Thread.currentThread().getName());
       } else {
         System.out.println("try lock 超時(shí) :" + Thread.currentThread().getName());
       }
    } catch (InterruptedException e) {
       e.printStackTrace();
    } finally {
       System.out.println("unlock:" + Thread.currentThread().getName());
      reentrantLock.unlock();
     }
   }
}

打印的日志：

try lock:thread1
try lock:thread2
try lock success :thread2
睡眠一下：thread2
try lock 超時(shí) :thread1
unlock:thread1
Exception in thread "thread1" java.lang.IllegalMonitorStateException
 at java.util.concurrent.locks.ReentrantLock$Sync.tryRelease(ReentrantLock.java:151)
 at java.util.concurrent.locks.AbstractQueuedSynchronizer.release(AbstractQueuedSynchronizer.java:1261)
 at java.util.concurrent.locks.ReentrantLock.unlock(ReentrantLock.java:457)
 at com.lanshifu.demo_module.test.lock.ReentranLockTest$Thread_tryLock.run(ReentranLockTest.java:60)
醒了：thread2
unlock:thread2

上面演示了trtLock的使用，trtLock設置獲取鎖的等待時(shí)間，超過(guò)3秒直接返回失敗，可以從日志中看到結果。有異常是因為thread1獲取鎖失敗，不應該調用unlock。

3.2 Condition 條件

public static void main(String[] args) {
  Thread_Condition thread_condition = new Thread_Condition();
  thread_condition.setName("測試Condition的線(xiàn)程");
  thread_condition.start();
  try {
    Thread.sleep(2000);
  } catch (InterruptedException e) {
    e.printStackTrace();
  }
    thread_condition.singal();
  }
static class Thread_Condition extends Thread {

  @Override
  public void run() {
    await();
  }

  private ReentrantLock lock = new ReentrantLock();
  public Condition condition = lock.newCondition();

  public void await() {
    try {
     System.out.println("lock");
     lock.lock();
     System.out.println(Thread.currentThread().getName() + ":我在等待通知的到來(lái)...");
     condition.await();//await 和 signal 對應
     //condition.await(2, TimeUnit.SECONDS); //設置等待超時(shí)時(shí)間
     System.out.println(Thread.currentThread().getName() + ":等到通知了，我繼續執行>>>");
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
     System.out.println("unlock");
     lock.unlock();
    }
   }

  public void singal() {
   try {
    System.out.println("lock");
    lock.lock();
    System.out.println("我要通知在等待的線(xiàn)程，condition.signal()");
    condition.signal();//await 和 signal 對應
    Thread.sleep(1000);
   } catch (InterruptedException e) {
     e.printStackTrace();
   } finally {
     System.out.println("unlock");
     lock.unlock();
   }
  }
}

運行打印日志

lock
測試Condition的線(xiàn)程:我在等待通知的到來(lái)...
lock
我要通知在等待的線(xiàn)程，condition.signal()
unlock
測試Condition的線(xiàn)程:等到通知了，我繼續執行>>>
unlock

上面演示了Condition的 await 和 signal 使用，前提要先lock。

3.3 公平鎖與非公平鎖

ReentrantLock 構造函數傳true表示公平鎖。

公平鎖表示線(xiàn)程獲取鎖的順序是按照線(xiàn)程加鎖的順序來(lái)分配的，即先來(lái)先得的順序。而非公平鎖就是一種鎖的搶占機制，是隨機獲得鎖的，可能會(huì )導致某些線(xiàn)程一致拿不到鎖，所以是不公平的。

3.4 ReentrantLock 注意點(diǎn)

1. ReentrantLock使用lock和unlock來(lái)獲得鎖和釋放鎖

2. unlock要放在finally中，這樣正常運行或者異常都會(huì )釋放鎖

3. 使用condition的await和signal方法之前，必須調用lock方法獲得對象監視器

四、并發(fā)包

通過(guò)上面分析，并發(fā)嚴重的情況下，使用鎖顯然效率低下，因為同一時(shí)刻只能有一個(gè)線(xiàn)程可以獲得鎖，其它線(xiàn)程只能乖乖等待。

Java提供了并發(fā)包解決這個(gè)問(wèn)題，接下來(lái)介紹并發(fā)包里一些常用的數據結構。

4.1 ConcurrentHashMap

我們都知道HashMap是線(xiàn)程不安全的數據結構，HashTable則在HashMap基礎上，get方法和put方法加上Synchronized修飾變成線(xiàn)程安全，不過(guò)在高并發(fā)情況下效率底下，最終被ConcurrentHashMap替代。

ConcurrentHashMap 采用分段鎖，內部默認有16個(gè)桶，get和put操作，首先將key計算hashcode，然后跟16取余，落到16個(gè)桶中的一個(gè)，然后每個(gè)桶中都加了鎖（ReentrantLock），桶中是HashMap結構（數組加鏈表，鏈表過(guò)長(cháng)轉紅黑樹(shù)）。

所以理論上最多支持16個(gè)線(xiàn)程同時(shí)訪(fǎng)問(wèn)。

4.2 LinkBlockingQueue

鏈表結構的阻塞隊列，內部使用多個(gè)ReentrantLock

/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();
/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();
/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();
/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();

private void signalNotEmpty() {
  final ReentrantLock takeLock = this.takeLock;
  takeLock.lock();
  try {
   notEmpty.signal();
  } finally {
    takeLock.unlock();
  }
}

/**
 * Signals a waiting put. Called only from take/poll.
*/
private void signalNotFull() {
  final ReentrantLock putLock = this.putLock;
  putLock.lock();
  try {
   notFull.signal();
  } finally {
   putLock.unlock();
  }
}

源碼不貼太多，簡(jiǎn)單說(shuō)一下LinkBlockingQueue 的邏輯：

1. 從隊列獲取數據，如果隊列中沒(méi)有數據，會(huì )調用notEmpty.await();進(jìn)入等待。

2. 在放數據進(jìn)去隊列的時(shí)候會(huì )調用notEmpty.signal();，通知消費者，1中的等待結束，喚醒繼續執行。

3. 從隊列里取到數據的時(shí)候會(huì )調用notFull.signal();，通知生產(chǎn)者繼續生產(chǎn)。

4. 在put數據進(jìn)入隊列的時(shí)候，如果判斷隊列中的數據達到最大值，那么會(huì )調用notFull.await();，等待消費者消費掉，也就是等待3去取數據并且發(fā)出notFull.signal();，這時(shí)候生產(chǎn)者才能繼續生產(chǎn)。

LinkBlockingQueue 是典型的生產(chǎn)者消費者模式，源碼細節就不多說(shuō)。

4.3 原子操作類(lèi)：AtomicInteger

內部采用CAS（compare and swap）保證原子性

舉一個(gè)int自增的例子

AtomicInteger atomicInteger = new AtomicInteger(0);
atomicInteger.incrementAndGet();//自增

源碼看一下

/**
 * Atomically increments by one the current value.
 *
 * @return the updated value
 */
 public final int incrementAndGet() {
   return U.getAndAddInt(this, VALUE, 1) + 1;
  }

U 是 Unsafe，看下 Unsafe#getAndAddInt

public final int getAndAddInt(Object var1, long var2, int var4) {
  int var5;
  do {
    var5 = this.getIntVolatile(var1, var2);
  } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));
    return var5;
  }

通過(guò)compareAndSwapInt保證原子性。

面試中問(wèn)到多線(xiàn)程并發(fā)問(wèn)題，可以這么答：

1. 當只有一個(gè)線(xiàn)程寫(xiě)，其它線(xiàn)程都是讀的時(shí)候，可以用volatile修飾變量

2. 當多個(gè)線(xiàn)程寫(xiě)，那么一般情況下并發(fā)不嚴重的話(huà)可以用Synchronized，Synchronized并不是一開(kāi)始就是重量級鎖，在并發(fā)不嚴重的時(shí)候，比如只有一個(gè)線(xiàn)程訪(fǎng)問(wèn)的時(shí)候，是偏向鎖；當多個(gè)線(xiàn)程訪(fǎng)問(wèn)，但不是同時(shí)訪(fǎng)問(wèn)，這時(shí)候鎖升級為輕量級鎖；當多個(gè)線(xiàn)程同時(shí)訪(fǎng)問(wèn)，這時(shí)候升級為重量級鎖。所以在并發(fā)不是很?chē)乐氐那闆r下，使用Synchronized是可以的。不過(guò)Synchronized有局限性，比如不能設置鎖超時(shí)，不能通過(guò)代碼釋放鎖。

3. ReentranLock 可以通過(guò)代碼釋放鎖，可以設置鎖超時(shí)。

4. 高并發(fā)下，Synchronized、ReentranLock 效率低，因為同一時(shí)刻只有一個(gè)線(xiàn)程能進(jìn)入同步代碼塊，如果同時(shí)有很多線(xiàn)程訪(fǎng)問(wèn)，那么其它線(xiàn)程就都在等待鎖。這個(gè)時(shí)候可以使用并發(fā)包下的數據結構，例如ConcurrentHashMap，LinkBlockingQueue，以及原子性的數據結構如：AtomicInteger。

面試的時(shí)候按照上面總結的這個(gè)思路回答基本就ok了。既然說(shuō)到并發(fā)包，那么除了ConcurrentHashMap，其它一些常用的數據結構的原理也需要去了解下，例如HashMap、HashTable、TreeMap原理，ArrayList、LinkedList對比，這些都是老生常談的，自己去看源碼或者一些博客。

關(guān)于多線(xiàn)程并發(fā)就先總結到這里，如果是應付面試的話(huà)按照這篇文章的思路來(lái)準備應該是沒(méi)太大問(wèn)題的。