review了一次ConcurrentHashMap

古城微笑少年丶 2022-12-04 08:33 149阅读 0赞

# 是什么 #

## 一、介绍 ##

*  支持读取高并发的Map
 *  所有操作支持高并发
 *  任何操作不会锁整表，保护全表获取操作，使之能高并发
 *  效果跟Hashtable一样，但不是使用synchronize
 *  读操作会在修改操作后，遵循heppen-before原则，谁以先操作的动作为准
 *  并发读只能影响全量操作中正在操作的节点
 *  迭代器读取的是创建迭代器之前的那个状态（创建迭代器的时候，始终遍历的是创建迭代器时的hash表）
 *  所以不会抛出并发修改异常（读的都是修改后的结果）
 *  获取长度，是否为空，是否包含值是瞬时态，就是那一个瞬间的结果，结合volatile
 *  到达负载因子.75就会扩容，为啥不能是1呢？因为存在hash冲突的情况
 *  扩容过程可能会比去其他的hash表慢
 *  `concurrencyLevel` 指定并发更新的线程数
 *  这个类跟hashTable一样不支持空值，而hashMap是支持空值的，所有方法参数必须非空
 *  设计目标，并发高效读，高效写
 *  Node类，是key-value映射的一个实例，还有下一个节点
    
     *  TreeNode类是Node的子类，为了实现红黑树
     *  TreeBins是红黑树的根节点，也充当一个锁对象
     *  ForwardingNodes是在扩容时的头节点
     *  当该箱子被迁移了，会在原来的地方放置一个ForwardingNodes
     *  ReservationNodes是占位符节点用于computeIfAbsent
     *  TreeBins/ForwardingNode/ReservationNode三个节点不维护真正的数据
 *  初始化的时候只有一个Node节点，当第一次插入的时候才会扩容
 *  用高位标志来做hash
 *  当第一个插入的时候是CAS
 *  利用每个bin的第一个节点当锁，因为不想浪费空间，此时这个箱子的第一个节点可能是普通节点，如果是普通节点那么直接用synchronize锁对象，如果是TreeBin节点，那么由TreeBin节点充当显示锁，保护树的内容。
 *  为啥可以锁每个箱子？在0.75的负载因子的作用下，实际上每个箱子的节点并不会很多，源码有统计数据。而且并发插入的时候实际要击中同一个箱子并不容易
 *  在hash冲突的时候转为红黑树
 *  当发现表格扩容的时候，将协助其扩容
 *  transferIndex字段减少争用
 *  sizeCtrl防止覆盖
 *  默认容量是16不是8
 *  树化阀值8
 *  非树化阀值6
 *  最小树化容量64，如果有太多链节点，会先扩容再转为树
 *  Node节点跟Entry节点的不同之处是Node是Entry的一个子类，但是Node节点不能被设置值
 *  sizeCtrl:表初始化和调整大小控件
    
     *  表正在初始化或调整大小:-1表示初始化，else -(1 +活动调整大小的线程数)。
     *  否则,当表为null时，保存要使用的初始表大小创建，或者默认为0。初始化后，保存下一个元素计算要调整表大小的值
 *  transferIndex:The next table index (plus one) to split while resizing.
 *  里面提供了视图，键，值，entrySet

## 二、常用属性与方法 ##

1、获取表格中的值的，增改查

static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) { 
            return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
        }
    
        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);
        }
    
        static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) { 
            U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
        }

2、表格的样子

/** * The array of bins. Lazily initialized upon first insertion. * Size is always a power of two. Accessed directly by iterators. */
        transient volatile Node<K,V>[] table;
    
        /** * The next table to use; non-null only while resizing. */
        private transient volatile Node<K,V>[] nextTable;

3、总量记录在哪里?

/** * Base counter value, used mainly when there is no contention, * but also as a fallback during table initialization * races. Updated via CAS. */
        private transient volatile long baseCount;

4、sizeCtl

/** * Table initialization and resizing control. When negative, the * table is being initialized or resized: -1 for initialization, * else -(1 + the number of active resizing threads). Otherwise, * when table is null, holds the initial table size to use upon * creation, or 0 for default. After initialization, holds the * next element count value upon which to resize the table. */
        private transient volatile int sizeCtl;
        //1. -1代表初始化
    	//2. -(1+N)N代表在支持扩容的线程数
    	//3. 下次扩容的数量

5、transferIndex

/**
     * The next table index (plus one) to split while resizing.
     */
    private transient volatile int transferIndex;
    //协助扩容时分配任务的分割点，由大往小减少

6、计数表格

/** * Spinlock (locked via CAS) used when resizing and/or creating CounterCells. */
        private transient volatile int cellsBusy;
    	//计数表格自旋锁
        /** * Table of counter cells. When non-null, size is a power of 2. */
        private transient volatile CounterCell[] counterCells;
    	//计数表格

## 三、方法 ##

### 1、 构造函数 ###

#### 1.1、如果初始化大小如果不是2的幂，那么会转换为2的幂 ####

/** * Creates a new, empty map with an initial table size * accommodating the specified number of elements without the need * to dynamically resize. * * @param initialCapacity The implementation performs internal * sizing to accommodate this many elements. * @throws IllegalArgumentException if the initial capacity of * elements is negative */
        public ConcurrentHashMap(int initialCapacity) { 
            if (initialCapacity < 0)
                throw new IllegalArgumentException();
            int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
                       MAXIMUM_CAPACITY :
                       tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
            this.sizeCtl = cap;
        }
    	//1. tableSizeFor
    
    	/** * Creates a new, empty map with an initial table size based on * the given number of elements ({@code initialCapacity}), table * density ({@code loadFactor}), and number of concurrently * updating threads ({@code concurrencyLevel}). * * @param initialCapacity the initial capacity. The implementation * performs internal sizing to accommodate this many elements, * given the specified load factor. * @param loadFactor the load factor (table density) for * establishing the initial table size * @param concurrencyLevel the estimated number of concurrently * updating threads. The implementation may use this value as * a sizing hint. * @throws IllegalArgumentException if the initial capacity is * negative or the load factor or concurrencyLevel are * nonpositive */
        public ConcurrentHashMap(int initialCapacity,
                                 float loadFactor, int concurrencyLevel) { 
            if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
                throw new IllegalArgumentException();
            if (initialCapacity < concurrencyLevel)   // Use at least as many bins
                initialCapacity = concurrencyLevel;   // as estimated threads
            long size = (long)(1.0 + (long)initialCapacity / loadFactor);
            int cap = (size >= (long)MAXIMUM_CAPACITY) ?
                MAXIMUM_CAPACITY : tableSizeFor((int)size);
            this.sizeCtl = cap;
        }
    	//1. sizeCtl出现在这里

### 2、size ###

#### 2.1、通过计数表格sum获取 ####

/** * {@inheritDoc} */
        public int size() { 
            long n = sumCount();
            return ((n < 0L) ? 0 :
                    (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                    (int)n);
        }
    
    	final long sumCount() { 
            CounterCell[] as = counterCells; CounterCell a;
        	long sum = baseCount;
        	if (as != null) { 
                for (int i = 0; i < as.length; ++i) { 
                    if ((a = as[i]) != null)
                        sum += a.value;
                }
            }
            return sum;
        }

### 3、isEmpty ###

#### 3.1、通过计数表格sum获取 ####

/** * {@inheritDoc} */
        public boolean isEmpty() { 
            return sumCount() <= 0L; // ignore transient negative values
        }

### 4、get ###

public V get(Object key) { 
            Node<K,V>[] tab; Node<K,V> 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为-1，下面三种状态之一，find有三种实现方式 static final int MOVED = -1; // hash for forwarding nodes static final int TREEBIN = -2; // hash for roots of trees static final int RESERVED = -3; // hash for transient reservations **/
                else if (eh < 0)
                    return (p = e.find(h, key)) != null ? p.val : null;//find 方法见下面
      
                //代表是普通的链节点，直接就next了
                while ((e = e.next) != null) { 
                    if (e.hash == h &&
                        ((ek = e.key) == key || (ek != null && key.equals(ek))))
                        return e.val;
                }
            }
            return null;
     	}
    		//ForwardingNodes find 方法，表示改元素已经被迁移了，那么会去下一个表的统一位置查这个元素
    		Node<K,V> find(int h, Object k) { 
                // loop to avoid arbitrarily deep recursion on forwarding nodes
                outer: for (Node<K,V>[] tab = nextTable;;) { 
                    Node<K,V> e; int n;
                    if (k == null || tab == null || (n = tab.length) == 0 ||
                        (e = tabAt(tab, (n - 1) & h)) == null)
                        return null;
                    for (;;) { 
                        int eh; K ek;
                        if ((eh = e.hash) == h &&
                            ((ek = e.key) == k || (ek != null && k.equals(ek))))
                            return e;
                        if (eh < 0) { 
                            if (e instanceof ForwardingNode) { 
                                tab = ((ForwardingNode<K,V>)e).nextTable;
                                continue outer;
                            }
                            else
                                return e.find(h, k);
                        }
                        if ((e = e.next) == null)
                            return null;
                    }
                }
            }
    
    		//TreeBinNodes find 方法，当拥有锁就红黑树查找，当没有锁就线性查找
            /** * Returns matching node or null if none. Tries to search * using tree comparisons from root, but continues linear * search when lock not available. */
            final Node<K,V> find(int h, Object k) { 
                if (k != null) { 
                    for (Node<K,V> e = first; e != null; ) { 
                        int s; K ek;
                        if (((s = lockState) & (WAITER|WRITER)) != 0) { 
                            if (e.hash == h &&
                                ((ek = e.key) == k || (ek != null && k.equals(ek))))
                                return e;
                            e = e.next;
                        }
                        else if (U.compareAndSwapInt(this, LOCKSTATE, s,
                                                     s + READER)) { 
                            TreeNode<K,V> r, p;
                            try { 
                                p = ((r = root) == null ? null :
                                     r.findTreeNode(h, k, null));
                            } finally { 
                                Thread w;
                                if (U.getAndAddInt(this, LOCKSTATE, -READER) ==
                                    (READER|WAITER) && (w = waiter) != null)
                                    LockSupport.unpark(w);
                            }
                            return p;
                        }
                    }
                }
                return null;
            }
    
    		//ReservationNode 占位节点，直接返回为空
            Node<K,V> find(int h, Object k) { 
                // loop to avoid arbitrarily deep recursion on forwarding nodes
                outer: for (Node<K,V>[] tab = nextTable;;) { 
                    Node<K,V> e; int n;
                    if (k == null || tab == null || (n = tab.length) == 0 ||
                        (e = tabAt(tab, (n - 1) & h)) == null)
                        return null;
                    for (;;) { 
                        int eh; K ek;
                        if ((eh = e.hash) == h &&
                            ((ek = e.key) == k || (ek != null && k.equals(ek))))
                            return e;
                        if (eh < 0) { 
                            if (e instanceof ForwardingNode) { 
                                tab = ((ForwardingNode<K,V>)e).nextTable;
                                continue outer;
                            }
                            else
                                return e.find(h, k);
                        }
                        if ((e = e.next) == null)
                            return null;
                    }
                }
            }

#### 4.1、流程图 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hheXJl_size_16_color_FFFFFF_t_70_pic_center]

#### 4.2、记忆方法，到hash表上，表此时可能是链表节点，可能是树节点，可能在迁移。需要根据状态判断查找方法，如果是树节点的判断是否能拿到锁，不能的话降级为线性查找，注意，这里的锁状态是通过位知识来判断的 ####

#### 4.3、为什么当发现是链表的时候不用加锁？为啥发现当前是锁状态的时候可以直接用next获取下一个？因为Node的next字段是volatile的，保证了它的可见性 ####

面试网易的时候，他问过我为什么get加锁没，我回答没有，我需要告诉他为什么没有。因为get的时候

*  如果发现链表那么直接调用next字段拿到下一个内容，next字段是volatile的，保持了他的可见性。
 *  如果发现是红黑树的时候，那么要判断当前的锁状态（通过TreeBin对象来判断），如果被锁了，那么退化成了方法一，如果没被锁的时候，那么就加下锁，使用红黑树的find查找

### 5、Traverser迭代器的实现 ###

/**
             * Advances if possible, returning next valid node, or null if none.
             */
            final Node<K,V> advance() {
                Node<K,V> e;
                if ((e = next) != null)
                    e = e.next;
                for (;;) {
                    Node<K,V>[] t; int i, n;  // must use locals in checks
                    if (e != null)
                        return next = e;
                    if (baseIndex >= baseLimit || (t = tab) == null ||
                        (n = t.length) <= (i = index) || i < 0)
                        return next = null;
                    if ((e = tabAt(t, i)) != null && e.hash < 0) {
                        if (e instanceof ForwardingNode) {//跳到新表
                            tab = ((ForwardingNode<K,V>)e).nextTable;
                            e = null;
                            pushState(t, i, n);
                            continue;
                        }
                        else if (e instanceof TreeBin)
                            e = ((TreeBin<K,V>)e).first;
                        else
                            e = null;
                    }
                    if (stack != null)
                        recoverState(n);//从新表回来
                    else if ((index = i + baseSize) >= n)
                        index = ++baseIndex; // visit upper slots if present
                }
            }

#### 5.1、流程图 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hheXJl_size_16_color_FFFFFF_t_70_pic_center 1]

#### 5.2、不为空直接返回，为空CAS从表上面拿下一个，如果发现元素被迁移，跳到新表拿，再回去旧表找下一个 ####

#### 5.3、哪些方法是基于它实现的 ####

*  containsValue

### 6、put ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hheXJl_size_16_color_FFFFFF_t_70_pic_center 2]

#### 6.1、key跟value都不能为空，记忆点value不能为空 ####

#### 6.2、用自己的话复盘 ####

加入的时候

*  可能哈希表还没初始化，那么就初始化
 *  可能哈希格子为空，那么尝试CAS设置值
 *  可能发现表格在迁移，那么协助扩容
    
     *  检查是否能满足协作扩容的条件
        
         *  线程数是否打满
         *  表格是否扩容完毕
         *  表格是否再度扩容等
     *  分配任务，协助扩容
        
         *  按总的任务数，根据线程数做规划划分
 *  尝试拿到锁，在拿到锁就开始插插插

#### 6.3、其他小小知识点 ####

*  初始化表格时候，如果发现别的线程在初始化的时候，那自己只自旋
 *  发现正在扩容那么会帮助扩容
 *  正常的话会加锁插入
 *  协助扩容的时候有两个关键的变量
    
     *  sizeCtl：大于0表示下一次扩容的容量，小于0时，高16位表示旧数组长度，低16位表示正在扩容的线程数
     *  transferindex：表示下一次开始的节点
 *  锁，迁移的最小粒度都是表格的每个格子，记住

#### 6.4、计数格子存在的必要性。减少并发时的计数冲突，要避免在并发情况下的经常访问。计数格子的最大数量跟CPU的数量是相等的。所以每条线程都在自己的计数格子里面计数。 ####

### 7、remove ###

*  使用replace方法
    
     *  replace方法跟get方法相似，不累述

### 8、keySet&values ###

- 表格是否扩容完毕
    - 表格是否再度扩容等

*  分配任务，协助扩容
    
     *  按总的任务数，根据线程数做规划划分
 *  尝试拿到锁，在拿到锁就开始插插插

#### 6.3、其他小小知识点 ####

### 7、remove ###

*  使用replace方法
    
     *  replace方法跟get方法相似，不累述

### 8、keySet&values ###

*  返回视图

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hheXJl_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/2ac12de45b6f4ed289d3c88d7f54c6ab.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hheXJl_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/ab67990e29ff443eb1af2217cf52fc53.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hheXJl_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/025b49f0965f454c941dee5d0c5c5bcc.png