前言
本篇文章主要決議juc包下面的并發容器 ConcurrentHashMap,1.7的實作，以及1.8之后做了哪些優化；對比hashmap做的優化，為什么在hashtable的原始碼注釋上推薦使用ConcurrentHashMap對比，
HashMap之實作原理及hash碰撞
HashMap之擴容 資料遷移
ConcurrentHashMap 概述
HashMap 是集合Collection的重要成員，也是Map族最為在作業用到常用的一種，正是因為查找插入等效率高導致它經常被使用，這個原因因此有它的局限性；那就是它不保證多執行緒情況下，資料的安全性，一旦多執行緒操作hashMap，就可能出現執行緒安全問題，例如擴容時死回圈問題；
從jdk1.1開始開發時，作者已經想到這個問題，因此有了hashtable這個類去解決問題，他當時覺得反正對于多執行緒操作map集合我們只要保證他的執行緒安全并不怎么管效率如何，加了synchronized關鍵字看起來至少了解決辦法，要不然你就對多執行緒下操作map自己加鎖，就是效率太低了，
Java 集合深入理解 （十四） ：Hashtable實作原理研究
一直到jdk1.5版本的時候，有了一個質的飛躍，不得不說到一個人，那就是Doug Lea 大神 想到了不使用synchronized關鍵字，使用synchronized關鍵字極不易擴展的，而且不能自己控制，并且鎖粒度太粗了，本來我多執行緒過來都是讀操作，對資料并不沖突，我可以讓他并發進行的，只對寫有進行鎖起來，對于map來，我們本來map就是鏈表陣列操作，將map進行分組，多個執行緒分組去查詢和寫入這是可以并行處理，當然我只說了一部分，例如阻塞 佇列，以及park和unpark,然后atomic 這些，都是juc包下面的實作，供給我們不一樣的想法，我們可以將粒度降的很低，
繼續說說ConcurrentHashMap 這個是jdk1.5 時，juc包提出時，就提出由這個類，當時提出將散串列，進行分組，多執行緒情況，按資料分組控制多執行緒，這在我看來已經挺好了，但是發覺在jdk1.8以后又繼續優化，確實在不斷的進步呀，因為在jdk1.8jvm給jdk提出了Unsafe 自旋操作，也是來源cpu在性能不斷增加，以及記憶體不斷增大，多執行緒在操作資料前做了一步檢查，看老資料是否正確，就是這步操作，在ConcurrentHashMap并不需要分組操作，而只需要自旋操作，效率又做了很大的提升，對于jdk中ConcurrentHashMap來說它認為在使用程序中，不可能資料一直沖突把，如果是一直沖突，那我們自己業務代碼肯定是需要修改的，來保證資料的準確性的，而且如果大量的多執行緒操作資料，大量自旋也是很消耗cpu 和記憶體，可能以后硬體的強度繼續提升到非常大時，有可能又會有新的好演算法好的思想出現，這篇文章主要也是研究ConcurrentHashMap是什么，也會從原始碼去分析，對比著hashmap去看，他們結構是非常相似的，
不外乎在原有操作新增了cas操作
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
Node<K,V> c, Node<K,V> v) {
return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
上圖為整個map的繼承實作圖，ConcurrentHashMap 即實作了ConcurrentMap 提供執行緒安全性和原子性的{@link java.util.Map}保證， 又繼承自AbstractMap 來實作基本map的基本方法 包括put remove size iterator迭代器等基礎方法，
ConcurrentHashMap原始碼分析
jdk1.7中
更新操作之間允許的并發性由可選的concurrencyLevel建構式引數 默認值16（這是來自于一個segments的長度為16），用作內部大小調整的提示，這個表是內部磁區的，以嘗試允許指定的無爭用的并發更新數，因為安置在哈希表本質上是隨機的情況下，實際的并發將變化，理想情況下，您應該選擇一個值來容納盡可能多的物件執行緒將同時修改表，使用遠高于需要的價值可能會浪費空間和時間，并且顯著較低的值可能會導致執行緒爭用，但是 在一個數量級內高估和低估通常不會有太多明顯的影響，
下面的屬性和hashmap中欄位是一致的，用來初始化散串列的大小和閾值，
static final int DEFAULT_INITIAL_CAPACITY = 16;
static final float DEFAULT_LOAD_FACTOR = 0.75f;然后下面獨有的默認的并發級別引數設定，這在1.8過后是取消掉的
/**
* 此表的默認并發級別，在建構式中未另行指定時使用，
*/
static final int DEFAULT_CONCURRENCY_LEVEL = 16;
/**
* 用于在段內建立索引的Shift值，
*/
final int segmentShift;
/** segments中每個元素都是一個專用的hashtable
* The segments, each of which is a specialized hash table.
*/
final Segment<K,V>[] segments;
put方法
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
int hash = hash(key); // 計算Hash值
int j = (hash >>> segmentShift) & segmentMask; //計算下標j
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
s = ensureSegment(j); //若j處有segment就回傳，若沒有就創建并回傳
return s.put(key, hash, value, false); //將值put到segment中去
}所有的操作都是segment 封裝好的物件進行操作，計算出hash值
主要Segment 進行封裝起來，呼叫Segment 的put方法進行添加資料
Segment 從屬性上看，就是一個hashtable.
而segment中put方法和hashmap中方法是一致的，
Segment方法看它怎么去保證資料執行緒安全，既然是juc下面的包，那肯定不會用synchronized，很不容易擴展，這點從繼承方法就能看出，
static final class Segment<K, V> extends ReentrantLock implements Serializable {針對put方法
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
HashEntry<K, V> node = tryLock() ? null : scanAndLockForPut(key, hash, value); // 如果tryLock成功，就回傳null，否則，，，
V oldValue;
try {
HashEntry<K, V>[] tab = table;
int index = (tab.length - 1) & hash; // 根據table陣列的長度 和 hash值計算index小標
HashEntry<K, V> first = entryAt(tab, index); // 找到table陣列在 index處鏈表的頭部
for (HashEntry<K, V> e = first;;) { // 從first開始遍歷鏈表
if (e != null) { // 若e!=null
K k;
if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { // 如果key相同
oldValue = e.value; // 獲取舊值
if (!onlyIfAbsent) { // 若absent=false
e.value = value; // 覆寫舊值
++modCount; //
}
break; // 若已經找到，就退出鏈表遍歷
}
e = e.next; // 若key不相同，繼續遍歷
} else { // 直到e為null
if (node != null) // 將元素放到鏈表頭部
node.setNext(first);
else
node = new HashEntry<K, V>(hash, key, value, first); // 創建新的Entry
int c = count + 1; // count 用來記錄元素個數
if (c > threshold && tab.length < MAXIMUM_CAPACITY) // 如果hashmap元素個數超過threshold，并且table長度小于最大容量
rehash(node); // rehash跟resize的功能差不多,將table的長度變為原來的兩倍，重新打包entries，并將給定的node添加到新的table
else // 如果還有容量
setEntryAt(tab, index, node); // 就在index處添加鏈表節點
++modCount; // 修改運算元
count = c; // 將count+1
oldValue = null; //
break;
}
}
} finally {
unlock(); // 執行完操作后，釋放鎖
}
return oldValue; // 回傳oldValue
}這里就是普通的HashEntry ，
這里保證資料執行緒安全的方法；就是下面的方法， 給物件進行加鎖
HashEntry<K, V> node = tryLock() ? null : scanAndLockForPut(key, hash, value);
這里這樣做的好處還是在于每個沒沖突的資料，保證多執行緒安全性， 散列陣列每一個資料添加鎖，
最后的現象，segment有多少個，并發量就有多少個，
jdk1.8
這里說一點hashmap對于鏈表轉換成紅黑樹和回退條件，
轉換成紅黑樹，鏈表長度要達到8，以及資料數達到64進行轉換；還是很困難的，因為泊松分布來看，一般更容易是擴容，而不是轉換成紅黑樹
紅黑樹轉鏈表，鏈表長度小于6，而發生條件，有兩種一是擴容，也有可能洗掉元素，都有可能導致，
主要利用的casTabAt 方法
if (casTabAt(tab, i, null, //直接用CAS操作，i處的元素
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin 想emptybin中假如元素的時候，不需要加鎖 static final <K, V> boolean casTabAt(Node<K, V>[] tab, int i, Node<K, V> c, Node<K, V> v) {
return U.compareAndSwapObject(tab, ((long) i << ASHIFT) + ABASE, c, v);
}這里對比著hashmap多了自旋操作，多執行緒下，有資料修改失敗的情況
掛載到鏈表上，鎖住頭上存的元素， 添加了synchronized,如果鏈表陣列元素為空時，就直接使用cas去修改資料，做自旋操作
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode()); //計算hash值
int binCount = 0;
for (Node<K,V>[] tab = table;;) { //自旋
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0) //table==null || table.length==0
tab = initTable(); //就initTable
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { //若下標 i 處的元素為null
if (casTabAt(tab, i, null, //直接用CAS操作，i處的元素
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin 想emptybin中假如元素的時候，不需要加鎖
}
else if ((fh = f.hash) == MOVED) //若下標 i 處的元素不為null，且f.hash==MOVED MOVED為常量值-1
tab = helpTransfer(tab, f); //
else { //如果是一般的節點
V oldVal = null;
synchronized (f) { //當頭部元素不為null，且不需要轉換成樹時，需要進行同步操作
if (tabAt(tab, i) == f) {
if (fh >= 0) { //若 鏈表頭部hash值 >=0
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) { //如果key相同
oldVal = e.val;
if (!onlyIfAbsent) //且不為absent
e.val = value; //舊值覆寫新值
break;
}
Node<K,V> pred = e;
if ((e = e.next) == null), { //如果鏈表遍歷完成，還沒退出，說明沒有相同的key存在，在尾部添加節點
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
else if (f instanceof TreeBin) { //如果f是Tree的節點
Node<K,V> p;
binCount = 2;
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}相對于jdk1.7時，提升了沒有hash碰撞時，做了優化使用cas操作，
1.8時，粒度更小了，并發量提高，
總結
整個concurrentHashMap對比各個版本去介紹他怎么達到多執行緒下資料安全，從hashtable開始到concurrenthashmap，體現出了java作者，對運行效率的優化和提升，在作業中根據業務去選擇不同的集合容器，怎么樣達到更快的效率，更好的記憶體，這都是需要研究和討論的
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/302439.html
標籤：其他
上一篇：攻防世界新手WEB