Pourquoi concurrenthashmap est-il thread-safe ?

1. Comparaison entre HashMap et ConcurrentHashMap

Nous utilisons un morceau de code pour prouver l'insécurité des threads de HashMap et la sécurité des threads de ConcurrentHashMap. La logique du code est très simple. 10 000 threads sont démarrés. Chaque thread effectue une opération simple, qui consiste à mettre une clé puis à supprimer une clé. En théorie, si la sécurité des threads est maintenue, la taille finale de la carte () doit être 0.

Tutoriel recommandé : "Java learning"

Map<Object,Object> myMap=new HashMap<>();
        // ConcurrentHashMap myMap=new ConcurrentHashMap();
        for (int i = 0; i <10000 ; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                        double a=Math.random();
                        myMap.put(a,a);
                        myMap.remove(a);
                }
            }).start();
        }
        Thread.sleep(15000l);//多休眠下，保证上面的线程操作完毕。
        System.out.println(myMap.size());

La taille de la carte est affichée ici=13, mais en fait il y a une clé dans la carte. Nous utilisons ConcurrentHashMap pour exécuter le même code, et le résultat map ==0

Cela prouve que ConcurrentHashMap est thread-safe. Analysons comment ConcurrentHashMap garantit la sécurité des threads à partir du code source. La version jdk. est 1,8.

2. Analyse du code source de ConcurrentHashMap

3.1 Attributs de base de ConcurrentHashMap

//默认最大的容量 
private static final int MAXIMUM_CAPACITY = 1 << 30;
//默认初始化的容量
private static final int DEFAULT_CAPACITY = 16;
//最大的数组可能长度
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
//默认的并发级别，目前并没有用，只是为了保持兼容性
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
//和hashMap一样，负载因子
private static final float LOAD_FACTOR = 0.75f;
//和HashMap一样，链表转换为红黑树的阈值，默认是8
static final int TREEIFY_THRESHOLD = 8;
//红黑树转换链表的阀值，默认是6
static final int UNTREEIFY_THRESHOLD = 6;
//进行链表转换最少需要的数组长度，如果没有达到这个数字，只能进行扩容
static final int MIN_TREEIFY_CAPACITY = 64;
//table扩容时, 每个线程最少迁移table的槽位个数
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;
//值为-1, 当Node.hash为MOVED时, 代表着table正在扩容
static final int MOVED     = -1;
//TREEBIN, 置为-2, 代表此元素后接红黑树
static final int TREEBIN   = -2;
//感觉是占位符，目前没看出来明显的作用
static final int RESERVED  = -3;
//主要用来计算Hash值的
static final int HASH_BITS = 0x7fffffff; 
//节点数组
transient volatile Node<K,V>[] table;
//table迁移过程临时变量, 在迁移过程中将元素全部迁移到nextTable上
private transient volatile Node<K,V>[] nextTable;
//基础计数器
private transient volatile long baseCount;
//table扩容和初始化的标记，不同的值代表不同的含义，默认为0，表示未初始化
//-1: table正在初始化;小于-1，表示table正在扩容；大于0，表示初始化完成后下次扩容的大小
private transient volatile int sizeCtl;
//table容量从n扩到2n时, 是从索引n->1的元素开始迁移, transferIndex代表当前已经迁移的元素下标
private transient volatile int transferIndex;
//扩容时候，CAS锁标记
private transient volatile int cellsBusy;
//计数器表，大小是2次幂
private transient volatile CounterCell[] counterCells;

Ce qui précède sont les attributs de base de ConcurrentHashMap La plupart d'entre nous sont les mêmes que HashMap, mais seulement. ajoutez quelques attributs. Plus tard Analysons comment les attributs ajoutés jouent un rôle.

2.2 Attributs communs de la méthode ConcurrentHashMap

méthode put

final V putVal(K key, V value, boolean onlyIfAbsent) {
        //key和value不允许为null
        if (key == null || value == null) throw new NullPointerException();
        //计算hash值
        int hash = spread(key.hashCode());
        int binCount = 0;
        for (Node<K,V>[] tab = table;;) {
            Node<K,V> f; int n, i, fh;
            //如果table没有初始化，进行初始化
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();
            //计算数组的位置    
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                //如果该位置为空，构造新节点添加即可
                if (casTabAt(tab, i, null,
                             new Node<K,V>(hash, key, value, null)))
                    break;                   // no lock when adding to empty bin
            }//如果正在扩容
            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<K,V> 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<K,V> pred = e;
                                //否则添加到节点的最后面
                                if ((e = e.next) == null) {
                                    pred.next = new Node<K,V>(hash, key,  value, null);
                                    break;
                                }
                            }
                        }//如果节点是树节点，就进行树节点添加操作
                        else if (f instanceof TreeBin) {
                            Node<K,V> p;
                            binCount = 2;
                            if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,alue)) != 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;
                }
            }
        }
        //计数器，采用CAS计算size大小，并且检查是否需要扩容
        addCount(1L, binCount);
        return null;
    }

Nous avons constaté que la logique de la méthode put de ConcurrentHashMap n'est pas très différente de celle de HashMap, principalement à cause de la nouvelle partie de sécurité des threads lors de l'ajout d'éléments, utilisez synchronisé pour garantir la sécurité des threads, puis utilisez les opérations CAS pour calculer la taille. L'ensemble du processus de put est relativement simple. Le résumé est le suivant :

1. Déterminez si la clé et la valeur sont vides, lancez directement une exception.

2. Déterminez si le tableau de table a été initialisé. Sinon, initialisez-le.

3. Calculez la valeur de hachage de la clé. Si la position est vide, construisez directement le nœud et insérez-le.

4. Si le tableau est en cours de développement, entrez la méthode d'expansion de l'aide.

5. Enfin, activez le verrou de synchronisation et effectuez l'opération d'insertion. Si l'option d'écrasement est activée, elle sera écrasée directement, sinon le nœud construit sera ajouté à la fin. de nœuds dépasse le seuil de l'arbre rouge-noir, la conversion de l'arbre rouge-noir sera effectuée. Si le nœud actuel est un nœud d'arborescence, effectuez une opération d'insertion d'arborescence.

6. Enfin, comptez la taille et calculez si une expansion est nécessaire.

Méthode get

public V get(Object key) {
        Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
        //计算hash值
        int h = spread(key.hashCode());
        //如果table已经初始化，并且计算hash值的索引位置node不为空
        if ((tab = table) != null && (n = tab.length) > 0 &&
            (e = tabAt(tab, (n - 1) & h)) != null) {
            //如果hash相等，key相等，直接返回该节点的value
            if ((eh = e.hash) == h) {
                if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                    return e.val;
            }//如果hash值为负值表示正在扩容，这个时候查的是ForwardingNode的find方法来定位到节点。
            else if (eh < 0)
                return (p = e.find(h, key)) != null ? p.val : null;
            //循环遍历链表，查询key和hash值相等的节点。
            while ((e = e.next) != null) {
                if (e.hash == h &&
                    ((ek = e.key) == key || (ek != null && key.equals(ek))))
                    return e.val;
            }
        }
        return null;
    }

La méthode get est relativement simple. Le processus principal est le suivant :

1. Calculez directement la valeur de hachage. les nœuds recherchés sont égaux, renvoyez le nœud directement. Seule la valeur du nœud fera l'affaire.

2. Si la table se développe, appelez la méthode find de ForwardingNode pour trouver le nœud.

3. S'il n'y a pas d'expansion, parcourez simplement la liste chaînée et recherchez la valeur du nœud avec la même clé et la même valeur de hachage.

Extension de ConcurrentHashMap

L'expansion de ConcurrentHashMap est relativement compliquée par rapport à l'expansion de HashMap car elle implique des opérations multithread. La méthode d'expansion ici est principalement le transfert. cette méthode et étudiez-la. Voici comment la développer.

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
        int n = tab.length, stride;
        //保证每个线程扩容最少是16,
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; // subdivide range
        if (nextTab == null) {            // initiating
            try {
            //扩容2倍
                @SuppressWarnings("unchecked")
                Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
                nextTab = nt;
            } catch (Throwable ex) {      // try to cope with OOME
                //出现异常情况就不扩容了。
                sizeCtl = Integer.MAX_VALUE;
                return;
            }
            //用新数组对象接收
            nextTable = nextTab;
            //初始化扩容下表为原数组的长度
            transferIndex = n;
        }
        int nextn = nextTab.length;
        //扩容期间的过渡节点
        ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
        boolean advance = true;
        boolean finishing = false; // to ensure sweep before committing nextTab
        
        for (int i = 0, bound = 0;;) {
            Node<K,V> f; int fh;
            while (advance) {
                int nextIndex, nextBound;
                //如果该线程已经完成了
                if (--i >= bound || finishing)
                    advance = false;
                //设置扩容转移下标，如果下标小于0，说明已经没有区间可以操作了，线程可以退出了
                else if ((nextIndex = transferIndex) <= 0) {
                    i = -1;
                    advance = false;
                }CAS操作设置区间
                else if (U.compareAndSwapInt
                         (this, TRANSFERINDEX, nextIndex,
                          nextBound = (nextIndex > stride ?
                                       nextIndex - stride : 0))) {
                    bound = nextBound;
                    i = nextIndex - 1;
                    advance = false;
                }
            }
            //如果计算的区间小于0了，说明区间分配已经完成，没有剩余区间分配了
            if (i < 0 || i >= n || i + n >= nextn) {
                int sc;
                if (finishing) {//如果扩容完成了，进行收尾工作
                    nextTable = null;//清空临时数组
                    table = nextTab;//赋值原数组
                    sizeCtl = (n << 1) - (n >>> 1);//重新赋值sizeCtl
                    return;
                }//如果扩容还在进行，自己任务完成就进行sizeCtl-1，这里是因为，扩容是通过helpTransfer()和addCount()方法来调用的，在调用transfer()真正扩容之前，sizeCtl都会+1，所以这里每个线程完成后就进行-1。
                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
                }
            }//如果在table中没找到，就用过渡节点
            else if ((f = tabAt(tab, i)) == null)
                //成功设置就进入下一个节点
                advance = casTabAt(tab, i, null, fwd);
            else if ((fh = f.hash) == MOVED)
                //如果节点不为空，并且该位置的hash值为-1，表示已经处理了，直接进入下一个循环即可
                advance = true; // already processed
            else {
            //这里说明老table该位置不为null，也没有被处理过，进行真正的处理逻辑。进行同步锁
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                        Node<K,V> ln, hn;
                        //如果hash值大于0
                        if (fh >= 0) {
                        //为运算结果
                            int runBit = fh & n;
                            Node<K,V> lastRun = f;
                            for (Node<K,V> 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<K,V> p = f; p != lastRun; p = p.next) {
                                int ph = p.hash; K pk = p.key; V pv = p.val;
                                //这里的逻辑和hashMap是一样的，都是采用2个链表进行处理，具体分析可以查看我分析HashMap的文章
                                if ((ph & n) == 0)
                                    ln = new Node<K,V>(ph, pk, pv, ln);
                                else
                                    hn = new Node<K,V>(ph, pk, pv, hn);
                            }
                            setTabAt(nextTab, i, ln);
                            setTabAt(nextTab, i + n, hn);
                            setTabAt(tab, i, fwd);
                            advance = true;
                        }//如果是树节点，执行树节点的扩容数据转移
                        else if (f instanceof TreeBin) {
                            TreeBin<K,V> t = (TreeBin<K,V>)f;
                            TreeNode<K,V> lo = null, loTail = null;
                            TreeNode<K,V> hi = null, hiTail = null;
                            int lc = 0, hc = 0;
                            for (Node<K,V> e = t.first; e != null; e = e.next) {
                                int h = e.hash;
                                TreeNode<K,V> p = new TreeNode<K,V>
                                    (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;
                                }
                            }
                            //这里判断是否进行树转换
                            ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
                                (hc != 0) ? new TreeBin<K,V>(lo) : t;
                            hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                                (lc != 0) ? new TreeBin<K,V>(hi) : t;
                           //这里把新处理的链表赋值到新数组中
                            setTabAt(nextTab, i, ln);
                            setTabAt(nextTab, i + n, hn);
                            setTabAt(tab, i, fwd);
                            advance = true;
                        }
                    }
                }
            }
        }
    }

L'expansion de ConcurrentHashMap est encore relativement compliquée. La complexité se reflète principalement dans le niveau de contrôle de l'expansion multi-thread. D'une part, cela. C'est effectivement plus compliqué. Dans certains endroits, je ne le suis pas. C'est très clair. Par contre, je pense que notre recherche consiste principalement à comprendre ses idées. Tant que nous pouvons comprendre les codes clés et les idées clés. Si nous ne réimplémentons pas un ensemble de fonctions similaires, je ne pense pas que nous ayons besoin de nous soucier de tous les détails. Pour résumer, les étapes d'expansion de ConcurrentHashMap sont les suivantes :

1. Obtenez la taille de l'étape de traitement d'expansion du thread, qui est d'au moins 16, qui est le nombre de nœuds pour qu'un seul thread gère l'expansion.

2. Créez un tableau avec deux fois la capacité d'origine et construisez un nœud de transition pour les opérations de requête pendant l'expansion de la capacité.

3. Effectuez une boucle infinie pour transférer les nœuds, principalement basée sur la variable de finition pour déterminer si l'expansion est terminée. Pendant la période d'expansion, l'opération d'expansion est effectuée en définissant différents index dans le tableau suivant pour différents. threads, c'est-à-dire différents threads, le tableau d'opérations Les segments sont différents et des verrous de synchronisation synchronisés sont utilisés pour verrouiller les nœuds d'exploitation afin de garantir la sécurité des threads.

4. La position réelle du nœud dans le nouveau tableau est la même que la logique d'expansion de HashMap. Grâce à des opérations sur les bits, il est calculé si la nouvelle liste chaînée est à la position d'origine ou à la position d'origine. + la durée de l'expansion Pour une analyse détaillée, vous pouvez me consulter dans cet article.

3.Résumé

1. La plupart du code logique de ConcurrentHashMap est le même que celui de HashMap. Il utilise principalement la somme synchronisée pour assurer la sécurité des threads lors de l'insertion et de l'expansion des nœuds. demandez ici, pourquoi ? Pourquoi utiliser synchronisé ? Au lieu d'utiliser le verrouillage optimiste, ou qu'en est-il du verrouillage ? Personnellement, je pense qu'il y a deux raisons possibles :

a. Le verrouillage optimiste est plus adapté aux scénarios avec moins de conflits de concurrence, cela entraînera des tentatives constantes, ce qui entraînera des performances inférieures.

Après optimisation, les performances de b.synchronized ne sont pas différentes de celles du verrouillage. Dans certains scénarios, elles peuvent être plus rapides que celles du verrouillage. Donc, je pense que c'est la raison pour laquelle nous utilisons synchronisé pour la synchronisation.

2. La logique d'expansion de base de ConcurrentHashMap consiste à allouer différents indices de tableau à différents threads, puis chaque thread traite les nœuds dans leurs plages respectives dans le tableau ci-dessous. Dans le même temps, le nœud de traitement réutilise la logique de hashMap. Grâce au fonctionnement sur bits, vous pouvez connaître la position du nœud après expansion, soit à la position d'origine, soit à la position d'origine + ancienne position, et enfin attribuer directement la valeur. .

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!