一、提高鎖性能的幾點(diǎn)建議
鎖的競(jìng)爭(zhēng)會(huì)導(dǎo)致程序整體性能的下降，如何降低鎖競(jìng)爭(zhēng)帶來的副作用是我們必須考慮的。下面提出幾點(diǎn)鎖優(yōu)化的建議：
1.1 減小鎖持有時(shí)間
單個(gè)線程對(duì)鎖的持有時(shí)間與系統(tǒng)的性能密切相關(guān)。如果線程持有鎖的時(shí)間越長(zhǎng)，那么鎖的競(jìng)爭(zhēng)程度就會(huì)越激烈。因此，應(yīng)盡可能減少線程對(duì)某個(gè)鎖的占有時(shí)間，進(jìn)而減少線程間互斥的可能。看下面這段代碼：
public synchronized void syncMethod() {
othercode1();
mutexMethod();
othercode2();
}
假設(shè)只有mutexMethod()有同步需要，而othercode1()和othercode2()不需要做同步控制。如果othercode1()和othercode2()都是重量級(jí)的方法，那么就會(huì)花費(fèi)較長(zhǎng)的CPU時(shí)間。改進(jìn)后的代碼如下：
public void syncMethod() {
othercode1();
synchronized(this) {
mutexMethod();
}
othercode2();
}
只對(duì)需要同步的方法進(jìn)行同步控制，這樣鎖的占用時(shí)間會(huì)大大減少，進(jìn)而提高系統(tǒng)的并行性能。
1.2 減小鎖粒度
對(duì)于HashMap來說，最重要的兩個(gè)方法就是get()和put()。一種最自然的的想法就是對(duì)整個(gè)HashMap加鎖，必然可以得到一個(gè)線程安全的對(duì)象。但是這樣做，我們就認(rèn)為加鎖粒度太大。對(duì)于ConcurrentHashMap，它內(nèi)部進(jìn)一步細(xì)分了若干個(gè)小的HashMap，稱之為段（SEGMENT）。默認(rèn)情況下，一個(gè)ConcurrentHashMap被進(jìn)一步細(xì)分為16個(gè)段。
如果需要在ConcurrentHashMap中增加一個(gè)新的表項(xiàng)，并不是將整個(gè)HashMap加鎖，而是首先根據(jù)hashcode得到該表項(xiàng)應(yīng)該被存放到哪個(gè)段中，然后對(duì)該段加鎖，并完成put()操作。在多線程環(huán)境中，如果多個(gè)線程同時(shí)進(jìn)行put()操作，只要被加入的表項(xiàng)不存放在同一個(gè)段中，則線程間便可以做到真正的并行。下面代碼展示了put()操作的過程：
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
//獲取段的序號(hào)
int hash = hash(key);
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
//得到段
s = ensureSegment(j);
return s.put(key, hash, value, false);
}
但是這樣會(huì)存在一個(gè)問題：當(dāng)系統(tǒng)需要取得全局鎖時(shí)，其消耗的資源會(huì)比較多。仍然以ConcurrentHashMap類為例，雖然其put()方法很好地分離了鎖，但是當(dāng)試圖訪問ConcurrentHashMap全局信息時(shí)，就會(huì)需要同時(shí)取得所有段的鎖方能順利實(shí)施。比如ConcurrentHashMap的size()方法，它將返回ConcurrentHashMap的有效表項(xiàng)的數(shù)量，即ConcurrentHashMap的全部有效表項(xiàng)之和。要獲得這個(gè)信息需要取得所有子段的鎖。下面是size()方法的部分代碼：
sum = 0;
for(int i=0; i<segments.length; ++i)
segments[i].lock();
for(int i=0; i<segments.length; ++i)
sum += segments[i].count;
for(int i=0; i<segments.length; ++i)
segments[i].unlock();
可以看到在計(jì)算總數(shù)時(shí)，先要獲得所有段的鎖，然后再求和。但是，ConcurrentHashMap的size()方法并不總是這樣執(zhí)行，事實(shí)上，size()方法會(huì)先使用無鎖的方式求和，如果失敗才會(huì)嘗試這種加鎖的方法。
1.3 讀寫分離鎖來替換獨(dú)占鎖
在讀多寫少的場(chǎng)合，讀寫鎖對(duì)系統(tǒng)性能是很有好處的。因?yàn)槿绻到y(tǒng)在讀寫數(shù)據(jù)時(shí)均使用獨(dú)占鎖，那么讀操作和寫操作間、讀操作和讀操作間、寫操作和寫操作間均不能做到真正的并發(fā)，并且需要相互等待。而讀操作本身不會(huì)影響數(shù)據(jù)的完整性和一致性。因此，理論上，在大部分情況下，應(yīng)該可以允許多線程同時(shí)讀，讀寫鎖正是實(shí)現(xiàn)了這種功能。
1.4 鎖分離
鎖分離是讀寫鎖的而進(jìn)一步延伸。 一個(gè)典型的案例就是java.util.concurrent.LinkedBlockingQueue的實(shí)現(xiàn)。
在LinkedBlockingQueue的實(shí)現(xiàn)中，take()函數(shù)和put()函數(shù)分別實(shí)現(xiàn)了從隊(duì)列中取得數(shù)據(jù)和往隊(duì)列中增加數(shù)據(jù)的功能。雖然兩個(gè)函數(shù)都對(duì)當(dāng)前隊(duì)列進(jìn)行了修改操作，但由于LinkedBlockingQueue是基于鏈表的，因此，兩個(gè)操作分別作用于隊(duì)列的前端和尾端，從理論上說，并不沖突。
如果使用獨(dú)占鎖，則要求在兩個(gè)操作進(jìn)行時(shí)獲取當(dāng)前隊(duì)列的獨(dú)占鎖，那么take()和put()操作就不可能真正的并發(fā)，在運(yùn)行時(shí)，它們會(huì)彼此等待對(duì)方釋放鎖資源。在這種情況下，鎖競(jìng)爭(zhēng)會(huì)相對(duì)比較激烈，從而影響程序在高并發(fā)時(shí)的性能。因此，在JDK的實(shí)現(xiàn)中，并沒有采用這樣的方式，取而代之的是兩把不同的鎖，分離了take()和put()操作。
/** 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();
take()與put()函數(shù)相互獨(dú)立，不存在鎖的競(jìng)爭(zhēng)關(guān)系。只需要在take()和take()間、put()和put()間分別對(duì)takeLock和putLock進(jìn)行競(jìng)爭(zhēng)。從而，削弱了鎖競(jìng)爭(zhēng)的可能性。
函數(shù)take()的實(shí)現(xiàn)如下：
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly(); //不能有兩個(gè)線程同時(shí)取數(shù)據(jù)
try {
while (count.get() == 0) { //如果當(dāng)前沒有可用數(shù)據(jù)，一直等待
notEmpty.await(); //等待，put()操作的通知
}
x = dequeue(); //取得第一個(gè)數(shù)據(jù)
c = count.getAndDecrement(); //數(shù)量減1，原子操作
if (c > 1)
notEmpty.signal(); //通知其他take()操作
} finally {
takeLock.unlock(); //釋放鎖
}
if (c == capacity)
signalNotFull(); //通知put()操作，已有空余空間
return x;
}
函數(shù)put()的實(shí)現(xiàn)如下:
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
int c = -1;
Node<E> node = new Node(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly(); //不能有兩個(gè)線程同時(shí)進(jìn)行put()
try {
while (count.get() == capacity) { //如果隊(duì)列已經(jīng)滿了
notFull.await(); //等待
}
enqueue(node); //插入數(shù)據(jù)
c = count.getAndIncrement(); //更新總數(shù)，變量c是count加1前的值
if (c + 1 < capacity)
notFull.signal(); //有足夠的空間，通知其他線程
} finally {
putLock.unlock(); //釋放鎖
}
if (c == 0)
signalNotEmpty(); //插入成功后，通知take()操作取數(shù)據(jù)
}
1.4 鎖粗化
通常情況下，為了保證多線程間的有效并發(fā)，會(huì)要求每個(gè)線程持有鎖的時(shí)間盡量短，即在使用完公共資源后，應(yīng)該立即釋放鎖。只有這樣，等待在這個(gè)鎖上的其他線程才能盡早地獲得資源執(zhí)行任務(wù)。但是，如果對(duì)同一個(gè)鎖不停地進(jìn)行請(qǐng)求、同步和釋放，其本身也會(huì)消耗系統(tǒng)寶貴的資源，反而不利于性能的優(yōu)化。
為此，虛擬機(jī)在遇到一連串連讀地對(duì)同一鎖不斷進(jìn)行請(qǐng)求和釋放的操作時(shí)，便會(huì)把所有的鎖作整合成對(duì)鎖的一次請(qǐng)求，從而減少對(duì)鎖的請(qǐng)求同步次數(shù)，這個(gè)操作叫做鎖是粗化。比如代碼段：
public void demoMethod() {
synchronized(lock) {
//do sth
}
//做其他不需要的同步的工作，但能很快執(zhí)行完畢
synchronized(lock) {
//do sth
}
}
按照鎖粗化的思想，整合后代碼如下：
public void demoMethod() {
synchronized(lock) {
//do sth
//做其他不需要的同步的工作，但能很快執(zhí)行完畢
}
}
二、Java虛擬機(jī)對(duì)鎖優(yōu)化所做的努力
2.1 偏向鎖
偏向鎖的核心思想是：如果一個(gè)線程獲得了鎖，那么鎖就進(jìn)入了偏向模式。當(dāng)這個(gè)線程再次請(qǐng)求鎖的時(shí)候無需再去做任何同步操作，節(jié)省了鎖的申請(qǐng)操作，提高程序的性能。偏向鎖不適合鎖競(jìng)爭(zhēng)激烈的情況。使用Java虛擬機(jī)參數(shù)-XX:UseBiasedLocking可以開啟偏向鎖。
2.2 輕量級(jí)鎖
如果偏向鎖失敗，虛擬機(jī)并不會(huì)立即掛起線程。它還會(huì)使用一種稱為輕量級(jí)鎖的優(yōu)化手段。輕量級(jí)鎖的操作也很輕便，它只是簡(jiǎn)單地將對(duì)象頭部作為指針，指向持有鎖的線程堆棧的內(nèi)部，來判斷一個(gè)線程是否持有對(duì)象鎖。如果線程獲得輕量級(jí)鎖成功，則可以順利進(jìn)入臨界區(qū)。如果輕量級(jí)鎖加鎖失敗，則表示其他線程搶先爭(zhēng)奪到了鎖，那么當(dāng)前線程的鎖請(qǐng)求就會(huì)膨脹為重量級(jí)鎖。
2.3 自旋鎖
鎖膨脹后，虛擬機(jī)為了避免線程真實(shí)地在操作系統(tǒng)層面掛起，虛擬機(jī)還會(huì)在做最后的努力--自旋鎖。由于當(dāng)前線程暫時(shí)無法獲得鎖，但是什么時(shí)候可以獲得鎖是一個(gè)未知數(shù)。也許在幾個(gè)CPU時(shí)鐘周期后，就可以得到鎖。如果這樣，簡(jiǎn)單粗暴地掛起線程可能是一種得不償失的操作。因此，系統(tǒng)會(huì)進(jìn)行一次賭注：它會(huì)假設(shè)在不久的將來，線程可以得到這把鎖。因此，虛擬機(jī)會(huì)讓當(dāng)前線程做幾個(gè)空循環(huán)，在經(jīng)過若干次循環(huán)后，如果可以得到鎖，那么就順利進(jìn)入臨界區(qū)。如果還不能得到鎖，才會(huì)真實(shí)地將線程在操作系統(tǒng)層面掛起。
2.4 鎖清除
鎖消除是一種更徹底的鎖優(yōu)化。Java虛擬機(jī)在JIT編譯時(shí)，通過對(duì)上下文的掃描，去除不可能存在共享資源競(jìng)爭(zhēng)的鎖。通過鎖消除，可以節(jié)省毫無意義的請(qǐng)求鎖時(shí)間。
例如，在一個(gè)不可能存在并發(fā)競(jìng)爭(zhēng)的場(chǎng)合使用Vector，而Vector內(nèi)部使用了Synchronized請(qǐng)求鎖。比如下面的代碼：
public String[] createStrings() {
Vector<String> v = new Vector<String>();
for(int i=0; i<100; i++) {
v.add(Integer.toString(i));
}
return v.toArray(new String[]{});
}
v屬于線程私有數(shù)據(jù)，不可能被其它線程訪問。這種情況下，Vector內(nèi)部所有加鎖同步都是沒有必要的，虛擬機(jī)檢測(cè)到這種情況就會(huì)將這些無用的鎖清除掉。
三、ThreadLocal
ThreadLocal是一個(gè)本地線程副本變量工具類。主要用于將私有線程和該線程存放的副本對(duì)象做一個(gè)映射，各個(gè)線程之間的變量互不干擾，在高并發(fā)場(chǎng)景下，可以實(shí)現(xiàn)無狀態(tài)的調(diào)用，特別適用于各個(gè)線程依賴不通的變量值完成操作的場(chǎng)景。
3.1 ThreadLocal實(shí)現(xiàn)原理
- 每個(gè)Thread線程內(nèi)部都有一個(gè)Map;
- Map里面存儲(chǔ)線程本地對(duì)象（key）和線程的變量副本（value）
- Thread內(nèi)部的Map是由ThreadLocal維護(hù)的，由ThreadLocal負(fù)責(zé)向map獲取和設(shè)置線程的變量值。
所以對(duì)于不同的線程，每次獲取副本值時(shí)，別的線程并不能獲取到當(dāng)前線程的副本值，形成了副本的隔離，互不干擾。
Thread線程內(nèi)部的Map在類中描述如下：
public class Thread implements Runnable {
/* ThreadLocal values pertaining to this thread. This map is maintained
* by the ThreadLocal class. */
ThreadLocal.ThreadLocalMap threadLocals = null;
}
ThreadLocal如何保證這些對(duì)象只被當(dāng)前線程所訪問，我們需要關(guān)注的是ThreadLocal的set()方法和get()方法。
- get()方法
get()方法用于獲取當(dāng)前線程的副本變量值。
public T get() {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null) {
ThreadLocalMap.Entry e = map.getEntry(this);
if (e != null)
return (T)e.value;
}
return setInitialValue();
}
ThreadLocalMap getMap(Thread t) {
return t.threadLocals;
}
private T setInitialValue() {
T value = initialValue();
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null)
map.set(this, value);
else
createMap(t, value);
return value;
}
protected T initialValue() {
return null;
}
get()方法是先取得當(dāng)前線程的ThreadLocalMap對(duì)象。然后，通過將自己作為key取得內(nèi)部的實(shí)際數(shù)據(jù)。
- set()方法
set()方法用于保存當(dāng)前線程的副本變量值。
public void set(T value) {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null)
map.set(this, value);
else
createMap(t, value);
}
ThreadLocalMap getMap(Thread t) {
return t.threadLocals;
}
void createMap(Thread t, T firstValue) {
t.threadLocals = new ThreadLocalMap(this, firstValue);
}
在set時(shí)，首先獲得當(dāng)前線程對(duì)象，然后通過getMap()拿到線程的ThreadLocalMap，并將值設(shè)入ThreadLocalMap中。
當(dāng)線程退出時(shí)，Thread類會(huì)進(jìn)行一些清理工作，其中就包括清理ThreadLocalMap：
private void exit() {
if (group != null) {
group.threadTerminated(this);
group = null;
}
/* Aggressively null out all reference fields: see bug 4006245 */
target = null;
/* Speed the release of some of these resources */
threadLocals = null;
inheritableThreadLocals = null;
inheritedAccessControlContext = null;
blocker = null;
uncaughtExceptionHandler = null;
}
如果使用線程池，意味著當(dāng)前線程未必會(huì)退出（比如固定大小的線程池，線程總是存在）如果這樣，將一些大大的對(duì)象設(shè)置到ThreadLocal中（它實(shí)際保存在線程持有的threadLocals Map內(nèi)），可能會(huì)使系統(tǒng)出現(xiàn)內(nèi)存泄漏的可能。此時(shí)，如果希望及時(shí)回收對(duì)象，最好使用ThreadLocal.remove()方法將整個(gè)變量移除。
既然Key是弱引用，那么我們要做的事，就是在調(diào)用ThreadLocal的get()、set()方法時(shí)完成后再調(diào)用remove方法，將Entry節(jié)點(diǎn)和Map的引用關(guān)系移除，這樣整個(gè)Entry對(duì)象在GC Roots分析后就變成不可達(dá)了，下次GC的時(shí)候就可以被回收。如果使用ThreadLocal的set方法之后，沒有顯示的調(diào)用remove方法，就有可能發(fā)生內(nèi)存泄露，所以養(yǎng)成良好的編程習(xí)慣十分重要，使用完ThreadLocal之后，記得調(diào)用remove方法。
