Java集合(六)Hashtable、ConcurrentHashMap

发布时间:2023年12月17日

??本系列文章:
????Java集合(一)集合框架概述
????Java集合(二)List、ArrayList、LinkedList
????Java集合(三)CopyOnWriteArrayList、Vector、Stack
????Java集合(四)Map、HashMap
????Java集合(五)LinkedHashMap、TreeMap
????Java集合(六)Hashtable、ConcurrentHashMap
????Java集合(七)Set、HashSet、LinkedHashSet、TreeSet
????Java集合(八)BlockingQueue、ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue、DelayQueue

????Java集合(八)BlockingQueue、ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue、DelayQueue

Hashtable

一、Hashtable介绍

??Hashtable是线程安全版的Map实现,属于遗留类,很多功能与HashMap类似,是线程安全的。但任一时间只有一个线程能写Hashtable,并发性不如ConcurrentHashMap,因为ConcurrentHashMap引入了分段锁。
??Hashtable不建议在新代码中使用,不需要线程安全的场合可以用HashMap,需要线程安全的场合可以用ConcurrentHashMap。

1.1 Hashtable特点

??HashMap和HashTable用法几乎一样,底层实现也几乎一样,但是HashTable的方法添加了synchronized关键字以确保线程安全,同时也导致效率较低。两者的区别:

HashMapHashtable
key和value是否可以为Null均可以为nullkey和value均不能为Null
底层实现数组+链表+红黑树数组+链表
对hash冲突的处理如果链表节点数小于8时是将新元素加入到链表的末尾将新元素加入到链表的开头
数组容量大小2的n次方,如果初始化时不符合要求会进行调整可以为任意正整数
默认容量1611
扩容方式默认原容量*2默认原容量*2+1
是否线程安全线程不安全线程安全
hash算法hashCode的高16位 异或 低16位(即(h = key.hashCode())^ (h >>>16))。这么做可以在数组table的length较小的时候,也能保证考虑到高低位都参与到 Hash 的计算中,计算出的位置更加分散,同时不会有太大的开销除留余数法,直接使用Object的hashcode计算

1.2 Hashtable的使用

??Hashtable的方法基本和HashMap一样。

  • 1、构造方法
	//默认初始容量(11)和负载因子(0.75)
	public Hashtable()
	//指定初始容量和默认负载因子(0.75)
	public Hashtable(int initialCapacity)
	//指定初始容量和负载因子
	public Hashtable(int initialCapacity, float loadFactor)
  • 2、判断是否包含key/value
	//是否包含某个Value
	public synchronized boolean contains(Object value)
	//是否包含某个key
	public synchronized boolean containsKey(Object key)
	//是否包含某个Value
	public boolean containsValue(Object value)
  • 3、遍历
	//返回散列表中包含的键值对的Set视图
	public Set<Map.Entry<K,V>> entrySet()
	//返回此散列表中包含的value的集合
	public Collection< V > values()	
	//返回散列表中包含的key的Set视图
	public Set< K > keySet()	
  • 4、获取value
	//获取key对应的value
	public synchronized V get(Object key)
	//获取key对应的value,key不存在则返回默认值
	public synchronized V getOrDefault(Object key, V defaultValue)
  • 5、添加元素
	//添加或替换键值对
	public synchronized V put(K key, V value)
	//将指定Map中的键值对全添加到散列表中
	public synchronized void putAll(Map<? extends K, ? extends V> t)
  • 6、删除键值对
	//从散列表中删除键(及其对应的值)
	public synchronized V remove(Object key)
	//如果指定键值对存在时,删除这个键值对
	public synchronized boolean remove(Object key, Object value)
  • 7、替换键值对
	//替换指定key对应的value
	public synchronized V replace(K key, V value) 
	//只有当特定的键值对存在时,替换value
	public synchronized boolean replace(K key, V oldValue, V newValue)
  • 8、其他方法
	//返回此散列表中的键值对数
	public synchronized int size()
	//散列表是否为空
	public synchronized boolean isEmpty()	

二、从源码理解Hashtable

??先看一些变量:

	//保存key-value的数组,支持泛型 
    // Entry同样采用链表解决冲突,每一个Entry本质上是一个单向链表  
    private transient Entry<?,?>[] table;
	//Hashtable中Entry对象的个数
    private transient int count;
    //临界值 当实际大小(容量*填充因子)超过临界值时,会进行扩容
    private int threshold;
	//负载因子,当元素个数count大于总容量 * 负载因子时,扩容
    private float loadFactor;
	//Entry被改变的次数,用于fail-fast机制的实现 
    private transient int modCount = 0;
    //最大容量
    private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

2.1 节点

??Hashtable中Entry也和HashMap中的Entry相似:

 private static class Entry<K,V> implements Map.Entry<K,V> {
    final int hash;  //hash值
    final K key;
    V value;
    Entry<K,V> next;  //下一个节点
    
    // 设置value。若value是null,则抛出异常。
    public V setValue(V value) {
        if (value == null)
            throw new NullPointerException();

        V oldValue = this.value;
        this.value = value;
        return oldValue;
    }

    // 覆盖equals()方法,判断两个Entry是否相等。
    // 若两个Entry的key和value都相等,则认为它们相等。
    public boolean equals(Object o) {
        if (!(o instanceof Map.Entry))
            return false;
        Map.Entry e = (Map.Entry)o;

        return (key==null ? e.getKey()==null : key.equals(e.getKey())) &&
           (value==null ? e.getValue()==null : value.equals(e.getValue()));
    }

    //计算键值对的hashCode
    public int hashCode() {
    	// "^" 按位异或, hash在调用构造器时传入
        return hash ^ Objects.hashCode(value);
    }
    ......
}

2.2 构造方法

??Hashtable和HashMap的构造方法相同的是,均是对初始容量和加载因子完成了设置。不同的地方有2点:

  • 1、HashMap对底层数组采取的懒加载,即当执行第一次插入时才会创建数组;而Hashtable在初始化时就创建了数组;
  • 2、HashMap中数组的默认初始容量是16,并且必须的是2的指数倍数;而Hashtable中默认的初始容量是11,并且不要求必须是2的指数倍数。
  • 1、无参构造方法
    public Hashtable() {
        //默认容量大小为11,负载因子设置为0.75
        this(11, 0.75f);
    }
  • 2、带有初始化容量大小的构造方法
    public Hashtable(int initialCapacity) {
        this(initialCapacity, 0.75f);
    }
  • 3、指定初始容量和负载因子的构造方法
    public Hashtable(int initialCapacity, float loadFactor) {
        //检查参数的合法性
        if (initialCapacity < 0)
            throw new IllegalArgumentException("Illegal Capacity: "+
                                               initialCapacity);
        if (loadFactor <= 0 || Float.isNaN(loadFactor))
            throw new IllegalArgumentException("Illegal Load: "+loadFactor);
 
        //如果设置初始容量为0,则默认修改为1
        if (initialCapacity==0)
            initialCapacity = 1;
        //设置负载因子
        this.loadFactor = loadFactor;
        
        //根据设置的初始化容量创建数组
        table = new Entry<?,?>[initialCapacity];
 
        //散列表防止经过n次扩容后,数组大小可能会超出整数的最大值,所以
        //这里设定一个上限的阈值
        threshold = (int)Math.min(initialCapacity, MAX_ARRAY_SIZE + 1);
    }

2.3 获取元素

??get的步骤:

  1. 计算key对应的哈希值;
  2. 根据哈希值计算数组下标;
  3. 遍历数组下标位置对应的链表,找到则返回对应的value,否则返回null。

??方法用synchronized修饰,同时只能有一个线程获取。

    public synchronized V get(Object key) {
        
        Entry<?,?> tab[] = table;
        //得到key的hashcode
        int hash = key.hashCode();
        //根据hashcode计算索引值
        int index = (hash & 0x7FFFFFFF) % tab.length;
        //根据index找到key对应Entry链表,遍历链表找到哈希值与键值均与key相同的元素
        for (Entry<?,?> e = tab[index] ; e != null ; e = e.next) {
        	//哈希值与key均一样,代表找到了指定key对应的元素
            if ((e.hash == hash) && e.key.equals(key)) {
                return (V)e.value;
            }
        }
        // 若没有找到,则返回null
        return null;
    }

2.4 存入元素

??put的步骤:

  1. 根据key获取计算index;
  2. 如果index位置的链表中,key已存在,更新key的映射值value,并返回旧值;
  3. 否则将新的键值对添加到链表上。在这个过程中,需要判断需不需要扩容,如果扩容了,需要重新计算key新的哈希值及存储位置,再将键值对存放在新的存储位置上。
    public synchronized V put(K key, V value) {
        // 检验数据值的合法性
        if (value == null) {
            throw new NullPointerException();
        }
 
        Entry<?,?> tab[] = table;
        //根据键值获取索引index
        int hash = key.hashCode();
        int index = (hash & 0x7FFFFFFF) % tab.length;
        //判断tab[index]是否已经有值
        @SuppressWarnings("unchecked")
        Entry<K,V> entry = (Entry<K,V>)tab[index];
        //如果有值,则遍历
        for(; entry != null ; entry = entry.next) {
             //当前键值key已存在,更新key的映射值value,并返回旧值
            if ((entry.hash == hash) && entry.key.equals(key)) {
                V old = entry.value;
                entry.value = value;
                return old;
            }
        }
 
        //若没有找到重复键值key,则将key和value添加链表末尾
        addEntry(hash, key, value, index);
        return null;
    }
    
    //添加元素
    private void addEntry(int hash, K key, V value, int index) {
        modCount++;
        Entry<?,?> tab[] = table;
        
        //判断当前数目是否超过阈值
        if (count >= threshold) {
            // 数目超过阈值,扩容,重排列
            rehash();
 
            //更新扩容后的数组信息
            tab = table;
            hash = key.hashCode();
            index = (hash & 0x7FFFFFFF) % tab.length;
        }
 
        // 没有超过阈值,则添加至数组中
        @SuppressWarnings("unchecked")
        Entry<K,V> e = (Entry<K,V>) tab[index];
        tab[index] = new Entry<>(hash, key, value, e);
        
        //增加元素数目
        count++;
    }

2.5 是否包含某个key/value

??containsKey方法中有行代码:if ((e.hash == hash) && e.key.equals(key)),这里其实只判断e.key.equals(key)即可,但是考虑到执行的速度,可以先判断e.hash == hash。因为hash是提前计算好的值,且该判断比e.key.equals(key)执行速度要快,而hashtable里大部分元素的hash值是不相同的,只有当hash值相同时才用equal判断,这样就可以快速筛选hashtable的元素。

    //判断是否含有value
    public boolean containsValue(Object value) {
        return contains(value);
    }
    
    public synchronized boolean contains(Object value) {
        //检查参数的合法性,从这里也可以看出:Hashtable的value不允许为空,不然会报空指针
        if (value == null) {
            throw new NullPointerException();
        }
 
        // 双重for循环,外循环遍历数组,内循环遍历链表
        Entry<?,?> tab[] = table;
        //从后向前遍历数组
        for (int i = tab.length ; i-- > 0 ;) {
        	//逐个遍历数组节点上的链表
            for (Entry<?,?> e = tab[i] ; e != null ; e = e.next) {
                if (e.value.equals(value)) {
                    return true;
                }
            }
        }
        return false;
    }
    
    // 判断是否包含键值key
    public synchronized boolean containsKey(Object key) {
        Entry<?,?> tab[] = table;
        int hash = key.hashCode();
        //计算数组的索引,Hashtable本质上采用除数取余法进行散列分布
        int index = (hash & 0x7FFFFFFF) % tab.length;
        // index定位数组位置,for遍历链表查找元素
        for (Entry<?,?> e = tab[index] ; e != null ; e = e.next) {
            if ((e.hash == hash) && e.key.equals(key)) {
                return true;
            }
        }
        return false;
    }

2.6 替换元素

	public synchronized V replace(K key, V value) {
        Objects.requireNonNull(value);
        //根据键值查找元素
        Entry<?,?> tab[] = table;
        //固定套路,根据key来计算哈希值,进而计算在数组中的位置index
        int hash = key.hashCode();
        int index = (hash & 0x7FFFFFFF) % tab.length;
        @SuppressWarnings("unchecked")
        Entry<K,V> e = (Entry<K,V>)tab[index];
        //根据index找到在数组中的位置后,遍历该位置上的链表
        for (; e != null; e = e.next) {
             //查找成功,替换元素值
            if ((e.hash == hash) && e.key.equals(key)) {
                V oldValue = e.value;
                e.value = value;
                return oldValue;
            }
        }
        return null;
    }

2.7 删除元素

	//根据键值删除元素,返回被删除元素值
	public synchronized V remove(Object key) {
        Entry<?,?> tab[] = table;
        int hash = key.hashCode();
        int index = (hash & 0x7FFFFFFF) % tab.length;
        @SuppressWarnings("unchecked")
        Entry<K,V> e = (Entry<K,V>)tab[index];
         //for遍历链表查找元素
        for(Entry<K,V> prev = null ; e != null ; prev = e, e = e.next) {
            //查找到元素进行链表的节点删除操作
            if ((e.hash == hash) && e.key.equals(key)) {
                modCount++;
                
                if (prev != null) {
                	//pre结点的next指针指向e的next,等价于e被删除
                    prev.next = e.next;
                } else {
                	//否则,说明需要删除的为起始结点
                    tab[index] = e.next;
                }
                count--;
                V oldValue = e.value;
                e.value = null;
                return oldValue;
            }
        }
        return null;
    }

2.8 哈希

    public synchronized int hashCode() {
        int h = 0;
        if (count == 0 || loadFactor < 0)
            return h;

		//Mark hashCode computation in progress
		//标记哈希计算开始
        loadFactor = -loadFactor; 
        Entry<?,?>[] tab = table;
        for (Entry<?,?> entry : tab) {
            while (entry != null) {
            	//计算每个键值对的哈希值并累加
                h += entry.hashCode();
                //获取下一个键值对
                entry = entry.next;
            }
        }

		//标记哈希计算结束
        loadFactor = -loadFactor;  

        return h;
    }

2.9 扩容

    //扩容方法
    protected void rehash() {
        //获取旧数组大小
        int oldCapacity = table.length;
        Entry<?,?>[] oldMap = table;
 
        // 创建新容量大小的Entry数组,数组容量大小为原数组的2倍+1  
        int newCapacity = (oldCapacity << 1) + 1;
        if (newCapacity - MAX_ARRAY_SIZE > 0) {
            if (oldCapacity == MAX_ARRAY_SIZE)
                //如果散列表的容量已经是MAX_ARRAY_SIZE,则不再扩容
                return;
            newCapacity = MAX_ARRAY_SIZE;
        }
        Entry<?,?>[] newMap = new Entry<?,?>[newCapacity];
 
        modCount++;
        //重新计算阈值
        threshold = (int)Math.min(newCapacity * loadFactor, MAX_ARRAY_SIZE + 1);
        table = newMap;
 
        //将原数组中元素拷贝至新数组
       for (int i = oldCapacity ; i-- > 0 ;) {
            for (Entry<K,V> old = (Entry<K,V>)oldMap[i] ; old != null ; ) {
                Entry<K,V> e = old;
                old = old.next;
 				//重新计算新数组的索引值
                int index = (e.hash & 0x7FFFFFFF) % newCapacity;
                //先插入起始值
                e.next = (Entry<K,V>)newMap[index];
                //对应的向右侧移动
                newMap[index] = e;
            }
        }
    }

ConcurrentHashMap

一、ConcurrentHashMap介绍

??ConcurrentHashMap是线程安全、并且效率更高(和HashTable相比)的HashMap容器。

1.1 不同JDK版本中的实现

  • JDK1.7版本中的ConcurrentHashMap实现
    ??在JDK1.7版本中,ConcurrentHashMap由一个个Segment组成, Segment代表“段”的意思,所以也称分段锁
    ??简单理解就是,ConcurrentHashMap是一个Segment数组,这样只要保证每个Segment是线程安全的,也就实现了全局的线程安全。
	static class Segment<K,V> extends ReentrantLock implements Serializable

??Segment继承了ReentrantLock,表明每个segment都可以当做一个锁。这
样对每个segment中的数据需要同步操作的话都是使用每个segment容器对象自身的锁来实现。只有对全局需要改变时锁定的是所有的segment。
??ConcurrentHashMap默认有16个Segments,所以理论上,默认最多可以同时支持16个线程并发写,只要它们的操作分别分布在不同的Segment 上。这个值可以在初始化的时候设置为其他值,但是一旦初始化以后,它是不可以改变的。

  • JDK1.8版本中的ConcurrentHashMap实现
    ??在JDK1.8版本中,ConcurrentHashMap的实现引入了红黑树,其底层结构为:数组+链表+红黑树。

    ??JDK1.8中ConcurrentHashMap取消了segment分段锁,而采用CAS和synchronized来保证并发安全。数据结构采用数组+链表/红黑二叉树。synchronized只锁定当前链表或红黑二叉树的首节点,相比1.7锁定HashEntry数组,锁粒度更小,支持更高的并发量。当链表长度过长时,Node会转换成TreeNode,提高查找速度。

1.2 ConcurrentHashMap的使用

  • 1、构造方法
	//创建空的ConcurrentHashMap
	public ConcurrentHashMap()
	//创建指定初始容量的ConcurrentHashMap
	public ConcurrentHashMap(int initialCapacity)
	//创建指定初始容量和加载因子的ConcurrentHashMap
	public ConcurrentHashMap(int initialCapacity, float loadFactor)
	//创建指定初始容量、加载因子和并发度的ConcurrentHashMap
	public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel)

??此处需要解释并发度这个概念,这个值用来确定Segment的个数,Segment的个数是大于等于concurrencyLevel的第一个2的n次方的数。比如,如果concurrencyLevel为12/13/14/15/16这些数,则Segment的数目为16(2的4次方)。默认值是16。
??理想情况下ConcurrentHashMap的真正的并发访问量能够达到concurrencyLevel,因为有concurrencyLevel个Segment,假如有concurrencyLevel个线程需要访问Map,并且需要访问的数据都恰好分别落在不同的Segment中,则这些线程能够无竞争地自由访问(因为他们不需要竞争同一把锁),达到同时访问的效果。

  • 2、判断是否包含某个value/判断是否包含某个key
	//判断是否包含某个value
	public boolean contains(Object value)
	public boolean containsValue(Object value)
	//判断是否包含某个key
	public boolean containsKey(Object key)
  • 3、获取key/value/键值对的集合
	//获取key的枚举
	public Enumeration<K> keys()
	//获取value的枚举
	public Enumeration<V> elements()
	//获取键值对的集合
	public Set<Map.Entry<K,V>> entrySet()	
	//获取value组成的集合
	public Collection<V> values()
  • 4、获取元素
	public V get(Object key)
	//获取不到时,取默认值
	public V getOrDefault(Object key, V defaultValue)
  • 5、添加键值对
	public V put(K key, V value)
  • 6、删除键值对
	//根据key删除键值对
	public V remove(Object key)
	//根据key和value来删除键值对
	public boolean remove(Object key, Object value)
  • 7、替换value
	//替换指定key对应的value
	public V replace(K key, V value) 
	//替换指定key-value对应的value
	public boolean replace(K key, V oldValue, V newValue)

二、从源码理解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;
	//加载因子
    private static final float LOAD_FACTOR = 0.75f;
	//链表转化为红黑树的阈值
    static final int TREEIFY_THRESHOLD = 8;
	//红黑树转化为链表的阈值
    static final int UNTREEIFY_THRESHOLD = 6;
	//链表转化为红黑树时,数组长度阈值
    static final int MIN_TREEIFY_CAPACITY = 64;
	//用于为每个线程计算允许处理的最少table桶首节点个数
    private static final int MIN_TRANSFER_STRIDE = 16;
	//stamp高位标识移动位数
    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;
	
    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
    static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
	//可用处理器数量
	static final int NCPU = Runtime.getRuntime().availableProcessors();
	//默认为null,初始化发生在第一次插入操作,默认大小为16
	//的数组,用来存储Node节点数据,扩容时大小总是2的幂次方
	transient volatile Node<K,V>[] table;
	//默认为null,扩容时新生成的数组,其大小为原数组的两倍
	private transient volatile Node<K,V>[] nextTable;
	//默认为0,用来控制table的初始化和扩容操作
	// -1:代表table正在初始化
	// -N:表示有N-1个线程正在进行扩容操作
	//如果table未初始化,表示table需要初始化的大小
	//如果table初始化完成,表示table的容量,默认是table大小的0.75倍
	private transient volatile int sizeCtl;

2.1 节点

??Node,普通节点,由哈希值、键、值和下一个节点组成:

    static class Node<K,V> implements Map.Entry<K,V> {
        final int hash;
        final K key;
        //value和next都用volatile修饰,保证并发的可见性
        volatile V val;
        volatile Node<K,V> next;
        //...
    }

??ForwardingNode:一个特殊的Node节点,hash值为-1,其中存储nextTable的引用。只有table发生扩容的时候,ForwardingNode才会发挥作用,作为一个占位符放在table中表示当前节点。

    static final class ForwardingNode<K,V> extends Node<K,V> {
        final Node<K,V>[] nextTable;
        ForwardingNode(Node<K,V>[] tab) {
            super(MOVED, null, null, null);
            this.nextTable = tab;
        }
        //...
    }

2.2 初始化

??实例化ConcurrentHashMap时,如果声明了table的容量,在初始化时会根据参数调整table大小,确保table的大小总是2的幂次方。默认的table大小为16。
??table的初始化操作回延缓到第一put操作再进行,并且初始化只会执行一次。只有第一次使用才初始化,为了防止初始化后的首次操作就需要扩容(比如putAll),从而影响效率。

	private final Node<K,V>[] initTable() {
	    Node<K,V>[] tab; int sc;
	    while ((tab = table) == null || tab.length == 0) {
			//如果一个线程发现sizeCtl<0,意味着另外的线程执行CAS操作成功,
			//当前线程只需要让出cpu时间片
	        if ((sc = sizeCtl) < 0) 
	            Thread.yield(); 
	        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);  //0.75*capacity
	                }
	            } finally {
	                sizeCtl = sc;
	            }
	            break;
	        }
	    }
	    return tab;
	}

2.3 添加元素

2.3.1 put过程描述

??假设table已经初始化完成,put操作采用CAS+synchronized实现并发插入或更新操作:

  • 1、当前bucket为空时,使用CAS操作,将Node放入对应的bucket中。
  • 2、出现hash冲突,则采用synchronized关键字。倘若当前hash对应的节点是链表的头节点,遍历链表,若找到对应的node节点,则修改node节点的val,否则在链表末尾添加node节点;倘若当前节点是红黑树的根节点,在树结构上遍历元素,更新或增加节点。
  • 3、倘若当前map正在扩容f.hash == MOVED, 则跟其他线程一起进行扩容。
final V putVal(K key, V value, boolean onlyIfAbsent) {
        if (key == null || value == null) throw new NullPointerException();
        int hash = spread(key.hashCode());
        int binCount = 0;
        for (Node<K,V>[] tab = table;;) {
            Node<K,V> f; int n, i, fh;
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();  // lazy Initialization
            // 当前bucket为空
            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
            }
            // 当前Map在扩容,先协助扩容,再更新值
            else if ((fh = f.hash) == MOVED)  
                tab = helpTransfer(tab, f); 
            // hash冲突
            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;
                                    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,
                                                           value)) != null) {
                                oldVal = p.val;
                                if (!onlyIfAbsent)
                                    p.val = value;
                            }
                        }
                    }
                }
                if (binCount != 0) {
                	//链表节点超过了8,链表转为红黑树
                    if (binCount >= TREEIFY_THRESHOLD)  
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        // 统计节点个数,检查是否需要resize
        addCount(1L, binCount);  
        return null;
    }  
2.3.2 几个注意点
  • 1、hash值的计算
	static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
	static final int spread(int h) {return (h ^ (h >>> 16)) & HASH_BITS;}
  • 2、定位索引
	int index = (n - 1) & hash  // n为bucket的个数
  • 3、获取table对应的索引元素f
	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);
    }

??之所以采用Unsafe.getObjectVolatie()来获取,而不是直接用table[index],是和ConcurrentHashMap的弱一致性有关。在Java内存模型中,我们已经知道每个线程都有一个工作内存,里面存储着table的副本,虽然table是volatile修饰的,但不能保证线程每次都拿到table中的最新元素,Unsafe.getObjectVolatile可以直接获取指定内存的数据,保证了每次拿到数据都是最新的。

2.4 扩容

??如果新增节点之后,所在的链表的元素个数大于等于8并且tab.length >= 64(MIN_TREEIFY_CAPACITY),则会调用treeifyBin把链表转换为红黑树。否则会将数组长度扩大到原来的两倍,并触发transfer,重新调整节点位置。
??新增节点后,addCount统计tab中的节点个数大于阈值(sizeCtl),会触发transfer,重新调整节点位置。

  • 1、addCount
private final void addCount(long x, int check) {
        CounterCell[] as; long b, s;
        // 利用CAS更新baseCount  
        if ((as = counterCells) != null ||
            !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
            CounterCell a; long v; int m;
            boolean uncontended = true;
            if (as == null || (m = as.length - 1) < 0 ||
                (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                !(uncontended =
                  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
                fullAddCount(x, uncontended); // 多线程修改baseCount时,竞争失败的线程会执行fullAddCount(x, uncontended),把x的值插入到counterCell类中
                return;
            }
            if (check <= 1)
                return;
            s = sumCount();
        }
        if (check >= 0) {
            Node<K,V>[] tab, nt; int n, sc;
            while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                   (n = tab.length) < MAXIMUM_CAPACITY) {
                int rs = resizeStamp(n);
                if (sc < 0) {  
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                        transferIndex <= 0)  // 其他线程在初始化,break;
                        break;
                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))  // 其他线程正在扩容,协助扩容
                        transfer(tab, nt);
                }
                else if (U.compareAndSwapInt(this, SIZECTL, sc,    
                                             (rs << RESIZE_STAMP_SHIFT) + 2))
                    transfer(tab, null);  // 仅当前线程在扩容
                s = sumCount();
            }
        }
    }
  • 2、treeify
private final void treeifyBin(Node<K,V>[] tab, int index) {  
        Node<K,V> b; int n, sc;  
        if (tab != null) {  
            if ((n = tab.length) < MIN_TREEIFY_CAPACITY)//如果table.length<64 就扩大一倍 返回  
                tryPresize(n << 1);  
            else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {  
                synchronized (b) {  
                    if (tabAt(tab, index) == b) {  
                        TreeNode<K,V> hd = null, tl = null;  
                        //构造了一个TreeBin对象 把所有Node节点包装成TreeNode放进去  
                        for (Node<K,V> e = b; e != null; e = e.next) {  
                            TreeNode<K,V> p =  
                                new TreeNode<K,V>(e.hash, e.key, e.val,  
                                                  null, null);//这里只是利用了TreeNode封装 而没有利用TreeNode的next域和parent域  
                            if ((p.prev = tl) == null)  
                                hd = p;  
                            else  
                                tl.next = p;  
                            tl = p;  
                        }  
                        //在原来index的位置 用TreeBin替换掉原来的Node对象  
                        setTabAt(tab, index, new TreeBin<K,V>(hd));  
                    }  
                }  
            }  
        }  
    }  
  • 3、transfer
    ??当table的元素数量达到容量阈值sizeCtl,需要对table进行扩容:

构建一个nextTable,大小为table两倍。
把table的数据复制到nextTable中。

??在扩容过程中,依然支持并发更新操作;也支持并发插入。

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
        int n = tab.length, stride;
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; // subdivide range
        if (nextTab == null) {            // initiating
            try {
                @SuppressWarnings("unchecked")
                Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];  // 构建一个nextTable,大小为table两倍
                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自循环处理每个槽位中的链表元素,默认advace为真,通过CAS设置transferIndex属性值,并初始化i和bound值,i指当前处理的槽位序号,bound指需要处理的槽位边界,先处理槽位15的节点;
        for (int i = 0, bound = 0;;) {
            Node<K,V> f; int fh;
            while (advance) { // 遍历table中的每一个节点 
                int nextIndex, nextBound;
                if (--i >= bound || finishing)
                    advance = false;
                else if ((nextIndex = transferIndex) <= 0) {
                    i = -1;
                    advance = false;
                }
                else if (U.compareAndSwapInt
                         (this, TRANSFERINDEX, nextIndex,
                          nextBound = (nextIndex > stride ?
                                       nextIndex - stride : 0))) {
                    bound = nextBound;
                    i = nextIndex - 1;
                    advance = false;
                }
            }
            if (i < 0 || i >= n || i + n >= nextn) {
                int sc;
                if (finishing) {  // //如果所有的节点都已经完成复制工作  就把nextTable赋值给table 清空临时对象nextTable  
                    nextTable = null;
                    table = nextTab;
                    sizeCtl = (n << 1) - (n >>> 1);  //扩容阈值设置为原来容量的1.5倍  依然相当于现在容量的0.75倍
                    return;
                }
                // 利用CAS方法更新这个扩容阈值,在这里面sizectl值减一,说明新加入一个线程参与到扩容操作
                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
                }
            }
            //如果遍历到的节点为空 则放入ForwardingNode指针 
            else if ((f = tabAt(tab, i)) == null)
                advance = casTabAt(tab, i, null, fwd);
            //如果遍历到ForwardingNode节点  说明这个点已经被处理过了 直接跳过  这里是控制并发扩容的核心  
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            else {
                synchronized (f) {
                    if (tabAt(tab, i) == f) {  
                        Node<K,V> ln, hn;
                        if (fh >= 0) {  // 链表节点
                            int runBit = fh & n;  // resize后的元素要么在原地,要么移动n位(n为原capacity),详解见:https://huanglei.rocks/coding/194.html#4%20resize()%E7%9A%84%E5%AE%9E%E7%8E%B0
                            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;
                                if ((ph & n) == 0)
                                    ln = new Node<K,V>(ph, pk, pv, ln);
                                else
                                    hn = new Node<K,V>(ph, pk, pv, hn);
                            }
                            //在nextTable的i位置上插入一个链表 
                            setTabAt(nextTab, i, ln);
                            //在nextTable的i+n的位置上插入另一个链表
                            setTabAt(nextTab, i + n, hn);
                            setTabAt(tab, i, fwd);
                            //设置advance为true 返回到上面的while循环中 就可以执行i--操作

                            advance = true;
                        }
                        //对TreeBin对象进行处理  与上面的过程类似 
                        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;
                                }
                            }
                            // (1)如果lo链表的元素个数小于等于UNTREEIFY_THRESHOLD,默认为6,则通过untreeify方法把树节点链表转化成普通节点链表;(2)否则判断hi链表中的元素个数是否等于0:如果等于0,表示lo链表中包含了所有原始节点,则设置原始红黑树给ln,否则根据lo链表重新构造红黑树。
                            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); // tab[i]已经处理完了
                            advance = true;
                        }
                    }
                }
            }
        }
    }

??如何在扩容时,并发地复制与插入?

  1. 遍历整个table,当前节点为空,则采用CAS的方式在当前位置放入fwd
  2. 当前节点已经为fwd(with hash field “MOVED”),则已经有有线程处理完了了,直接跳过 ,这里是控制并发扩容的核心
  3. 当前节点为链表节点或红黑树,重新计算链表节点的hash值,移动到nextTable相应的位置(构建了一个反序链表和顺序链表,分别放置在i和i+n的位置上)。移动完成后,用Unsafe.putObjectVolatile在tab的原位置赋为为fwd, 表示当前节点已经完成扩容。

2.5 获取指定key对应的value

??get方法的取值逻辑:

  1. 根据计算出来的hashcode寻址,如果就在桶上那么直接返回值。
  2. 如果是红黑树那就按照树的方式获取值。
  3. 不满足那就按照链表的方式遍历获取值。
	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;
	        }
	        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;
	}

2.5 获取元素个数

??ConcurrentHashMap的元素个数等于baseCounter和数组里每个CounterCell的值之和,这样做的原因是,当多个线程同时执行CAS修改baseCount值,失败的线程会将值放到CounterCell中。所以统计元素个数时,要把baseCount和counterCells数组都考虑。

	private transient volatile long baseCount;
	private transient volatile CounterCell[] counterCells;

	static final class CounterCell {
        volatile long value;
        CounterCell(long x) { value = x; }
    }

	 //返回容器的大小。这个方法应该被用来代替size()方法,因为
     //ConcurrentHashMap的容量大小可能会大于int的最大值。
     //返回的值是一个估计值;如果有并发插入或者删除操作,则实际的数量可能有所不同。
	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;
    }

	public int size() {
        long n = sumCount();
        return ((n < 0L) ? 0 :
                (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                (int)n);
    }

2.6 清空Map

??清空tab的过程:遍历tab中每一个bucket,

  1. 当前bucket正在扩容,先协助扩容;
  2. 给当前bucket上锁,删除元素;
  3. 更新map的size。
	public void clear() { // 移除所有元素  
	    long delta = 0L; // negative number of deletions  
	    inti = 0;  
	    Node<K,V>[] tab = table;  
	    while (tab != null && i < tab.length) {  
	       intfh;  
	       Node<K,V> f = tabAt(tab, i);  
	       if (f == null) // 为空,直接跳过  
	           ++i;  
	       else if ((fh = f.hash) == MOVED) { //检测到其他线程正对其扩容  
	//则协助其扩容,然后重置计数器,重新挨个删除元素,避免删除了元素,其他线程又新增元素。  
	           tab = helpTransfer(tab, f);  
	           i = 0; // restart  
	       }  
	       else{  
	           synchronized (f) { // 上锁  
	               if (tabAt(tab, i) == f) { // 其他线程没有在此期间操作f  
	                  Node<K,V> p = (fh >= 0 ? f :  
	                               (finstanceof TreeBin) ?  
	                               ((TreeBin<K,V>)f).first : null);  
	                   while (p != null) { // 首先删除链、树的末尾元素,避免产生大量垃圾  
	                       --delta;  
	                       p = p.next;  
	                   }  
	                   setTabAt(tab, i++, null); // 利用CAS无锁置null  
	               }  
	           }  
	       }  
	    }  
	    if (delta != 0L)  
	       addCount(delta, -1); // 无实际意义,参数check<=1,直接return。  
	}  

三、ConcurrentHashMap相关的一些问题

3.1 ConcurrentHashMap在JDK1.7和JDK1.8中的不同

JDK1.7JDK1.8
底层实现
同步机制分段锁,每个segment继承ReentrantLockCAS + synchronized保证并发更新
存储结构数组+链表数组+链表+红黑树
键值对HashEntryNode
put多个线程同时竞争获取同一个segment锁,获取成功的线程更新map;失败的线程尝试多次获取锁仍未成功,则挂起线程,等待释放锁访问相应的bucket时,使用sychronizeded关键字,防止多个线程同时操作同一个bucket,如果该节点的hash不小于0,则遍历链表更新节点或插入新节点;如果该节点是TreeBin类型的节点,说明是红黑树结构,则通过putTreeVal方法往红黑树中插入节点;更新了节点数量,还要考虑扩容和链表转红黑树
size统计每个Segment对象中的元素个数,然后进行累加,但是这种方式计算出来的结果并不一样的准确的。先采用不加锁的方式,连续计算元素的个数,最多计算3次:如果前后两次计算结果相同,则说明计算出来的元素个数是准确的;如果前后两次计算结果都不同,则给每个Segment进行加锁,再计算一次元素的个数通过累加baseCount和CounterCell数组中的数量,即可得到元素的总个数

??ConcurrentHashMap在JDK1.8中所造的改进主要是2方面。
??1)取消segments字段,直接采用transient volatile HashEntry<K,V>[] table保存数据,采用table数组元素作为锁,从而实现了对每一行数据进行加锁,进一步减少并发冲突的概率。
??2)将原先table数组+单向链表的数据结构,变更为table数组+单向链表+红黑树的结构。对于hash表来说,最核心的能力在于将key hash之后能均匀的分布在数组中。如果hash之后散列的很均匀,那么table数组中的每个队列长度主要为0或者1。但实际情况并非总是如此理想,虽然ConcurrentHashMap类默认的加载因子为0.75,但是在数据量过大或者运气不佳的情况下,还是会存在一些队列长度过长的情况,如果还是采用单向列表方式,那么查询某个节点的时间复杂度为O(n);因此,对于个数超过8(默认值)的列表,jdk1.8中采用了红黑树的结构,那么查询的时间复杂度可以降低到O(logN),可以改进性能。

3.2 ConcurrentHashMap能完全替代HashTable吗

??HashTable虽然性能上不如ConcurrentHashMap,但并不能完全被取代,两者的迭代器的一致性不同的,`HashTable的迭代器是强一致性的,ConcurrentHashMap是弱一致的·。 ConcurrentHashMap的get、clear、iterator 都是弱一致性的。

  • Hashtable的任何操作都会把整个表锁住,是阻塞的。好处是总能获取最实时的更新,比如说线程A调用putAll写入大量数据,期间线程B调用get,线程B就会被阻塞,直到线程A完成putAll,因此线程B肯定能获取到线程A写入的完整数据。坏处是所有调用都要排队,效率较低。
  • ConcurrentHashMap 是设计为非阻塞的。在更新时会局部锁住某部分数据,但不会把整个表都锁住。同步读取操作则是完全非阻塞的。好处是在保证合理的同步前提下,效率很高。坏处是严格来说读取操作不能保证反映最近的更新。例如线程A调用putAll写入大量数据,期间线程B调用get,则只能get到目前为止已经顺利插入的部分 数据。

??选择哪一个,是在性能与数据一致性之间权衡。ConcurrentHashMap适用于追求性能的场景,大多数线程都只做insert/delete操作,对读取数据的一致性要求较低。

3.3 SynchronizedMap和ConcurrentHashMap有什么区别?

??SynchronizedMap来源自Collections.synchronizedMap()方法。
??SynchronizedMap一次锁住整张表来保证线程安全,所以每次只能有一个线程来访为 map;ConcurrentHashMap使用分段锁(JDK1.6)来保证在多线程下的性能
??ConcurrentHashMap(JDK1.6)中则是一次锁住一个桶。ConcurrentHashMap默认将hash表分为16个桶,诸如 get、put、remove 等常用操作只锁当前需要用到的桶。这样,原来只能一个线程进入,现在却能同时有16个写线程执行,并发性能的提升是显而易见的。
??同时,ConcurrentHashMap使用了一种不同的迭代方式。在这种迭代方式中,当iterator被创建后集合再发生改变就不再是抛出ConcurrentModificationException,取而代之的是在改变时new新的数据从而不影响原有的数据,iterator完成后再将头指针替换为新的数据 ,这样iterator线程可以使用原来老的数据,而写线程也可以并发的完成改变。

3.4 ConcurrentHashMap的并发度是什么

??ConcurrentHashMap的并发度就是segment的大小,默认为16。这意味着最多同时可以有16条线程操作ConcurrentHashMap,这也是ConcurrentHashMap对Hashtable的最大优势。

3.5 为什么使用红黑树而不使用AVL树?

??ConcurrentHashMap在put的时候会加锁,使用红黑树插入速度更快,可以减少等待锁释放的时间。红黑树是对AVL树的优化,只要求部分平衡,用非严格的平衡来换取增删节点时候旋转次数的降低,提高了插入和删除的性能。

文章来源:https://blog.csdn.net/m0_37741420/article/details/135028798
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。