增加了ConcurrentHashMap1.7的文档

后端/JAVA高阶/HashMap/第三节ConcurrentHashMap1.7.md

@@ -0,0 +1,252 @@
+## CAS与volatile关键字回顾
+
+### volatile
+　一旦一个共享变量（类的成员变量、类的静态成员变量）被volatile修饰之后，那么就具备了两层语义：
+1）	保证了不同线程对这个变量进行操作时的可见性，即一个线程修改了某个变量的值，这新值对其他线程来说是立即可见的。
+2）	禁止进行指令重排序。
+
+### CAS
+
+CAS机制当中使用了3个基本操作数：内存地址V，旧的预期值A，要修改的新值B。更新一个变量的时候，只有当变量的预期值A和内存地址V当中的实际值相同时，才会将内存地址V对应的值修改为B。
+
+## hashtable底层实现
+
+底层是synchronized锁实现，put和get方法不能同时实现。有可能阻塞线程。并且如果有多个线程执行put操作，会使得操作变成单线程。
+
+## 实现原理
+
+![image-20211109155715277](../../../照片/image-20211109155715277.png)
+将一个大的ConcurrentHashMap集合，拆分成n多个不同的小的hashtable，在每个小的hashtable中都有自己独立的table数组。
+
+每一个segment对象就是一个hashtable，存放的时候会计算两次index值
+
+hashtable的个数即segment的个数不会新增，只会扩容自己独立的table
+
+## 核心参数分析
+
+1. static final int DEFAULT_INITIAL_CAPACITY = 16;  
+segment集合的默认大小
+2. static final int DEFAULT_CONCURRENCY_LEVEL = 16;
+并发的数量，分成segment[16]
+3. static final float DEFAULT_LOAD_FACTOR = 0.75f;
+segment的加载因子，就是table的加载因子
+
+## 无参构造源码分析
+
+```
+    public ConcurrentHashMap(int initialCapacity,
+                             float loadFactor, int concurrencyLevel) {
+        if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
+            throw new IllegalArgumentException();
+        // 并发级别的最大值
+        if (concurrencyLevel > MAX_SEGMENTS)
+            concurrencyLevel = MAX_SEGMENTS;
+        // Find power-of-two sizes best matching arguments
+        // 计算出ssize2的平法的个数
+        int sshift = 0;
+        // segment数组容量
+        int ssize = 1;
+        while (ssize < concurrencyLevel) {
+            ++sshift;
+            ssize <<= 1;
+        }
+        // sshift=4 ssize=16
+        // 计算index 右移动28位置
+        this.segmentShift = 32 - sshift; // 28
+        // 与运算均匀分布
+        this.segmentMask = ssize - 1; // 15
+        if (initialCapacity > MAXIMUM_CAPACITY)
+            initialCapacity = MAXIMUM_CAPACITY;
+        // c实际上就是hashtable的初始化容量   
+        int c = initialCapacity / ssize; // 1
+        if (c * ssize < initialCapacity)
+            ++c;
+        // hashtable默认容量为2
+        int cap = MIN_SEGMENT_TABLE_CAPACITY;
+        while (cap < c)
+            cap <<= 1;
+        // create segments and segments[0]
+        // 创建一个segment s0对象，放入下标为0的位置。默认大小为2，默认扩容大小为1，加载因子为0，75f
+        // 为什么在构造函数初始化s0？  方便后期其他key落到不同的segment中，能够知道加载因子，和默认容量一些基本参数，就是相当于提供了一个模板
+        Segment<K,V> s0 =
+            new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
+                             (HashEntry<K,V>[])new HashEntry[cap]);
+        Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
+        // ss[0] = s0
+        UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
+        this.segments = ss;
+    }
+```
+
+## 1.7put源码解析
+
+```
+    @SuppressWarnings("unchecked")
+    public V put(K key, V value) {
+        Segment<K,V> s;
+        if (value == null)
+            throw new NullPointerException();
+        // 获取key的hash值
+        int hash = hash(key);
+        // 右移动28位
+        /**
+        *  this.segmentShift = 32 - sshift; // 28
+        *  this.segmentMask = ssize - 1; // 15
+        */
+        int j = (hash >>> segmentShift) & segmentMask;
+        // s = Segment[j] == null
+        if ((s = (Segment<K,V>)UNSAFE.getObject          
+             (segments, (j << SSHIFT) + SBASE)) == null)
+            // 如果为null，则创建一个segment对象
+            s = ensureSegment(j);
+        return s.put(key, hash, value, false);
+    }
+```
+
+
+## ensureSegment函数解析
+```
+    @SuppressWarnings("unchecked")
+    private Segment<K,V> ensureSegment(int k) {
+        final Segment<K,V>[] ss = this.segments;
+        long u = (k << SSHIFT) + SBASE; // raw offset
+        Segment<K,V> seg;
+        // 强制读取主内存的数据
+        // 查询segment是否为空？，如果为空，拿到模板
+        if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
+            Segment<K,V> proto = ss[0]; // use segment 0 as prototype
+            int cap = proto.table.length;
+            float lf = proto.loadFactor;
+            int threshold = (int)(cap * lf);
+            HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
+            if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
+                == null) { // recheck
+                Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
+                while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
+                       == null) {
+                    if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
+                        break;
+                }
+            }
+        }
+        return seg;
+    }
+```
+为什么会执行3次这个判断？    就是为了线程安全
+seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null
+
+第一次判断==null
+获取s0默认参数信息，创建table对象HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
+第二个判断==null
+创建Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);对象
+第三次判断==null
+通过CAS操作赋值让segment[index] = ss
+
+## 深度解析segment的put方法
+
+解析``` return s.put(key, hash, value, false);```这一句话
+
+```
+final V put(K key, int hash, V value, boolean onlyIfAbsent) {
+    HashEntry<K,V> node = tryLock() ? null :
+            scanAndLockForPut(key, hash, value);
+    V oldValue;
+    try {
+        HashEntry<K,V>[] tab = table;
+        // 计算hashtable的下标
+        int index = (tab.length - 1) & hash;
+        HashEntry<K,V> first = entryAt(tab, index);
+        for (HashEntry<K,V> e = first;;) {
+            if (e != null) {
+                K k;
+                // 是链表，找到要存放的位置
+                if ((k = e.key) == key ||
+                        (e.hash == hash && key.equals(k))) {
+                    oldValue = e.value;
+                    if (!onlyIfAbsent) {
+                        e.value = value;
+                        ++modCount;
+                    }
+                    break;
+                }
+                e = e.next;
+            }
+            else {
+           		// node节点为null
+                if (node != null)
+                    node.setNext(first);
+                else
+                	// 头插法
+                    node = new HashEntry<K,V>(hash, key, value, first);
+                // 判断是否扩容
+                int c = count + 1;
+                if (c > threshold && tab.length < MAXIMUM_CAPACITY)
+                    rehash(node);
+                else
+                	// 赋值
+                    setEntryAt(tab, index, node);
+                ++modCount;
+                count = c;
+                oldValue = null;
+                break;
+            }
+        }
+    } finally {
+        unlock();
+    }
+    return oldValue;
+}
+```
+
+误区：ConcurrentHashMap不会造成阻塞？
+如果有16个线程，16个key，正好落到不同segment位置的情况不会阻塞
+但是相同的key会导致线程阻塞
+
+因为put方法底层使用了lock锁，调用trylock方法（该方法常用在自旋操作中），他与lock方法不同的地方是没有获取到锁的地方不会阻塞，返回false
+
+## put方法没有抢到锁
+
+```
+private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
+	// 找到链表的位置
+    HashEntry<K,V> first = entryForHash(this, hash);
+    HashEntry<K,V> e = first;
+    HashEntry<K,V> node = null;
+    int retries = -1; // negative while locating node
+    while (!tryLock()) {
+        HashEntry<K,V> f; // to recheck first below
+        if (retries < 0) {
+        	// 假设当前e不为空
+            if (e == null) {
+                if (node == null) // speculatively create node
+                    node = new HashEntry<K,V>(hash, key, value, null);
+                retries = 0;
+            }
+            else if (key.equals(e.key))
+                retries = 0;
+            else
+                e = e.next;
+        }
+        // 重试获取所得状态，如果超过最大次数，则阻塞等待
+        else if (++retries > MAX_SCAN_RETRIES) {
+            lock();
+            break;
+        } 
+        // 重新拿到它的链表
+        // 需要注意这行代码，不单单是保证线程的时刻更新，也将死循环的时候最大限度利用CPU资源，形成cache，为后续的插入操作做准备
+        else if ((retries & 1) == 0 &&
+                (f = entryForHash(this, hash)) != first) {
+            e = first = f; // re-traverse if entry changed
+            retries = -1;
+        }
+    }
+    return node;
+}
+```
+
+## ConcurrentHashMap底层实现原理
+
+1. 有多个不同的segment对象组成
+2. lock锁
+3. UNSAFE查询内存最新的数据
+4. 使用cas做修改