HashTable与ConcurrentHashMap

骑猪看日落 2022-11-29 13:17 168阅读 0赞

## 前言 ##

HashMap在多线程环境下存在线程安全问题，主要体现：

*  在jdk1.7中，在多线程环境下，扩容时会造成环形链或数据丢失。
 *  \#2.在jdk1.8中，在多线程环境下，会发生数据覆盖的情况。

因此，一般在多线程的场景，会使用一下几种不同的方式去代替：

*  使用Collections.synchronizedMap(Map)创建线程安全的map集合；
 *  Hashtable
 *  ConcurrentHashMap

不过出于线程并发度的原因，一般会舍弃前两者使用最后的ConcurrentHashMap，他的性能和效率明显高于前两者。

## Collections.synchronizedMap(Map) ##

在SynchronizedMap内部维护了一个普通对象Map，还有排斥锁mutex

public static <K,V> Map<K,V> synchronizedMap(Map<K,V> m) { 
            return new SynchronizedMap<>(m);
        }
    
        /** * @serial include */
        private static class SynchronizedMap<K,V>
            implements Map<K,V>, Serializable { 
            private static final long serialVersionUID = 1978198479659022715L;
    
            private final Map<K,V> m;     // Backing Map
            final Object      mutex;        // Object on which to synchronize
    
            SynchronizedMap(Map<K,V> m) { 
                this.m = Objects.requireNonNull(m);
                mutex = this;
            }
    
            SynchronizedMap(Map<K,V> m, Object mutex) { 
                this.m = m;
                this.mutex = mutex;
            }

我们在调用这个方法的时候就需要传入一个Map，可以看到有两个构造器，如果你传入了mutex参数，则将对象排斥锁赋值为传入的对象。

如果没有，则将对象排斥锁赋值为this，即调用synchronizedMap的对象，就是上面的Map。

创建出synchronizedMap之后，再操作map的时候，就会对方法上锁  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzIxNzA2NQ_size_16_color_FFFFFF_t_70_pic_center]

## HashTable ##

Hashtable 继承了 Dictionary类，而 HashMap 继承的是 AbstractMap 类。Hashtable 初始容量为：11，负载因子默认都是：0.75。当现有容量大于总容量 \* 负载因子时，HashMap 扩容规则为当前容量翻倍，Hashtable 扩容规则为当前容量翻倍 + 1。跟HashMap相比Hashtable是线程安全的，适合在多线程的情况下使用，但是效率可不太乐观。在对数据操作的时候都会直接在方法上锁。

public synchronized V put(K key, V value) { 
            // Make sure the value is not null
            if (value == null) { 
                throw new NullPointerException();
            }
    
            // Makes sure the key is not already in the hashtable.
            Entry<?,?> tab[] = table;
            int hash = key.hashCode();
            int index = (hash & 0x7FFFFFFF) % tab.length;
            @SuppressWarnings("unchecked")
            Entry<K,V> entry = (Entry<K,V>)tab[index];
            for(; entry != null ; entry = entry.next) { 
                if ((entry.hash == hash) && entry.key.equals(key)) { 
                    V old = entry.value;
                    entry.value = value;
                    return old;
                }
            }
    
            addEntry(hash, key, value, index);
            return null;
        }

**为啥Hashtable 是不允许键或值为 null 的，HashMap 的键值则都可以为 null？**

> 因为Hashtable使用的是安全失败机制（fail-safe），这种机制会使你此次读到的数据不一定是最新的数据。
> 
> 如果你使用null值，就会使得其无法判断对应的key是不存在还是为空，因为你无法再调用一次contain(key）来对key是否存在进行判断，ConcurrentHashMap同理。

## ConcurrentHashMap ##

### 1.7的ConcurrentHashMap ###

数据结构：是由 Segment 数组、HashEntry 组成，和 HashMap 一样，仍然是数组加链表。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzIxNzA2NQ_size_16_color_FFFFFF_t_70_pic_center 1]  
ConcurrentHashMap 采用了分段锁技术，其中 Segment 继承于 ReentrantLock。每一个Semgent可以看成是一个HashTable。

static final class Segment<K,V> extends ReentrantLock implements Serializable { 
    
        private static final long serialVersionUID = 2249069246763182397L;
    
        // 和 HashMap 中的 HashEntry 作用一样，真正存放数据的桶
        transient volatile HashEntry<K,V>[] table;
    
        transient int count;
            // 记得快速失败（fail—fast）么？
        transient int modCount;
            // 大小
        transient int threshold;
            // 负载因子
        final float loadFactor;
    
    }

不会像 HashTable 那样不管是 put 还是 get 操作都需要做同步处理，理论上 ConcurrentHashMap 支持 CurrencyLevel (Segment 数组数量)的线程并发。

每当一个线程占用锁访问一个 Segment 时，不会影响到其他的 Segment。

就是说如果容量大小是16他的并发度就是16，可以同时允许16个线程操作16个Segment而且还是线程安全的。  
ConcurrentHashMap在进行put操作时，先定位到Segment，然后再进行put操作。

public V put(K key, V value) { 
        Segment<K,V> s;
        if (value == null)
            throw new NullPointerException();//这就是为啥他不可以put null值的原因
        int hash = hash(key);
        int j = (hash >>> segmentShift) & segmentMask;
        if ((s = (Segment<K,V>)UNSAFE.getObject          
             (segments, (j << SSHIFT) + SBASE)) == null) 
            s = ensureSegment(j);
        return s.put(key, hash, value, false);
    }

final V put(K key, int hash, V value, boolean onlyIfAbsent) { 
              // 将当前 Segment 中的 table 通过 key 的 hashcode 定位到 HashEntry
                HashEntry<K,V> node = tryLock() ? null :
                    scanAndLockForPut(key, hash, value);
                V oldValue;
                try { 
                    HashEntry<K,V>[] tab = table;
                    int index = (tab.length - 1) & hash;
                    HashEntry<K,V> first = entryAt(tab, index);
                    for (HashEntry<K,V> e = first;;) { 
                        if (e != null) { 
                            K k;
     // 遍历该 HashEntry，如果不为空则判断传入的 key 和当前遍历的 key 是否相等，相等则覆盖旧的 value。
                            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 { 
                     // 不为空则需要新建一个 HashEntry 并加入到 Segment 中，同时会先判断是否需要扩容。
                            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;
            }

首先第一步的时候会尝试获取锁，如果获取失败肯定就有其他线程存在竞争，则利用 scanAndLockForPut() 自旋获取锁。

*  尝试自旋获取锁。
 *  如果重试的次数达到了 MAX\_SCAN\_RETRIES 则改为阻塞锁获取，保证能获取成功。

get 逻辑比较简单，只需要将 Key 通过 Hash 之后定位到具体的 Segment ，再通过一次 Hash 定位到具体的元素上。

由于 HashEntry 中的 value 属性是用 volatile 关键词修饰的，保证了内存可见性，所以每次获取时都是最新值。

ConcurrentHashMap 的 get 方法是非常高效的，因为整个过程都不需要加锁。

### 1.8的ConcurrentHashMap ###

抛弃了原有的 Segment 分段锁，而采用了 CAS + synchronized 来保证并发安全性。把值和next采用了volatile去修饰，保证了可见性，并且也引入了红黑树，在链表大于一定值的时候会转换（默认是8）。

ConcurrentHashMap在进行put操作的还是比较复杂的，大致可以分为以下步骤：

1.  根据 key 计算出 hashcode 。
2.  判断是否需要进行初始化。
3.  即为当前 key 定位出的 Node，如果为空表示当前位置可以写入数据，利用 CAS 尝试写入，失败则自旋保证成功。
4.  如果当前位置的 hashcode == MOVED == -1,则需要进行扩容。
5.  如果都不满足，则利用 synchronized 锁写入数据。
6.  如果数量大于 TREEIFY\_THRESHOLD 则要转换为红黑树。

public V put(K key, V value) { 
            return putVal(key, value, false);
        }
    
        /** Implementation for put and putIfAbsent */
        final V putVal(K key, V value, boolean onlyIfAbsent) { 
            //插入key或value为null时，抛出空指针异常
            if (key == null || value == null) throw new NullPointerException();
            // 1. 根据 key 计算出 hashcode 。
            int hash = spread(key.hashCode());
            int binCount = 0;
            for (Node<K,V>[] tab = table;;) { 
                Node<K,V> f; int n, i, fh;
                //2. 判断是否需要进行初始化。调用initTable()方法初始化table
                if (tab == null || (n = tab.length) == 0)
                    tab = initTable();
                //3. 即为当前 key 定位出的 Node，如果为空表示当前位置可以写入数据，利用 CAS 尝试写入，失败则自旋保证成功。
                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
                }
                //4. 如果当前位置的 hashcode == MOVED == -1,表示当前有一个线程正在对map扩容，则需要帮助扩容。
                else if ((fh = f.hash) == MOVED)
                    tab = helpTransfer(tab, f);
                // 5. 如果都不满足，则利用 synchronized 锁写入数据。
                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;
                                    if (e.hash == hash &&
                                        ((ek = e.key) == key ||
                                         (ek != null && key.equals(ek)))) { 
                                        oldVal = e.val;
                                        //判断链表是否允许被修改，如果onlyIfAbsent为false,则直接修改value
                                        if (!onlyIfAbsent)
                                            e.val = value;
                                        break;
                                    }
                                    Node<K,V> pred = e;
                                    //遍历到链表尾部，使用尾插法插入新的node节点
                                    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,
                                                               value)) != null) { 
                                    oldVal = p.val;
                                    if (!onlyIfAbsent)
                                        p.val = value;
                                }
                            }
                        }
                    }
                    if (binCount != 0) { 
                        //6.链表长度大于TREEIFY_THRESHOLD时，将链表转化为红黑树
                        if (binCount >= TREEIFY_THRESHOLD)
                            treeifyBin(tab, i);
                        if (oldVal != null)
                            return oldVal;
                        break;
                    }
                }
            }
            //添加数量，并检查是否需要进行扩容
            addCount(1L, binCount);
            return null;
        }

**initTable()方法源码**

private final Node<K,V>[] initTable() { 
            Node<K,V>[] tab; int sc;
            while ((tab = table) == null || tab.length == 0) { 
                if ((sc = sizeCtl) < 0)
                    //放弃CPU资源，重新竞争
                    Thread.yield(); // lost initialization race; just spin
                //通过CAS，只允许一个线程进行初始化
                else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { 
                    try { 
                        if ((tab = table) == null || tab.length == 0) { 
                            //根据是否有赋初值，来初始化table大小
                            int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                            //初始化table
                            @SuppressWarnings("unchecked")
                            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                            table = tab = nt;
                            //计算阈值：为 n - 1/4 n
                            sc = n - (n >>> 2);
                        }
                    } finally { 
                        sizeCtl = sc;
                    }
                    break;
                }
            }
            return tab;
        }

**ConcurrentHashMap的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());
            //1.根据计算出来的 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;
                }
                //如果是红黑树那就按照树的方式获取值。
                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;
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzIxNzA2NQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/4d7c8aa20b2c4d4f8ee684562f5552d7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzIxNzA2NQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221124/003cede3d7e8495dae5f291902b5154f.png