java concurrentmap原理_Java集合番外篇 -- ConcurrentHashMap底层实现和原理
概述
距離上一次集合篇結(jié)束已經(jīng)過了好久了， 之前說要寫一下番外，但是太忙了，總也找不出相對松散的時間，也有點靜不下心來，最近花了點時間，于是便有了這篇博客。
在開始之前先介紹一個算法， 這個算法和Concurrent的實現(xiàn)是分不開的。
CAS算法：
CAS是英文單詞Compare And Swap的縮寫，翻譯過來就是比較并替換。
CAS機制當中使用了3個基本操作數(shù)：內(nèi)存地址V，舊的預(yù)期值A(chǔ)，要修改的新值B。
更新一個變量的時候，只有當變量的預(yù)期值A(chǔ)和內(nèi)存地址V當中的實際值相同時，才會將內(nèi)存地址V對應(yīng)的值修改為B
從思想上來說，Synchronized屬于悲觀鎖，悲觀地認為程序中的并發(fā)情況嚴重，所以嚴防死守。CAS屬于樂觀鎖，樂觀地認為程序中的并發(fā)情況不那么嚴重，所以讓線程不斷去嘗試更新。
ConcurrentHashMap是一個線程安全的Map集合，可以應(yīng)對高并發(fā)的場景，保證線程安全。相比較HashTable，它的鎖粒度更加的細化，因為HashTable的方法都是用Synchronized修飾的，效率灰常的底下。1.8之前ConcurrentHashMap使用鎖分段技術(shù)，將數(shù)據(jù)分成一段段的存儲，每一個數(shù)據(jù)段配置一把鎖，相互之間不影響，而1.8之后摒棄了Segment(鎖段)的概念，啟用了全新的實現(xiàn)，也就是利用CAS+Synchronized來保證并發(fā)更新的安全，底層采用的依然是數(shù)組+鏈表+紅黑樹。本篇文章是基于JDK1.8 。
數(shù)據(jù)結(jié)構(gòu)
繼承關(guān)系
public class ConcurrentHashMap
extends AbstractMap
implements ConcurrentMap, Serializable
ConcurrentHashMap 繼承了AbstractMap ,并且實現(xiàn)了ConcurrentMap接口。
與HashMap比對：
相同點：都集成了AbstractMap接口
不同點：HashMap實現(xiàn)了Map接口，ConcurrentHashMap實現(xiàn)了ConcurrentMap接口，而ConcurrentMap繼承了Map接口，使用default關(guān)鍵字定義了一些方法 。
從繼承關(guān)系上看ConcurrentHashMap與HashMap并沒有太大的區(qū)別。
基本屬性
private static final int MAXIMUM_CAPACITY = 1 << 30; //最大容量2的30次方
private static final int DEFAULT_CAPACITY = 16; //默認容量 1<<4
private static final float LOAD_FACTOR = 0.75f; //負載因子
static final int TREEIFY_THRESHOLD = 8; //鏈表轉(zhuǎn)為紅黑樹
static final int UNTREEIFY_THRESHOLD = 6; //樹轉(zhuǎn)列表
static final int MIN_TREEIFY_CAPACITY = 64; //
private static final int MIN_TRANSFER_STRIDE = 16;
private static int RESIZE_STAMP_BITS = 16;
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
static final int MOVED = -1; // forwarding nodes 的hash值
static final int TREEBIN = -2; // roots of trees 的hash值
static final int RESERVED = -3; // transient reservations 的hash值
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
static final int NCPU = Runtime.getRuntime().availableProcessors(); //可用處理器數(shù)量
重點說一下 sizeCtrl屬性：這個屬性在ConcurrentHashMap中扮演者重要的角色。
//表初始化或者擴容的一個控制標識位
//負數(shù)代表正在進行初始化或者擴容的操作
// -1 代表初始化
// -N 代表有n-1個線程在進行擴容操作
//正數(shù)或者0表示沒有進行初始化操作，這個數(shù)值表示初始化或者下一次要擴容的大小。
//transient 修飾的屬性不會被序列化，volatile保證可見性
private transient volatile int sizeCtl;
構(gòu)造方法
//無參構(gòu)造方法，沒有進行任何操作
public ConcurrentHashMap() {}
//指定初始化大小構(gòu)造方法，判斷參數(shù)的合法性，并創(chuàng)建了計算初始化的大小
public ConcurrentHashMap(int initialCapacity) {}
//將指定的集合轉(zhuǎn)化為ConcurrentHashMap
public ConcurrentHashMap(Map extends K, ? extends V> m) {}
//指定初始化大小和負載因子的構(gòu)造方法
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity, loadFactor, 1);
}
//指定初始化大小，負載因子和concurrentLevel并發(fā)更新線程的數(shù)量，也可以理解為segment的個數(shù)
public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) {}
ConcurrentHashMap的構(gòu)造方法并沒做太多的工作，主要是進行了參數(shù)的合法性校驗，和初始值大小的轉(zhuǎn)換。這個方法 tableSizeFor()說明一下， 主要的功能就是將指定的初始化參數(shù)轉(zhuǎn)換為2的冪次方形式， 如果初始化參數(shù)為9 ，轉(zhuǎn)換后初始大小為16 。
內(nèi)部數(shù)據(jù)結(jié)構(gòu)
Node
首當其沖，因為它是ConcurrentHashMap的核心，它包裝了key-value的鍵值對，所有插入的數(shù)據(jù)都包裝在這里面，與HashMap很相似，但是有一些差別：
static class Node implements Map.Entry {
final int hash;
final K key;
volatile V val;
volatile Node next;
Node(int hash, K key, V val, Node next) {
this.hash = hash;
this.key = key;
this.val = val;
this.next = next;
}
}
value 和 next使用了volatile修飾，保證了線程之間的可見性。也不允許調(diào)用setValue()方法直接改變Node的值。并增加了find()方法輔助map.get()方法。
TreeNode
樹節(jié)點類，另外一個核心的數(shù)據(jù)結(jié)構(gòu)。當鏈表長度過長的時候，會轉(zhuǎn)換為TreeNode。但是與HashMap不相同的是，它并不是直接轉(zhuǎn)換為紅黑樹，而是把這些結(jié)點包裝成TreeNode放在TreeBin對象中，由TreeBin完成對紅黑樹的包裝。而且TreeNode在ConcurrentHashMap集成自Node類，而并非HashMap中的集成自LinkedHashMap.Entry類，也就是說TreeNode帶有next指針，這樣做的目的是方便基于TreeBin的訪問。
TreeBin
這個類并不負責包裝用戶的key、value信息，而是包裝的很多TreeNode節(jié)點。它代替了TreeNode的根節(jié)點，也就是說在實際的ConcurrentHashMap“數(shù)組”中，存放的是TreeBin對象，而不是TreeNode對象，這是與HashMap的區(qū)別。另外這個類還帶有了讀寫鎖。
ForwardingNode
一個用于連接兩個table的節(jié)點類。它包含一個nextTable指針，用于指向下一張表。而且這個節(jié)點的key value next指針全部為null，它的hash值為-1. 這里面定義的find的方法是從nextTable里進行查詢節(jié)點，而不是以自身為頭節(jié)點進行查找
ConcurrentHashMap常用方法
initTable 初始化方法
初始化方法是很重要的一個方法，因為在ConcurrentHashMap的構(gòu)造方法中只是簡單的進行了一些參數(shù)校驗和參數(shù)轉(zhuǎn)換的操作。整個Map的初始化是在插入元素的時候觸發(fā)的。這一點在下面的put方法中會進行說明。
//執(zhí)行初始化操作，單線程操作
private final Node[] initTable() {
Node[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0)
//sizeCtl < 0 表示有線程正在進行初始化操作，從運行狀態(tài)變?yōu)榫途w狀態(tài)。
Thread.yield(); // lost initialization race; just spin
//設(shè)置SIZECTL的值為-1，阻塞其他線程的操作
//該方法有四個參數(shù)
//第一個參數(shù)：需要改變的對象
//第二個參數(shù)：偏移量
//第三個參數(shù)：期待的值
//第四個參數(shù)：更新后的值
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
//再次檢查是否有線程進行了初始化操作
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
//初始化Node對象數(shù)組
Node[] nt = (Node[])new Node,?>[n];
table = tab = nt;
//sc的值設(shè)置為n的0.75倍
sc = n - (n >>> 2); //相當于n*0.75
}
} finally {
sizeCtl = sc; //更改sizeCtl的值
}
break; //中斷循壞返回
}
}
return tab; //返回初始化的值
}
擴容方法
當ConcurrentHashMap 容量不足的時候，需要對table進行擴容，這個方法是支持多個線程并發(fā)擴容的，我們所說的擴容，從本質(zhì)上來說，無非是從一個數(shù)組到另外一個數(shù)組的拷貝。
擴容方法分為兩個部分：
創(chuàng)建擴容后的新數(shù)組，容量變?yōu)樵瓉淼膬杀?，新數(shù)組的創(chuàng)建時單線程完成
將原來的數(shù)組元素復(fù)制到新的數(shù)組中，這個是多線程操作。
//幫助擴容
final Node[] helpTransfer(Node[] tab, Node f) {
Node[] nextTab; int sc;
if (tab != null && (f instanceof ForwardingNode) &&(nextTab = ((ForwardingNode)f).nextTable) != null) {
int rs = resizeStamp(tab.length);
while (nextTab == nextTable && table == tab &&(sc = sizeCtl) < 0) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}
//tab = table ,nextTab 一個Node[]類型的變量
private final void transfer(Node[] tab, Node[] nextTab) {
//n 是tab的長度 ， stride 初始值為0
int n = tab.length, stride;
//判斷cpu處理多線程的能力，如果小于16就直接賦值為16
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
//構(gòu)造一個容量是原來兩倍的Node 類型數(shù)組
Node[] nt = (Node[])new Node,?>[n << 1];
nextTab = nt; //賦值
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab; //賦值
transferIndex = n; //將數(shù)組長度賦值給transferIndex
}
int nextn = nextTab.length; //獲取新數(shù)組的長度
ForwardingNode fwd = new ForwardingNode(nextTab); //創(chuàng)建fwd節(jié)點
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
//使用for循環(huán)來處理每個槽位中的鏈表元素，CAS設(shè)置transferIndex屬性值，并初始化i和bound值
// i 指當前的槽位序號，bound值需要處理的邊界，先處理槽位為15的節(jié)點
for (int i = 0, bound = 0;;) {
//創(chuàng)建兩個變量，一個為Node 類型，一個為int類型
Node f; int fh;
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance = false;
//將transferIndex的值賦值給 nextIndex ,并判斷nextIndex的值是否小于等于0
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
//更新nextIndex的值
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ? nextIndex - stride : 0))) {
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
//
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
//如果table已經(jīng)復(fù)制結(jié)束
if (finishing) {
nextTable = null; //清空nextTable
table = nextTab; //把nextTab 賦值給 table
sizeCtl = (n << 1) - (n >>> 1); //閾值設(shè)置為容量的1.5倍
return;
}
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
finishing = advance = true;
i = n; // recheck before commit
}
}
//CAS算法獲取某個數(shù)組節(jié)點，為空就設(shè)置為fwd
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//如果某個節(jié)點的hash為-1，跳過
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
//對頭節(jié)點加鎖，禁止其他線程進入
synchronized (f) {
if (tabAt(tab, i) == f) {
//構(gòu)造兩個鏈表 ，將該節(jié)點的列表拆分為兩個部分，一個是原鏈表的排列順序，一個是反序
Node ln, hn;
if (fh >= 0) { // fh 當前節(jié)點的hash值 若 >= 0
int runBit = fh & n;
Node lastRun = f; //將當前節(jié)點賦值給 lastRun 節(jié)點
for (Node p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
//差分列表操作
for (Node p = f; p != lastRun; p = p.next) {
int ph = p.hash; K pk = p.key; V pv = p.val;
if ((ph & n) == 0)
ln = new Node(ph, pk, pv, ln);
else
hn = new Node(ph, pk, pv, hn);
}
//在nextTab 的i 位置上放置ln節(jié)點
setTabAt(nextTab, i, ln);
//在nextTab 的 i+n 位置上放置 hn節(jié)點
setTabAt(nextTab, i + n, hn);
//在tab節(jié)點i位置上插入插入forwardNode節(jié)點，表示該節(jié)點已經(jīng)處理
setTabAt(tab, i, fwd);
advance = true;
}
//對TreeBin對象進行處理，過程與上面有些類似
//也把節(jié)點分類，分別插入到lo和hi為頭節(jié)點的鏈表中
//
else if (f instanceof TreeBin) {
TreeBin t = (TreeBin)f;
TreeNode lo = null, loTail = null;
TreeNode hi = null, hiTail = null;
int lc = 0, hc = 0;
for (Node e = t.first; e != null; e = e.next) {
int h = e.hash;
TreeNode p = new TreeNode
(h, e.key, e.val, null, null);
if ((h & n) == 0) {
if ((p.prev = loTail) == null)
lo = p;
else
loTail.next = p;
loTail = p;
++lc;
}
else {
if ((p.prev = hiTail) == null)
hi = p;
else
hiTail.next = p;
hiTail = p;
++hc;
}
}
//如果擴容后 不在需要tree結(jié)構(gòu)，反向轉(zhuǎn)換成鏈表結(jié)構(gòu)
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new TreeBin(lo) : t;
hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
(lc != 0) ? new TreeBin(hi) : t;
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
put方法
put操作是最長用的方法，接下來看一下put()方法的具體實現(xiàn)：
put()要求鍵值都不能為空
需要經(jīng)過兩次散列， 是數(shù)據(jù)均勻分散，減少碰撞的次數(shù)
判斷tab是否進行了初始化，沒有則調(diào)用initTable進行初始化操作(單線程)
數(shù)組i的位置沒有元素存在，直接放入
如果i的位置在進行MOVE操作，也就是在進行擴容操作，則多線程幫助擴容
如果i的位置有元素存在，則在該節(jié)點加鎖Synchronized，判斷是鏈表還是紅黑樹，按照相應(yīng)的插入規(guī)則插入
final V putVal(K key, V value, boolean onlyIfAbsent) {
//key|value == null 拋出異常
//ConcurrentHashMap不允許鍵或者值為null的這種情況發(fā)生
//這一點和HashMap有區(qū)別
if (key == null || value == null) throw new NullPointerException();
//散列在散列， 讓數(shù)據(jù)均勻分布，減少碰撞次數(shù)
int hash = spread(key.hashCode()); -->static final int spread(int h) {return (h ^ (h >>> 16)) & HASH_BITS;}
int binCount = 0;
//死循環(huán) 相當于while(true) ,將table賦值給 tab
for (Node[] tab = table;;) {
//創(chuàng)建一個Node類型的變量f , int 類型的變量 n i fh
Node f; int n, i, fh;
//判斷tab是否為null ,是否進行了初始化操作，如果沒有執(zhí)行初始化，執(zhí)行初始化操作
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//tabAt 獲取值
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//添加到table中
if (casTabAt(tab, i, null,
new Node(hash, key, value, null)))
break; //退出循環(huán) // no lock when adding to empty bin
}
//node的hash值為 -1
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node e = f;; ++binCount) {
K ek;
//key 相等，使用新值替換舊值
if (e.hash == hash &&((ek = e.key) == key ||(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node pred = e;
//放在鏈表的尾部
if ((e = e.next) == null) {
pred.next = new Node(hash, key,value, null);
break;
}
}
}
//紅黑樹替換
else if (f instanceof TreeBin) {
Node p;
binCount = 2;
if ((p = ((TreeBin)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;
}
Get方法
Get方法也是最長用的方法，元素放入了，總要取出來
根據(jù)傳入的key,獲取相應(yīng)的hash值
然后判斷當前的table數(shù)組是否為空
計算指定的key在table中存儲的位置
鏈表或者紅黑樹轉(zhuǎn)換相依的方法處理
不存在則返回null
public V get(Object key) {
Node[] tab; Node e, p; int n, eh; K ek;
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
//eh< 0 表示紅黑樹節(jié)點
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
//鏈表遍歷
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
總結(jié)
以上是生活随笔為你收集整理的java concurrentmap原理_Java集合番外篇 -- ConcurrentHashMap底层实现和原理的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 闭锁java_java多线程学习十:::
- 下一篇： java memcached 存储对象_