淺談Java源碼ConcurrentHashMap

目錄

一、記錄形式

打算直接把過(guò)程寫(xiě)在源碼中，會(huì )按序進(jìn)行注釋?zhuān)殚喌臅r(shí)候可以按序號只看注釋部分

二、ConcurrentHashMap

直接模擬該類(lèi)的使用過(guò)程，從而一步步看其怎么運作的吧，當然最好還是帶著(zhù)問(wèn)題一遍思考一遍總結會(huì )比較好，我閱讀源碼的時(shí)候帶著(zhù)以下幾個(gè)問(wèn)題

• 并發(fā)體現在哪里？怎么保證線(xiàn)程安全的
• 怎么擴容的？擴容是怎么保證線(xiàn)程安全的？
• 怎么put的？put是怎么保證線(xiàn)程安全的？
• 用了哪些鎖？這些鎖的作用是什么？
• 需要留意哪些關(guān)鍵點(diǎn)？

我們最簡(jiǎn)單地使用方法是怎么樣的？

• new一個(gè)ConcurrentHashMap對象
• 調用put方法放入k-v對
• 調用get方法獲取k-v對

那么很顯然，考慮只有在進(jìn)行修改與更新時(shí)需要考慮并發(fā)，所以我關(guān)注的重點(diǎn)在put方法與擴容上

首先new一個(gè)對象時(shí)，我們傳參入一個(gè)size，調用其只有一個(gè)參數的構造器

public ConcurrentHashMap(int initialCapacity) {
        if (initialCapacity < 0)
            throw new IllegalArgumentException();
        // 1、判斷傳入的大小是否超過(guò)最大值的一半，若超過(guò)則取最大值
        // 2、若沒(méi)超過(guò)，則調用tableSizeFor
        // 3、tableSizeFor可以讓size為2的倍數，為什么要是2的倍數呢？因為對hash取模的時(shí)候
        // 用的是位運算，只有size為2的倍數才能這么做
        int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
                   MAXIMUM_CAPACITY :
                   tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
        this.sizeCtl = cap;
    }

// 1、判斷傳入的大小是否超過(guò)最大值的一半，若超過(guò)則取最大值
// 2、若沒(méi)超過(guò)，則調用tableSizeFor
// 3、tableSizeFor讓size為2的倍數，為什么要是2的倍數呢？因為對hash取模的時(shí)候
// 用的是位運算，只有size為2的倍數才能這么做

注意，此時(shí)并沒(méi)有創(chuàng )建map數據結構，所以ConcurrentHashMap是懶惰創(chuàng )建的

接著(zhù)我們調用put方法放入數據，put方法調的putVal

final V putVal(K key, V value, boolean onlyIfAbsent) {
   		// 1、k-v都不能為空，不然拋異常
        if (key == null || value == null) throw new NullPointerException();
        // 2、獲取key的hashcode的hash值
        int hash = spread(key.hashCode());
        // 3、使用binCount來(lái)統計鏈表有多少個(gè)元素的，便于后面判斷是否需要變成紅黑樹(shù)
        int binCount = 0;
        // 4、創(chuàng  )建tab臨時(shí)變量，并賦值為table，由于還沒(méi)初始化，值為null
        // 這里注意了，table的類(lèi)型是Node數組，這個(gè)Node其實(shí)就是Map.Entry<K,V>
        for (Node<K,V>[] tab = table;;) {
            Node<K,V> f; int n, i, fh;
            // 5、判斷tab為空后才進(jìn)行初始化，初始化完成后重新進(jìn)入循環(huán)
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();
            // 6、此時(shí)已經(jīng)初始化好了，可以開(kāi)始插入數據。
            // 這里用tabAt(tab,i)獲取tab的第i個(gè)下標上的鏈表指針
            // 注意了，（n-1）& hash其實(shí)就是hash%n，只有n為2的次方才能這么做
            // 位運算可以提升效率
            // 7、f就是獲取到的第i個(gè)位置的鏈表頭指針
            // 如果為null說(shuō)明什么都沒(méi)有，可以嘗試插入元素
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            	// 8、考慮插入那就要考慮并發(fā)了，casTab表示用cas方法進(jìn)行插入
            	// 插入一個(gè)新的Node結點(diǎn)，這個(gè)是能夠保證線(xiàn)程安全的
            	// 我們知道保證線(xiàn)程安全除了用cas之外還不夠，還需要保證操作對象的可見(jiàn)性
            	// 在這里是對tab進(jìn)行操作，tab在前面用table賦值，而table是加了volatile的
            	// 所以沒(méi)毛病哈
                if (casTabAt(tab, i, null,
                             new Node<K,V>(hash, key, value, null)))
                    break;                   // no lock when adding to empty bin
            }
            // 9、如果f不為空，并且f.hash==MOVED(-1)，說(shuō)明當前這個(gè)位置正在被移動(dòng)
            // 說(shuō)明有線(xiàn)程在擴容，那么當然不能這時(shí)候還去插入了，這里調用helpTransfer去幫助擴容
            // 注意了，這意味著(zhù)擴容時(shí)是一個(gè)一個(gè)位置來(lái)移動(dòng)的，每移動(dòng)一個(gè)就將f.hash改成MOVED
            // 也就意味著(zhù)如果當前線(xiàn)程想要操作的位置還沒(méi)有被移動(dòng)時(shí)是可以操作的，這使得并發(fā)度更高了
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            // 10、如果f.hash表示沒(méi)有被移動(dòng)，且f不為null說(shuō)明可以這個(gè)位置已經(jīng)有東西了
            // 需要對其遍歷，并找到合適的位置插入
            else {
                V oldVal = null;
                // 11、由于要進(jìn)行插入了，所以得加鎖，注意了哈，這里用的synchronized
                // 并且鎖住的是當前位置對象（不一定是鏈表也可能是樹(shù)）
                // 意味著(zhù)除了當前線(xiàn)程，其他線(xiàn)程都不準操作了哈
                // 如果這時(shí)候正在擴容的線(xiàn)程擴到這里了，會(huì )被阻塞的哈
                synchronized (f) {
                	// 確定f為起點(diǎn)
                    if (tabAt(tab, i) == f) {
                    	// 12、判斷f.hash是大于0，大于0表示當前這個(gè)東西是鏈表
                    	// 下面執行鏈表的更新操作
                        if (fh >= 0) {
                            binCount = 1;
                            // 13、接著(zhù)就是具體的遍歷鏈表，查找是否對應值是否存在
                            // 如果遍歷完都找不到，那么就在尾部插入新的結點(diǎn)
                            // 注意了哈這就是尾插
                            // 14、同時(shí)，每遍歷一個(gè)結點(diǎn)還要累加binCount
                            // 即統計一下當前鏈表個(gè)數，便于后面轉紅黑樹(shù)判斷
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;
                                if (e.hash == hash &&
                                    ((ek = e.key) == key ||
                                     (ek != null && key.equals(ek)))) {
                                    oldVal = e.val;
                                    if (!onlyIfAbsent)
                                        e.val = value;
                                    break;
                                }
                                Node<K,V> pred = e;
                                if ((e = e.next) == null) {
                                    pred.next = new Node<K,V>(hash, key,
                                                              value, null);
                                    break;
                                }
                            }
                        }
                        // 13、如果f對應的是樹(shù)結構，那就執行樹(shù)的更新方法
                        else if (f instanceof TreeBin) {
                            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;
                            }
                        }
                    }
                }
                // 14、前面說(shuō)了哈，binCount就是鏈表元素個(gè)數，接著(zhù)就判斷是否大于閾值
                // 大于則轉樹(shù)，可以看這個(gè)閾值TREEIFY_THRESHOLD=8
                if (binCount != 0) {
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        // 15、addCount是讓size+1的
        // 這里要注意，加法也是分多步的，需要先get才能＋，因此為了保證線(xiàn)程安全也是需要用cas的
        // 這里加size的方式并不是直接往size上加，因為size是經(jīng)常修改的，如果經(jīng)常訪(fǎng)問(wèn)的話(huà)效率很低
        // 那么做法和LongAdder這一原子累加器類(lèi)似，用一個(gè)CountCell數組，每個(gè)線(xiàn)程只操作數組中的
        // 某一個(gè)Cell，最后匯總即可
        addCount(1L, binCount);
        return null;
    }

總結一下put的過(guò)程

1.先判斷map是否創(chuàng )建，沒(méi)創(chuàng )建則先創(chuàng )建，結構為 Node<K,V>[ ] extend Map.Entry

2.接著(zhù)找key應該放在哪個(gè)位置 i

3.判斷該位置是否為空，為空則使用CAS插入一個(gè)新的Node

4.不為空則判斷當前位置狀態(tài)是否為MOVED，是則說(shuō)明當前位置正在被其他線(xiàn)程改動(dòng)，當前線(xiàn)程需要幫助完成移動(dòng)

5.不為空且不為MOVED，則判斷是鏈表還是樹(shù)，分別執行對應的更新方法

6.如果為鏈表

• 先對鏈表上鎖，用syn
• 則遍歷并查找是否已存在
• 找到最后都不存在則直接尾插
• 同時(shí)統計鏈表上的元素

7.判斷鏈表元素是否超過(guò)變成樹(shù)的閾值 8 ，超過(guò)則直接變成樹(shù)，變成樹(shù)也是加syn

8.使用addCount更新size，更新方式類(lèi)似LongAdder

三、關(guān)鍵點(diǎn)

• 懶惰加載map
• 對當前位置操作之前需要判斷當前位置的存儲的內容是否被其他線(xiàn)程移動(dòng)了，如果被移動(dòng)則先去幫助完成移動(dòng)

執行擴容移動(dòng)的線(xiàn)程是挨個(gè)移動(dòng)每個(gè)位置的鏈表，移動(dòng)前會(huì )先將當前位置的狀態(tài)用CAS改成MOVED

注意了這個(gè)是提升并發(fā)度的關(guān)鍵所在

因為插入和移動(dòng)（擴容）的粒度是細化到每個(gè)位置的鏈表上

意味著(zhù)擴容和插入可以同時(shí)進(jìn)行

意味著(zhù)插入時(shí)上鎖后，擴容線(xiàn)程執行到該位置需要阻塞

而不是直接整個(gè)map都鎖住

• 當前位置沒(méi)內容時(shí)，通過(guò)CAS插入新Node，而操作鏈表時(shí)用的是syn

如果面試問(wèn)到ConcurrentHashMap中用了什么鎖就心中有數了

• 更新size的時(shí)候用的是LongAdder類(lèi)似的方法

有一個(gè)CountCell數組，每個(gè)線(xiàn)程更新后，對數組中的某個(gè)Cell+1

如果沒(méi)有競爭則只有一個(gè)baseCell，對其+1

統計size時(shí)匯總即可

再細化一下前面的流程
思考初始化map的時(shí)候怎么保證線(xiàn)程安全？防止多個(gè)線(xiàn)程同時(shí)初始化？
答案在initTable方法中

• 可以看到，sizeCtl如果是負數說(shuō)明正在擴容或者初始化
• 因此當需要初始化時(shí)，就去CAS地改變sizeCtl，將其變?yōu)樨摂?/li>
• 哪個(gè)線(xiàn)程CAS成功，則可以執行初始化，這就保證了線(xiàn)程安全
• 可以再去看看，sizeCtl是volatile修飾的哈
• 并且SIZECTL是sizeCtl的offset，這些都是原子類(lèi)類(lèi)似的東西了

private final Node<K,V>[] initTable() {
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
            if ((sc = sizeCtl) < 0)
                Thread.yield(); // lost initialization race; just spin
            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<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = tab = nt;
                        sc = n - (n >>> 2);
                    }
                } finally {
                    sizeCtl = sc;
                }
                break;
            }
        }
        return tab;
    }

get方法就不說(shuō)了，因為不涉及并發(fā)，就是查找而已
感覺(jué)差不多了，把put方法搞清楚了，ConcurrentHashMap怎么解決線(xiàn)程安全的也清楚了，并且并發(fā)關(guān)鍵點(diǎn)在哪也清楚了

到此這篇關(guān)于淺談Java源碼ConcurrentHashMap的文章就介紹到這了,更多相關(guān)Java ConcurrentHashMap內容請搜索腳本之家以前的文章或繼續瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！