當前位置：首頁 > 编程语言 > java >内容正文

java

java concurrentmap原理_Java集合番外篇 -- ConcurrentHashMap底层实现和原理

發布時間：2025/3/20 java 35 豆豆

生活随笔收集整理的這篇文章主要介紹了 java concurrentmap原理_Java集合番外篇 -- ConcurrentHashMap底层实现和原理小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

概述

距離上一次集合篇結束已經過了好久了，之前說要寫一下番外，但是太忙了，總也找不出相對松散的時間，也有點靜不下心來，最近花了點時間，于是便有了這篇博客。

在開始之前先介紹一個算法，這個算法和Concurrent的實現是分不開的。

CAS算法：

CAS是英文單詞Compare And Swap的縮寫，翻譯過來就是比較并替換。

CAS機制當中使用了3個基本操作數：內存地址V，舊的預期值A，要修改的新值B。

更新一個變量的時候，只有當變量的預期值A和內存地址V當中的實際值相同時，才會將內存地址V對應的值修改為B

從思想上來說，Synchronized屬于悲觀鎖，悲觀地認為程序中的并發情況嚴重，所以嚴防死守。CAS屬于樂觀鎖，樂觀地認為程序中的并發情況不那么嚴重，所以讓線程不斷去嘗試更新。

ConcurrentHashMap是一個線程安全的Map集合，可以應對高并發的場景，保證線程安全。相比較HashTable，它的鎖粒度更加的細化，因為HashTable的方法都是用Synchronized修飾的，效率灰常的底下。1.8之前ConcurrentHashMap使用鎖分段技術，將數據分成一段段的存儲，每一個數據段配置一把鎖，相互之間不影響，而1.8之后摒棄了Segment(鎖段)的概念，啟用了全新的實現，也就是利用CAS+Synchronized來保證并發更新的安全，底層采用的依然是數組+鏈表+紅黑樹。本篇文章是基于JDK1.8 。

數據結構

繼承關系

public class ConcurrentHashMap

extends AbstractMap

implements ConcurrentMap, Serializable

ConcurrentHashMap 繼承了AbstractMap ,并且實現了ConcurrentMap接口。

與HashMap比對：

相同點：都集成了AbstractMap接口

不同點：HashMap實現了Map接口，ConcurrentHashMap實現了ConcurrentMap接口，而ConcurrentMap繼承了Map接口，使用default關鍵字定義了一些方法。

從繼承關系上看ConcurrentHashMap與HashMap并沒有太大的區別。

基本屬性

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; //鏈表轉為紅黑樹

static final int UNTREEIFY_THRESHOLD = 6; //樹轉列表

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(); //可用處理器數量

重點說一下 sizeCtrl屬性：這個屬性在ConcurrentHashMap中扮演者重要的角色。

//表初始化或者擴容的一個控制標識位

//負數代表正在進行初始化或者擴容的操作

// -1 代表初始化

// -N 代表有n-1個線程在進行擴容操作

//正數或者0表示沒有進行初始化操作，這個數值表示初始化或者下一次要擴容的大小。

//transient 修飾的屬性不會被序列化，volatile保證可見性

private transient volatile int sizeCtl;

構造方法

//無參構造方法，沒有進行任何操作

public ConcurrentHashMap() {}

//指定初始化大小構造方法，判斷參數的合法性，并創建了計算初始化的大小

public ConcurrentHashMap(int initialCapacity) {}

//將指定的集合轉化為ConcurrentHashMap

public ConcurrentHashMap(Map extends K, ? extends V> m) {}

//指定初始化大小和負載因子的構造方法

public ConcurrentHashMap(int initialCapacity, float loadFactor) {

this(initialCapacity, loadFactor, 1);

}

//指定初始化大小，負載因子和concurrentLevel并發更新線程的數量，也可以理解為segment的個數

public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) {}

ConcurrentHashMap的構造方法并沒做太多的工作，主要是進行了參數的合法性校驗，和初始值大小的轉換。這個方法 tableSizeFor()說明一下，主要的功能就是將指定的初始化參數轉換為2的冪次方形式，如果初始化參數為9 ，轉換后初始大小為16 。

內部數據結構

Node

首當其沖，因為它是ConcurrentHashMap的核心，它包裝了key-value的鍵值對，所有插入的數據都包裝在這里面，與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修飾，保證了線程之間的可見性。也不允許調用setValue()方法直接改變Node的值。并增加了find()方法輔助map.get()方法。

TreeNode

樹節點類，另外一個核心的數據結構。當鏈表長度過長的時候，會轉換為TreeNode。但是與HashMap不相同的是，它并不是直接轉換為紅黑樹，而是把這些結點包裝成TreeNode放在TreeBin對象中，由TreeBin完成對紅黑樹的包裝。而且TreeNode在ConcurrentHashMap集成自Node類，而并非HashMap中的集成自LinkedHashMap.Entry類，也就是說TreeNode帶有next指針，這樣做的目的是方便基于TreeBin的訪問。

TreeBin

這個類并不負責包裝用戶的key、value信息，而是包裝的很多TreeNode節點。它代替了TreeNode的根節點，也就是說在實際的ConcurrentHashMap“數組”中，存放的是TreeBin對象，而不是TreeNode對象，這是與HashMap的區別。另外這個類還帶有了讀寫鎖。

ForwardingNode

一個用于連接兩個table的節點類。它包含一個nextTable指針，用于指向下一張表。而且這個節點的key value next指針全部為null，它的hash值為-1. 這里面定義的find的方法是從nextTable里進行查詢節點，而不是以自身為頭節點進行查找

ConcurrentHashMap常用方法

initTable 初始化方法

初始化方法是很重要的一個方法，因為在ConcurrentHashMap的構造方法中只是簡單的進行了一些參數校驗和參數轉換的操作。整個Map的初始化是在插入元素的時候觸發的。這一點在下面的put方法中會進行說明。

//執行初始化操作，單線程操作

private final Node[] initTable() {

Node[] tab; int sc;

while ((tab = table) == null || tab.length == 0) {

if ((sc = sizeCtl) < 0)

//sizeCtl < 0 表示有線程正在進行初始化操作，從運行狀態變為就緒狀態。

Thread.yield(); // lost initialization race; just spin

//設置SIZECTL的值為-1，阻塞其他線程的操作

//該方法有四個參數

//第一個參數：需要改變的對象

//第二個參數：偏移量

//第三個參數：期待的值

//第四個參數：更新后的值

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對象數組

Node[] nt = (Node[])new Node,?>[n];

table = tab = nt;

//sc的值設置為n的0.75倍

sc = n - (n >>> 2); //相當于n*0.75

}

} finally {

sizeCtl = sc; //更改sizeCtl的值

}

break; //中斷循壞返回

}

return tab; //返回初始化的值

}

擴容方法

當ConcurrentHashMap 容量不足的時候，需要對table進行擴容，這個方法是支持多個線程并發擴容的，我們所說的擴容，從本質上來說，無非是從一個數組到另外一個數組的拷貝。

擴容方法分為兩個部分：

創建擴容后的新數組，容量變為原來的兩倍，新數組的創建時單線程完成

將原來的數組元素復制到新的數組中，這個是多線程操作。

//幫助擴容

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")

//構造一個容量是原來兩倍的Node 類型數組

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; //將數組長度賦值給transferIndex

}

int nextn = nextTab.length; //獲取新數組的長度

ForwardingNode fwd = new ForwardingNode(nextTab); //創建fwd節點

boolean advance = true;

boolean finishing = false; // to ensure sweep before committing nextTab

//使用for循環來處理每個槽位中的鏈表元素，CAS設置transferIndex屬性值，并初始化i和bound值

// i 指當前的槽位序號，bound值需要處理的邊界，先處理槽位為15的節點

for (int i = 0, bound = 0;;) {

//創建兩個變量，一個為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已經復制結束

if (finishing) {

nextTable = null; //清空nextTable

table = nextTab; //把nextTab 賦值給 table

sizeCtl = (n << 1) - (n >>> 1); //閾值設置為容量的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算法獲取某個數組節點，為空就設置為fwd

else if ((f = tabAt(tab, i)) == null)

advance = casTabAt(tab, i, null, fwd);

//如果某個節點的hash為-1，跳過

else if ((fh = f.hash) == MOVED)

advance = true; // already processed

else {

//對頭節點加鎖，禁止其他線程進入

synchronized (f) {

if (tabAt(tab, i) == f) {

//構造兩個鏈表，將該節點的列表拆分為兩個部分，一個是原鏈表的排列順序，一個是反序

Node ln, hn;

if (fh >= 0) { // fh 當前節點的hash值若 >= 0

int runBit = fh & n;

Node lastRun = f; //將當前節點賦值給 lastRun 節點

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節點

setTabAt(nextTab, i, ln);

//在nextTab 的 i+n 位置上放置 hn節點

setTabAt(nextTab, i + n, hn);

//在tab節點i位置上插入插入forwardNode節點，表示該節點已經處理

setTabAt(tab, i, fwd);

advance = true;

}

//對TreeBin對象進行處理，過程與上面有些類似

//也把節點分類，分別插入到lo和hi為頭節點的鏈表中

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結構，反向轉換成鏈表結構

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()方法的具體實現：

put()要求鍵值都不能為空

需要經過兩次散列，是數據均勻分散，減少碰撞的次數

判斷tab是否進行了初始化，沒有則調用initTable進行初始化操作(單線程)

數組i的位置沒有元素存在，直接放入

如果i的位置在進行MOVE操作，也就是在進行擴容操作，則多線程幫助擴容

如果i的位置有元素存在，則在該節點加鎖Synchronized，判斷是鏈表還是紅黑樹，按照相應的插入規則插入

final V putVal(K key, V value, boolean onlyIfAbsent) {

//key|value == null 拋出異常

//ConcurrentHashMap不允許鍵或者值為null的這種情況發生

//這一點和HashMap有區別

if (key == null || value == null) throw new NullPointerException();

//散列在散列，讓數據均勻分布，減少碰撞次數

int hash = spread(key.hashCode()); -->static final int spread(int h) {return (h ^ (h >>> 16)) & HASH_BITS;}

int binCount = 0;

//死循環相當于while(true) ,將table賦值給 tab

for (Node[] tab = table;;) {

//創建一個Node類型的變量f , int 類型的變量 n i fh

Node f; int n, i, fh;

//判斷tab是否為null ,是否進行了初始化操作，如果沒有執行初始化，執行初始化操作

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; //退出循環 // 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方法也是最長用的方法，元素放入了，總要取出來

根據傳入的key,獲取相應的hash值

然后判斷當前的table數組是否為空

計算指定的key在table中存儲的位置

鏈表或者紅黑樹轉換相依的方法處理

不存在則返回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 表示紅黑樹節點

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;

}

總結

以上是生活随笔為你收集整理的java concurrentmap原理_Java集合番外篇 -- ConcurrentHashMap底层实现和原理的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇：闭锁java_java多线程学习十:::
下一篇： java字面量方法区_（一）java的