鎖
鎖是用來控制多個線程訪問共享資源的方式，一般來說，一個鎖能夠防止多個線程同時訪問共享資源（但是有些鎖可以允許多個線程并發(fā)的訪問共享資源，比如讀寫鎖）。在Lock接口出現(xiàn)之前，Java程序是靠synchronized關(guān)鍵字實現(xiàn)鎖功能的，而JavaSE5之后，并發(fā)包中新增了Lock接口（以及相關(guān)實現(xiàn)類）用來實現(xiàn)鎖功能，它提供了與synchronized關(guān)鍵字類似的同步功能，只是在使用時需要顯式地獲取和釋放鎖。雖然它缺少了（通過synchronized塊或者方法所提供的）隱式獲取釋放鎖的便捷性，但是卻擁有了鎖獲取與釋放的可操作性、可中斷的獲取鎖以及超時獲取鎖等多種synchronized關(guān)鍵字所不具備的同步特性。
重入鎖ReentrantLock
重入鎖ReentrantLock，顧名思義，就是支持重進入的鎖，它表示該鎖能夠支持一個線程對資源的重復(fù)加鎖。除此之外，該鎖的還支持獲取鎖時的公平和非公平性選擇。ReentrantLock是java.unti.concurrent包下的一個類，它的一般使用結(jié)構(gòu)如下所示：
public void lockMethod() {
ReentrantLock myLock = new ReentrantLock();
myLock.lock();
try{
// 受保護的代碼段
//critical section
} finally {
// 可以保證發(fā)生異常 鎖可以得到釋放 避免死鎖的發(fā)生
myLock.unlock();
}
}
ReentrantLock與synchronized的比較
- 相同：ReentrantLock提供了synchronized類似的功能和內(nèi)存語義。
- 不同：
- ReentrantLock功能性方面更全面，比如時間鎖等候，可中斷鎖等候，鎖投票等，因此更有擴展性。在多個條件變量和高度競爭鎖的地方，用ReentrantLock更合適，ReentrantLock還提供了Condition，對線程的等待和喚醒等操作更加靈活，一個ReentrantLock可以有多個Condition實例，所以更有擴展性。
- ReentrantLock 的性能比synchronized會好點。
- ReentrantLock提供了可輪詢的鎖請求，他可以嘗試的去取得鎖，如果取得成功則繼續(xù)處理，取得不成功，可以等下次運行的時候處理，所以不容易產(chǎn)生死鎖，而synchronized則一旦進入鎖請求要么成功，要么一直阻塞，所以更容易產(chǎn)生死鎖。
公平性
在Java的ReentrantLock構(gòu)造函數(shù)中提供了兩種鎖：創(chuàng)建公平鎖和非公平鎖（默認(rèn)）。代碼如下：
public ReentrantLock() {
sync = new NonfairSync();
}
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
在公平的鎖上，線程按照他們發(fā)出請求的順序獲取鎖，但在非公平鎖上，則允許‘插隊’：當(dāng)一個線程請求非公平鎖時，如果在發(fā)出請求的同時該鎖變成可用狀態(tài)，那么這個線程會跳過隊列中所有的等待線程而獲得鎖。
非公平鎖性能高于公平鎖性能的原因：
在恢復(fù)一個被掛起的線程與該線程真正運行之間存在著嚴(yán)重的延遲。
讀寫鎖ReentrantReadWriteLock
之前提到鎖基本都是排他鎖，這些鎖在同一時刻只允許一個線程進行訪問，而讀寫鎖在同一時刻可以允許多個讀線程訪問，但是在寫線程訪問時，所有的讀線程和其他寫線程均被阻塞。讀寫鎖維護了一對鎖，一個讀鎖和一個寫鎖，通過分離讀鎖和寫鎖，使得并發(fā)性相比一般的排他鎖有了很大提升。
一般情況下，讀寫鎖的性能都會比排它鎖好，因為大多數(shù)場景讀是多于寫的。在讀多于寫
的情況下，讀寫鎖能夠提供比排它鎖更好的并發(fā)性和吞吐量。Java并發(fā)包提供讀寫鎖的實現(xiàn)是ReentrantReadWriteLock
public class Cache {
static Map<String, Object> map = new HashMap<String, Object>();
static ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
static Lock r = rwl.readLock();
static Lock w = rwl.writeLock();
// 獲取一個key對應(yīng)的value
public static final Object get(String key) {
r.lock();
try {
return map.get(key);
} finally {
r.unlock();
}
}
// 設(shè)置key對應(yīng)的value，并返回舊的value
public static final Object put(String key, Object value) {
w.lock();
try {
return map.put(key, value);
} finally {
w.unlock();
}
}
// 清空所有的內(nèi)容
public static final void clear() {
w.lock();
try {
map.clear();
} finally {
w.unlock();
}
}
}
Cache組合一個非線程安全的HashMap作為緩存的實現(xiàn)，同時使用讀寫鎖的
讀鎖和寫鎖來保證Cache是線程安全的。在讀操作get(String key)方法中，需要獲取讀鎖，這使
得并發(fā)訪問該方法時不會被阻塞。寫操作put(String key,Object value)方法和clear()方法，在更新
HashMap時必須提前獲取寫鎖，當(dāng)獲取寫鎖后，其他線程對于讀鎖和寫鎖的獲取均被阻塞，而
只有寫鎖被釋放之后，其他讀寫操作才能繼續(xù)。
Condition接口
Condition是在java 1.5中才出現(xiàn)的，它用來替代傳統(tǒng)的Object的wait()、notify()實現(xiàn)線程間的協(xié)作，相比使用Object的wait()、notify()，使用Condition的await()、signal()這種方式實現(xiàn)線程間協(xié)作更加安全和高效。
調(diào)用Condition的await()和signal()方法，都必須在lock保護之內(nèi)，就是說必須在lock.lock()和lock.unlock之間才可以使用
- Conditon中的await()對應(yīng)Object的wait()
- Condition中的signal()對應(yīng)Object的notify()
- Condition中的signalAll()對應(yīng)Object的notifyAll()
public class ConTest {
final Lock lock = new ReentrantLock();
final Condition condition = lock.newCondition();
public static void main(String[] args) {
// TODO Auto-generated method stub
ConTest test = new ConTest();
Producer producer = test.new Producer();
Consumer consumer = test.new Consumer();
consumer.start();
producer.start();
}
class Consumer extends Thread{
@Override
public void run() {
consume();
}
private void consume() {
try {
lock.lock();
System.out.println("我在等一個新信號"+this.currentThread().getName());
condition.await();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally{
System.out.println("拿到一個信號"+this.currentThread().getName());
lock.unlock();
}
}
}
class Producer extends Thread{
@Override
public void run() {
produce();
}
private void produce() {
try {
lock.lock();
System.out.println("我拿到鎖"+this.currentThread().getName());
condition.signalAll();
System.out.println("我發(fā)出了一個信號："+this.currentThread().getName());
} finally{
lock.unlock();
}
}
}
}
執(zhí)行結(jié)果：
我在等一個新信號Thread-1
我拿到鎖Thread-0
我發(fā)出了一個信號：Thread-0
拿到一個信號Thread-1
并發(fā)容器
CopyOnWrite容器
CopyOnWrite容器即寫時復(fù)制的容器。通俗的理解是當(dāng)我們往一個容器添加元素的時候，不直接往當(dāng)前容器添加，而是先將當(dāng)前容器進行Copy，復(fù)制出一個新的容器，然后新的容器里添加元素，添加完元素之后，再將原容器的引用指向新的容器。這樣做的好處是我們可以對CopyOnWrite容器進行并發(fā)的讀，而不需要加鎖，因為當(dāng)前容器不會添加任何元素。所以CopyOnWrite容器也是一種讀寫分離的思想，讀和寫不同的容器
在使用CopyOnWriteArrayList之前，我們先閱讀其源碼了解下它是如何實現(xiàn)的。以下代碼是向ArrayList里添加元素，可以發(fā)現(xiàn)在添加的時候是需要加鎖的，否則多線程寫的時候會Copy出N個副本出來。
public boolean add(T e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
Object[] elements = getArray();
int len = elements.length;
// 復(fù)制出新數(shù)組
Object[] newElements = Arrays.copyOf(elements, len + 1);
// 把新元素添加到新數(shù)組里
newElements[len] = e;
// 把原數(shù)組引用指向新數(shù)組
setArray(newElements);
return true;
} finally {
lock.unlock();
}
}
final void setArray(Object[] a) {
array = a;
}
讀的時候不需要加鎖，如果讀的時候有多個線程正在向ArrayList添加數(shù)據(jù)，讀還是會讀到舊的數(shù)據(jù)，因為寫的時候不會鎖住舊的ArrayList。
public E get(int index) {
return get(getArray(), index);
}
ConcurrentHashMap的實現(xiàn)原理與使用
ConcurrentHashMap是線程安全且高效的HashMap。ConcurrentHashMap是由Segment數(shù)組結(jié)構(gòu)和HashEntry數(shù)組結(jié)構(gòu)組成。Segment是一種可重入鎖ReentrantLock，在ConcurrentHashMap里扮演鎖的角色，HashEntry則用于存儲鍵值對數(shù)據(jù)。一個ConcurrentHashMap里包含一個Segment數(shù)組，Segment的結(jié)構(gòu)和HashMap類似，是一種數(shù)組和鏈表結(jié)構(gòu)，一個Segment里包含一個HashEntry數(shù)組，每個HashEntry是一個鏈表結(jié)構(gòu)的元素， 每個Segment守護者一個HashEntry數(shù)組里的元素,當(dāng)對HashEntry數(shù)組的數(shù)據(jù)進行修改時，必須首先獲得它對應(yīng)的Segment鎖。
為什么要使用ConcurrentHashMap
- 線程不安全的HashMap
在多線程環(huán)境下，使用HashMap進行put操作會引起死循環(huán)，導(dǎo)致CPU利用率接近100%，所以在并發(fā)情況下不能使用HashMap
- 效率低下的HashTable
HashTable容器使用synchronized來保證線程安全，但在線程競爭激烈的情況下HashTable的效率非常低下。因為當(dāng)一個線程訪問HashTable的同步方法，其他線程也訪問HashTable的同步方法時，會進入阻塞或輪詢狀態(tài)。如線程1使用put進行元素添加，線程2不但不能使用put方法添加元素，也不能使用get方法來獲取元素，所以競爭越激烈效率越低
- ConcurrentHashMap的鎖分段技術(shù)可有效提升并發(fā)訪問率
HashTable容器在競爭激烈的并發(fā)環(huán)境下表現(xiàn)出效率低下的原因是所有訪問HashTable的線程都必須競爭同一把鎖，假如容器里有多把鎖，每一把鎖用于鎖容器其中一部分?jǐn)?shù)據(jù)，那么當(dāng)多線程訪問容器里不同數(shù)據(jù)段的數(shù)據(jù)時，線程間就不會存在鎖競爭，從而可以有效提高并發(fā)訪問效率，這就是ConcurrentHashMap所使用的鎖分段技術(shù)。首先將數(shù)據(jù)分成一段一段地存儲，然后給每一段數(shù)據(jù)配一把鎖，當(dāng)一個線程占用鎖訪問其中一個段數(shù)據(jù)的時候，其他段的數(shù)據(jù)也能被其他線程訪問
hash定位
在定位元素的代碼里我們可以發(fā)現(xiàn)，定位HashEntry和定位Segment的散列算法雖然一樣，都與數(shù)組的長度減去1再相“與”，但是相“與”的值不一樣，定位Segment使用的是元素的hashcode通過再散列后得到的值的高位，而定位HashEntry直接使用的是再散列后的值。其目的是避免兩次散列后的值一樣，雖然元素在Segment里散列開了，但是卻沒有在HashEntry里散列開
hash >>> segmentShift) & segmentMask // 定位Segment所使用的hash算法
int index = hash & (tab.length - 1); // 定位HashEntry所使用的hash算法
get
Segment的get操作實現(xiàn)非常簡單和高效。先經(jīng)過一次再散列，然后使用這個散列值通過散
列運算定位到Segment，再通過散列算法定位到元素，代碼如下
public V get(Object key) {
int hash = hash(key.hashCode());
return segmentFor(hash).get(key, hash);
}
get操作的高效之處在于整個get過程不需要加鎖，除非讀到的值是空才會加鎖重讀。我們
知道HashTable容器的get方法是需要加鎖的，那么ConcurrentHashMap的get操作是如何做到不加鎖的呢？原因是它的get方法里將要使用的共享變量都定義成volatile類型，如用于統(tǒng)計當(dāng)前Segement大小的count字段和用于存儲值的HashEntry的value。定義成volatile的變量，能夠在線程之間保持可見性，能夠被多線程同時讀，并且保證不會讀到過期的值
put
由于put方法里需要對共享變量進行寫入操作，所以為了線程安全，在操作共享變量時必須加鎖。put方法首先定位到Segment，然后在Segment里進行插入操作。插入操作需要經(jīng)歷兩個步驟，第一步判斷是否需要對Segment里的HashEntry數(shù)組進行擴容，第二步定位添加元素的位置，然后將其放在HashEntry數(shù)組里
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
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);
}
Segment的put方法
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
HashEntry<K,V> first = entryAt(tab, index);
for (HashEntry<K,V> e = first;;) {
if (e != null) {
K k;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
e = e.next;
}
else {
if (node != null)
node.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1;
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
return oldValue;
}
size
ConcurrentHashMap的做法是先嘗試2次通過不鎖住Segment的方式來統(tǒng)計各個Segment大小，如果統(tǒng)計的過程中，容器的count發(fā)生了變化，則再采用加鎖的方式來統(tǒng)計所有Segment的大小。
那么ConcurrentHashMap是如何判斷在統(tǒng)計的時候容器是否發(fā)生了變化呢？使用modCount變量，在put、remove和clean方法里操作元素前都會將變量modCount進行加1，那么在統(tǒng)計size前后比較modCount是否發(fā)生變化，從而得知容器的大小是否發(fā)生變化
阻塞隊列
阻塞隊列（BlockingQueue）是一個支持兩個附加操作的隊列。這兩個附加的操作支持阻塞的插入和移除方法。
- 支持阻塞的插入方法：意思是當(dāng)隊列滿時，隊列會阻塞插入元素的線程，直到隊列不
滿。 - 支持阻塞的移除方法：意思是在隊列為空時，獲取元素的線程會等待隊列變?yōu)榉强铡Ｗ枞犃谐Ｓ糜谏a(chǎn)者和消費者的場景，生產(chǎn)者是向隊列里添加元素的線程，消費者是從隊列里取元素的線程。阻塞隊列就是生產(chǎn)者用來存放元素、消費者用來獲取元素的容器
插入和移除操作的4中處理方式
| 方法/處理方式 | 拋出異常 | 返回特殊值 | 一直阻塞 | 超時退出 |
|---|---|---|---|---|
| 插入方法 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
| 移除方法 | remove() | poll() | take() | poll(e,time,unit) |
| 檢查方法 | element() | peek() | 不可用 | 不可用 |
- 拋出異常：當(dāng)隊列滿時，如果再往隊列里插入元素，會拋出IllegalStateException（"Queue
full"）異常。當(dāng)隊列空時，從隊列里獲取元素會拋出NoSuchElementException異常。 - 返回特殊值：當(dāng)往隊列插入元素時，會返回元素是否插入成功，成功返回true。如果是移
除方法，則是從隊列里取出一個元素，如果沒有則返回null。 - 一直阻塞：當(dāng)阻塞隊列滿時，如果生產(chǎn)者線程往隊列里put元素，隊列會一直阻塞生產(chǎn)者
線程，直到隊列可用或者響應(yīng)中斷退出。當(dāng)隊列空時，如果消費者線程從隊列里take元素，隊
列會阻塞住消費者線程，直到隊列不為空。 - 超時退出：當(dāng)阻塞隊列滿時，如果生產(chǎn)者線程往隊列里插入元素，隊列會阻塞生產(chǎn)者線程
一段時間，如果超過了指定的時間，生產(chǎn)者線程就會退出
Java里的阻塞隊列
JDK 7提供了7個阻塞隊列，如下。
- ArrayBlockingQueue：一個由數(shù)組結(jié)構(gòu)組成的有界阻塞隊列。
ArrayBlockingQueue是一個用數(shù)組實現(xiàn)的有界阻塞隊列。此隊列按照先進先出（FIFO）的原則對元素進行排序。默認(rèn)情況下不保證線程公平的訪問隊列
- LinkedBlockingQueue：一個由鏈表結(jié)構(gòu)組成的有界阻塞隊列。
LinkedBlockingQueue是一個用鏈表實現(xiàn)的有界阻塞隊列。此隊列的默認(rèn)和最大長度為Integer.MAX_VALUE。此隊列按照先進先出的原則對元素進行排序。
- PriorityBlockingQueue：一個支持優(yōu)先級排序的無界阻塞隊列。
PriorityBlockingQueue是一個支持優(yōu)先級的無界阻塞隊列。默認(rèn)情況下元素采取自然順序升序排列。也可以自定義類實現(xiàn)compareTo()方法來指定元素排序規(guī)則，或者初始化PriorityBlockingQueue時，指定構(gòu)造參數(shù)Comparator來對元素進行排序。需要注意的是不能保證同優(yōu)先級元素的順序
- DelayQueue：一個使用優(yōu)先級隊列實現(xiàn)的無界阻塞隊列。
DelayQueue是一個支持延時獲取元素的無界阻塞隊列。隊列使用PriorityQueue來實現(xiàn)。隊列中的元素必須實現(xiàn)Delayed接口，在創(chuàng)建元素時可以指定多久才能從隊列中獲取當(dāng)前元素。只有在延遲期滿時才能從隊列中提取元素
- SynchronousQueue：一個不存儲元素的阻塞隊列。
SynchronousQueue是一個不存儲元素的阻塞隊列。每一個put操作必須等待一個take操作，否則不能繼續(xù)添加元素
- LinkedTransferQueue：一個由鏈表結(jié)構(gòu)組成的無界阻塞隊列。
LinkedTransferQueue是一個由鏈表結(jié)構(gòu)組成的無界阻塞TransferQueue隊列。相對于其他阻塞隊列，LinkedTransferQueue多了tryTransfer和transfer方法。
- LinkedBlockingDeque：一個由鏈表結(jié)構(gòu)組成的雙向阻塞隊列。
LinkedBlockingDeque是一個由鏈表結(jié)構(gòu)組成的雙向阻塞隊列。所謂雙向隊列指的是可以從隊列的兩端插入和移出元素。雙向隊列因為多了一個操作隊列的入口，在多線程同時入隊時，也就減少了一半的競爭。相比其他的阻塞隊列，LinkedBlockingDeque多了addFirst、addLast、offerFirst、offerLast、peekFirst和peekLast等方法
