本文討論的重點在于多線程應用程序的性能問題。我們會先給性能和擴展性下一個定義，然后再仔細學習一下Amdahl法則。下面的內容我們會考察一下如何用不同的技術方法來減少鎖競爭，以及如何用代碼來實現。
1、性能
我們都知道，多線程可以用來提高程序的性能，背后的原因在于我們有多核的CPU或多個CPU。每個CPU的內核都可以自己完成任務，因此把一個大的任務分解成一系列的可彼此獨立運行的小任務就可以提高程序的整體性能了。可以舉個例子，比如有個程序用來將硬盤上某個文件夾下的所有圖片的尺寸進行修改，應用多線程技術就可以提高它的性能。使用單線程的方式只能依次遍歷所有圖片文件并且執行修改，如果我們的CPU有多個核心的話，毫無疑問，它只能利用其中的一個核。使用多線程的方式的話，我們可以讓一個生產者線程掃描文件系統把每個圖片都添加到一個隊列中，然后用多個工作線程來執行這些任務。如果我們的工作線程的數量和CPU總的核心數一樣的話，我們就能保證每個CPU核心都有活可干，直到任務被全部執行完成。
對于另外一種需要較多IO等待的程序來說，利用多線程技術也能提高整體性能。假設我們要寫這樣一個程序，需要抓取某個網站的所有HTML文件，并且將它們存儲到本地磁盤上。程序可以從某一個網頁開始，然后解析這個網頁中所有指向本網站的鏈接，然后依次抓取這些鏈接，這樣周而復始。因為從我們對遠程網站發起請求到接收到所有的網頁數據需要等待一段時間，所以我們可以將此任務交給多個線程來執行。讓一個或稍微更多一點的線程來解析已經收到的HTML網頁以及將找到的鏈接放入隊列中，讓其他所有的線程負責請求獲取頁面。與上一個例子不同的是，在這個例子中，你即便使用多于CPU核心數量的線程也仍然能夠獲得性能提升。
上面這兩個例子告訴我們，高性能就是在短的時間窗口內做盡量多的事情。這個當然是對性能一詞的最經典解釋了。但是同時，使用線程也能很好地提升我們程序的響應速度。想象我們有這樣一個圖形界面的應用程序，上方有一個輸入框，輸入框下面有一個名字叫“處理”的按鈕。當用戶按下這個按鈕的時候，應用程序需要重新對按鈕的狀態進行渲染（按鈕看起來被按下了，當松開鼠標左鍵時又恢復原狀），并且開始對用戶的輸入進行處理。如果處理用戶輸入的這個任務比較耗時的話，單線程的程序就無法繼續響應用戶其他的輸入動作了，比如，來自操作系統傳送過來的用戶單擊鼠標事件或鼠標指針移動事件等等，這些事件的響應需要有獨立的線程來響應。
可擴展性（Scalability）的意思是程序具備這樣的能力：通過添加計算資源就可以獲得更高的性能。想象我們需要調整很多圖片的大小，因為我們機器的CPU核心數是有限的，所以增加線程數量并不總能相應提高性能。相反，因為調度器需要負責更多線程的創建和關閉，也會占用CPU資源，反而有可能降低性能。
1.1 Amdahl法則
上一段提到了在某些情形下，添加額外的運算資源可以提高程序的整體性能。為了能夠計算出當我們添加了額外的資源的時候到底能獲得多少性能提升，我們有必要來檢查一下程序有哪些部分是串行運行（或同步運行），有哪些部分是并行運行的。如果我們把需要同步執行的代碼占比量化為B（例如，需要同步執行的代碼的行數），把CPU的總核心數記為n，那么，根據Amdahl法則，我們可以獲得的性能提升的上限是：
如果n趨于無窮大的話，(1-B)/n就收斂于0。因此，我們可以忽略這個表達式的值，因此性能提升位數收斂于1/B，這里面的B代表是那些必須同步運行的代碼比例。如果B等于0.5的話，那意味著程序的一半代碼無法并行運行，0.5的倒數是2，因此，即使我們添加無數個CPU核心，我們獲得的性能提升也最多是2倍。假設我們現在把程序修改了一下，修改之后只有0.25的代碼必須同步運行，現在1/0.25=4，表示我們的程序如果在具有大量CPU的硬件上運行時速度將會比在單核的硬件上快大概4倍。
另一方面，通過Amdahl法則，我們也能根據我們想獲得的提速的目標計算出程序應該的同步代碼的比例。如果我們想要達到100倍的提速，而1/100=0.01，意味著，我們程序同步執行的代碼的數量最多不能超過1%。
總結Amdahl法則我們可以看出，我們通過添加額外CPU來獲得性能提升的最大值取決于程序同步執行部分代碼所占的比例有多小。雖然在實際中，想要計算出這個比例并不總是那么容易，更別說面對一些大型的商業系統應用了，但是Amdahl法則給了我們很重要的啟示，那就是，我們必須非常仔細地去考慮那些必須同步執行的代碼，并且力圖減少這部分代碼。
1.2 對性能的影響
文章寫到這里，我們已經表明這樣一個觀點：增加更多的線程可以提高程序的性能和響應速度。但是另一方面，想要取得這些好處卻并非輕而易舉，也需要付出一些代價。線程的使用對性能的提升也會有所影響。
首先，第一個影響來自線程創建的時候。線程的創建過程中，JVM需要從底層操作系統申請相應的資源，并且在調度器中初始化數據結構，以便決定執行線程的順序。
如果你的線程的數量和CPU的核心數量一樣的話，每個線程都會運行在一個核心上，這樣或許他們就不會經常被打斷了。但是事實上，在你的程序運行的時候，操作系統也會有些自己的運算需要CPU去處理。所以，即使這種情形下，你的線程也會被打斷并且等待操作系統來重新恢復它的運行。當你的線程數量超過CPU的核心數量的時候，情況有可能變得更壞。在這種情況下，JVM的進程調度器會打斷某些線程以便讓其他線程執行，線程切換的時候，剛才正在運行的線程的當前狀態需要被保存下來，以便等下次運行的時候可以恢復數據狀態。不僅如此，調度器也會對它自己內部的數據結構進行更新，而這也需要消耗CPU周期。所有這些都意味著，線程之間的上下文切換會消耗CPU計算資源，因此帶來相比單線程情況下沒有的性能開銷。
多線程程序所帶來的另外一個開銷來自對共享數據的同步訪問保護。我們可以使用synchronized關鍵字來進行同步保護，也可以使用Volatile關鍵字來在多個線程之間共享數據。如果多于一個線程想要去訪問某一個共享數據結構的話，就發生了爭用的情形，這時，JVM需要決定哪個進程先，哪個進程后。如果決定該要執行的線程不是當前正在運行的線程，那么就會發生線程切換。當前線程需要等待，直到它成功獲得了鎖對象。JVM可以自己決定如何來執行這種“等待”，假如JVM預計離成功獲得鎖對象的時間比較短，那JVM可以使用激進等待方法，比如，不停地嘗試獲得鎖對象，直到成功，在這種情況下這種方式可能會更高效，因為比較進程上下文切換來說，還是這種方式更快速一些。把一個等待狀態的線程挪回到執行隊列也會帶來額外的開銷。
因此，我們要盡力避免由于鎖競爭而帶來的上下文切換。下面一節將闡述兩種降低這種競爭發生的方法。
1.3 鎖競爭
像上一節所說的那樣，兩個或更多線程對鎖的競爭訪問會帶來額外的運算開銷，因為競爭的發生逼迫調度器來讓一個線程進入激進等待狀態，或者讓它進行等待狀態而引發兩次上下文切換。有某些情況下，鎖競爭的惡果可以通過以下方法來減輕：
1、減少鎖的作用域；
2、減少需要獲取鎖的頻率；
3、盡量使用由硬件支持的樂觀鎖操作，而不是synchronized；
4、盡量少用synchronized；
5、減少使用對象緩存
1.3.1 縮減同步域
如果代碼持有鎖超過必要的時間，那么可以應用這第一種方法。通常我們可以將一行或多行代碼移出同步區域來降低當前線程持有鎖的時間。在同步區域里運行的代碼數量越少，當前線程就會越早地釋放鎖，從而讓其他線程更早地獲得鎖。這與Amdahl法則相一致的，因為這樣做減少了需要同步執行的代碼量。
為了更好地理解，看下面的源碼：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
public class ReduceLockDuration implements Runnable { private static final int NUMBER_OF_THREADS = 5; private static final Map<String, Integer> map = new HashMap<String, Integer>(); public void run() { for (int i = 0; i < 10000; i++) { synchronized (map) { UUID randomUUID = UUID.randomUUID(); Integer value = Integer.valueOf(42); String key = randomUUID.toString(); map.put(key, value); } Thread.yield(); } } public static void main(String[] args) throws InterruptedException { Thread[] threads = new Thread[NUMBER_OF_THREADS]; for (int i = 0; i < NUMBER_OF_THREADS; i++) { threads[i] = new Thread(new ReduceLockDuration()); } long startMillis = System.currentTimeMillis(); for (int i = 0; i < NUMBER_OF_THREADS; i++) { threads[i].start(); } for (int i = 0; i < NUMBER_OF_THREADS; i++) { threads[i].join(); } System.out.println((System.currentTimeMillis()-startMillis)+"ms"); }} |
在上面的例子中，我們讓五個線程來競爭訪問共享的Map實例，為了在同一時刻只有一個線程可以訪問到Map實例，我們將向Map中添加Key/Value的操作放到了synchronized保護的代碼塊中。當我們仔細察看這段代碼的時候，我們可以看到，計算key和value的幾句代碼并不需要同步執行，key和value只屬于當前執行這段代碼的線程，僅僅對當前線程有意義，并且不會被其他線程所修改。因此，我們可以把這幾句移出同步保護。如下：
|
1
2
3
4
5
6
7
8
9
10
11
|
public void run() { for (int i = 0; i < 10000; i++) { UUID randomUUID = UUID.randomUUID(); Integer value = Integer.valueOf(42); String key = randomUUID.toString(); synchronized (map) { map.put(key, value); } Thread.yield(); }} |
降低同步代碼所帶來的效果是可以測量的。在我的機器上，整個程序的執行時間從420ms降低到了370ms。看看吧，僅僅把三行代碼移出同步保護塊就可以將程序運行時間減少11%。Thread.yield()這句代碼是為了誘發線程上下文切換的，因為這句代碼會告訴JVM當前線程想要交出當前使用的計算資源，以便讓其他等待運行的線程運行。這樣也會帶來更多的鎖競爭的發生，因為，如果不如此的話某一個線程就會更久地占用某個核心繼而減少了線程上下文切換。
1.3.2 分拆鎖
另外一種減少鎖競爭的方法是將一塊被鎖定保護的代碼分散到多個更小的保護塊中。如果你的程序中使用了一個鎖來保護多個不同對象的話，這種方式會有用武之地。假設我們想要通過程序來統計一些數據，并且實現了一個簡單的計數類來持有多個不同的統計指標，并且分別用一個基本計數變量來表示（long類型）。因為我們的程序是多線程的，所以我們需要對訪問這些變量的操作進行同步保護，因為這些操作動作來自不同的線程。要達到這個目的，最簡單的方式就是對每個訪問了這些變量的函數添加synchronized關鍵字。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
public static class CounterOneLock implements Counter { private long customerCount = 0; private long shippingCount = 0; public synchronized void incrementCustomer() { customerCount++; } public synchronized void incrementShipping() { shippingCount++; } public synchronized long getCustomerCount() { return customerCount; } public synchronized long getShippingCount() { return shippingCount; }} |
這種方式也就意味著，對這些變量的每次修改都會引發對其他Counter實例的鎖定。其他線程如果想要對另外一個不同的變量調用increment方法，那也只能等待前一個線程釋放了鎖控制之后才能有機會去完成。在此種情況下，對每個不同的變量使用單獨的synchronized保護將會提高執行效率。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
public static class CounterSeparateLock implements Counter { private static final Object customerLock = new Object(); private static final Object shippingLock = new Object(); private long customerCount = 0; private long shippingCount = 0; public void incrementCustomer() { synchronized (customerLock) { customerCount++; } } public void incrementShipping() { synchronized (shippingLock) { shippingCount++; } } public long getCustomerCount() { synchronized (customerLock) { return customerCount; } } public long getShippingCount() { synchronized (shippingLock) { return shippingCount; } }} |
這種實現為每個計數指標引入了一個單獨synchronized對象，因此，一個線程想要增加Customer計數的時候，它必須等待另一個正在增加Customer計數的線程完成，而并不用等待另一個正在增加Shipping計數的線程完成。
使用下面的類，我們可以非常容易地計算分拆鎖所帶來的性能提升。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
|
public class LockSplitting implements Runnable { private static final int NUMBER_OF_THREADS = 5; private Counter counter; public interface Counter { void incrementCustomer(); void incrementShipping(); long getCustomerCount(); long getShippingCount(); } public static class CounterOneLock implements Counter { ... } public static class CounterSeparateLock implements Counter { ... } public LockSplitting(Counter counter) { this.counter = counter; } public void run() { for (int i = 0; i < 100000; i++) { if (ThreadLocalRandom.current().nextBoolean()) { counter.incrementCustomer(); } else { counter.incrementShipping(); } } } public static void main(String[] args) throws InterruptedException { Thread[] threads = new Thread[NUMBER_OF_THREADS]; Counter counter = new CounterOneLock(); for (int i = 0; i < NUMBER_OF_THREADS; i++) { threads[i] = new Thread(new LockSplitting(counter)); } long startMillis = System.currentTimeMillis(); for (int i = 0; i < NUMBER_OF_THREADS; i++) { threads[i].start(); } for (int i = 0; i < NUMBER_OF_THREADS; i++) { threads[i].join(); } System.out.println((System.currentTimeMillis() - startMillis) + "ms"); }} |
在我的機器上，單一鎖的實現方法平均花費56ms，兩個單獨鎖的實現是38ms。耗時大約降低了大概32%。
另外一種提升方式是，我們甚至可以更進一步地將讀寫分開用不同的鎖來保護。原來的Counter類提供了對計數指標分別提供了讀和寫的方法，但是事實上，讀操作并不需要同步保護，我們可以放心讓多個線程并行讀取當前指標的數值，同時，寫操作必須得到同步保護。java.util.concurrent包里提供了有對ReadWriteLock接口的實現，可以方便地實現這種區分。
ReentrantReadWriteLock實現維護了兩個不同的鎖，一個保護讀操作，一個保護寫操作。這兩個鎖都有獲取鎖和釋放鎖的操作。僅僅當在沒有人獲取讀鎖的時候，寫鎖才能成功獲得。反過來，只要寫鎖沒有被獲取，讀鎖可以被多個線程同時獲取。為了演示這種方法，下面的Counter類使用了ReadWriteLock，如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
|
public static class CounterReadWriteLock implements Counter { private final ReentrantReadWriteLock customerLock = new ReentrantReadWriteLock(); private final Lock customerWriteLock = customerLock.writeLock(); private final Lock customerReadLock = customerLock.readLock(); private final ReentrantReadWriteLock shippingLock = new ReentrantReadWriteLock(); private final Lock shippingWriteLock = shippingLock.writeLock(); private final Lock shippingReadLock = shippingLock.readLock(); private long customerCount = 0; private long shippingCount = 0; public void incrementCustomer() { customerWriteLock.lock(); customerCount++; customerWriteLock.unlock(); } public void incrementShipping() { shippingWriteLock.lock(); shippingCount++; shippingWriteLock.unlock(); } public long getCustomerCount() { customerReadLock.lock(); long count = customerCount; customerReadLock.unlock(); return count; } public long getShippingCount() { shippingReadLock.lock(); long count = shippingCount; shippingReadLock.unlock(); return count; }} |
所有的讀操作都被讀鎖保護，同時，所有的寫操作都被寫鎖所保護。如果程序中執行的讀操作要遠大于寫操作的話，這種實現可以帶來比前一節的方式更大的性能提升，因為讀操作可以并發進行。
1.3.3 分離鎖
上面一個例子展示了如何將一個單獨的鎖分開為多個單獨的鎖，這樣使得各線程僅僅獲得他們將要修改的對象的鎖就可以了。但是另一方面，這種方式也增加了程序的復雜度，如果實現不恰當的話也可能造成死鎖。
分離鎖是與分拆鎖類似的一種方法，但是分拆鎖是增加鎖來保護不同的代碼片段或對象，而分離鎖是使用不同的鎖來保護不同范圍的數值。JDK的java.util.concurrent包里的ConcurrentHashMap即使用了這種思想來提高那些嚴重依賴HashMap的程序的性能。在實現上，ConcurrentHashMap內部使用了16個不同的鎖，而不是封裝一個同步保護的HashMap。16個鎖每一個負責保護其中16分之一的桶位（bucket）的同步訪問。這樣一來，不同的線程想要向不同的段插入鍵的時候，相應的操作會受到不同的鎖來保護。但是反過來也會帶來一些不好的問題，比如，某些操作的完成現在需要獲取多個鎖而不是一個鎖。如果你想要復制整個Map的話，這16個鎖都需要獲得才能完成。
1.3.4 原子操作
另外一種減少鎖競爭的方法是使用原子操作，這種方式會在其他文章中詳細闡述原理。java.util.concurrent包對一些常用基礎數據類型提供了原子操作封裝的類。原子操作類的實現基于處理器提供的“比較置換”功能（CAS），CAS操作只在當前寄存器的值跟操作提供的舊的值一樣的時候才會執行更新操作。
這個原理可以用來以樂觀的方式來增加一個變量的值。如果我們的線程知道當前的值的話，就會嘗試使用CAS操作來執行增加操作。如果期間別的線程已經修改了變量的值，那么線程提供的所謂的當前值已經跟真實的值不一樣了，這時JVM來嘗試重新獲得當前值，并且再嘗試一次，反反復復直到成功為止。雖然循環操作會浪費一些CPU周期，但是這樣做的好處是，我們不需要任何形式的同步控制。
下面的Counter類的實現就利用了原子操作的方式，你可以看到，并沒有使用任何synchronized的代碼。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
public static class CounterAtomic implements Counter { private AtomicLong customerCount = new AtomicLong(); private AtomicLong shippingCount = new AtomicLong(); public void incrementCustomer() { customerCount.incrementAndGet(); } public void incrementShipping() { shippingCount.incrementAndGet(); } public long getCustomerCount() { return customerCount.get(); } public long getShippingCount() { return shippingCount.get(); }} |
與CounterSeparateLock類相比，平均運行時間從39ms降低到了16ms，大約降低了58%。
1.3.5 避免熱點代碼段
一個典型的LIST實現通過會在內容維護一個變量來記錄LIST自身所包含的元素個數，每一次從列表里刪除或增加元素的時候，這個變量的值都會改變。如果LIST在單線程應用中使用的話，這種方式無可厚非，每次調用size()時直接返回上一次計算之后的數值就行了。如果LIST內部不維護這個計數變量的話，每次調用size()操作都會引發LIST重新遍歷計算元素個數。
這種很多數據結構都使用了的優化方式，當到了多線程環境下時卻會成為一個問題。假設我們在多個線程之間共享一個LIST，多個線程同時地去向LIST里面增加或刪除元素，同時去查詢大的長度。這時，LIST內部的計數變量成為一個共享資源，因此所有對它的訪問都必須進行同步處理。因此，計數變量成為整個LIST實現中的一個熱點。
下面的代碼片段展示了這個問題：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
|
public static class CarRepositoryWithCounter implements CarRepository { private Map<String, Car> cars = new HashMap<String, Car>(); private Map<String, Car> trucks = new HashMap<String, Car>(); private Object carCountSync = new Object(); private int carCount = 0; public void addCar(Car car) { if (car.getLicencePlate().startsWith("C")) { synchronized (cars) { Car foundCar = cars.get(car.getLicencePlate()); if (foundCar == null) { cars.put(car.getLicencePlate(), car); synchronized (carCountSync) { carCount++; } } } } else { synchronized (trucks) { Car foundCar = trucks.get(car.getLicencePlate()); if (foundCar == null) { trucks.put(car.getLicencePlate(), car); synchronized (carCountSync) { carCount++; } } } } } public int getCarCount() { synchronized (carCountSync) { return carCount; } }} |
上面這個CarRepository的實現內部有兩個LIST變量，一個用來放洗車元素，一個用來放卡車元素，同時，提供了查詢這兩個LIST總共的大小的方法。采用的優化方式是，每次添加一個Car元素的時候，都會增加內部的計數變量的值，同時增加的操作受synchronized保護，返回計數值的方法也是一樣。
為了避免帶來這種額外的代碼同步開銷，看下面另外一種CarRepository的實現：它不再使用一個內部的計數變量，而是在返回汽車總數的方法里實時計數這個數值。如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
public static class CarRepositoryWithoutCounter implements CarRepository { private Map<String, Car> cars = new HashMap<String, Car>(); private Map<String, Car> trucks = new HashMap<String, Car>(); public void addCar(Car car) { if (car.getLicencePlate().startsWith("C")) { synchronized (cars) { Car foundCar = cars.get(car.getLicencePlate()); if (foundCar == null) { cars.put(car.getLicencePlate(), car); } } } else { synchronized (trucks) { Car foundCar = trucks.get(car.getLicencePlate()); if (foundCar == null) { trucks.put(car.getLicencePlate(), car); } } } } public int getCarCount() { synchronized (cars) { synchronized (trucks) { return cars.size() + trucks.size(); } } }} |
現在，僅僅在getCarCount()方法里，兩個LIST的訪問需要同步保護，像上一種實現那樣每次添加新元素時的同步開銷已經不存在了。
1.3.6 避免對象緩存復用
在JAVA VM的第一版里，使用new關鍵字來創建新對象的開銷比較大，因此，很多開發人員習慣了使用對象復用模式。為了避免一次又一次重復創建對象，開發人員維護一個緩沖池，每次創建完對象實例之后可以把它們保存在緩沖池里，下次其他線程再需要使用的時候就可以直接從緩沖池里去取。
乍一看，這種方式是很合理的，但是這種模式在多線程應用程序里會出現問題。因為對象的緩沖池在多個線程之間共享，因此所有線程在訪問其中的對象時的操作需要同步保護。而這種同步所帶來的開銷已經大過了創建對象本身了。當然了，創建過多的對象會加重垃圾回收的負擔，但是即便把這個考慮在內，避免同步代碼所帶來的性能提升仍然要好過使用對象緩存池的方式。
本文所講述的這些優化方案再一次的表明，每一種可能的優化方式在真正應用的時候一定需要多多仔細評測。不成熟的優化方案表面看起來好像很有道理，但是事實上很有可能會反過來成為性能的瓶頸。
