当前位置:主页 > 软件编程 > JAVA代码 >

深入学习java并发包ConcurrentHashMap源码

时间:2020-10-12 09:32:46 | 栏目:JAVA代码 | 点击:

正文

以前写过介绍HashMap的文章,文中提到过HashMap在put的时候,插入的元素超过了容量(由负载因子决定)的范围就会触发扩容操作,就是rehash,这个会重新将原数组的内容重新hash到新的扩容数组中,在多线程的环境下,存在同时其他的元素也在进行put操作,如果hash值相同,可能出现同时在同一数组下用链表表示,造成闭环,导致在get时会出现死循环,所以HashMap是线程不安全的。

JDK1.7的实现

整个 ConcurrentHashMap 由一个个 Segment 组成,Segment 代表”部分“或”一段“的意思,所以很多地方都会将其描述为分段锁。注意,行文中,我很多地方用了“槽”来代表一个 segment。

简单理解就是,ConcurrentHashMap 是一个 Segment 数组,Segment 通过继承 ReentrantLock 来进行加锁,所以每次需要加锁的操作锁住的是一个 segment,这样只要保证每个 Segment 是线程安全的,也就实现了全局的线程安全。

concurrencyLevel:并行级别、并发数、Segment 数。默认是 16,也就是说 ConcurrentHashMap 有 16 个 Segments,所以理论上,这个时候,最多可以同时支持 16 个线程并发写,只要它们的操作分别分布在不同的 Segment 上。这个值可以在初始化的时候设置为其他值,但是一旦初始化以后,它是不可以扩容的。

再具体到每个 Segment 内部,其实每个 Segment 很像之前介绍的 HashMap,不过它要保证线程安全,所以处理起来要麻烦些。

初始化
initialCapacity:初始容量,这个值指的是整个 ConcurrentHashMap 的初始容量,实际操作的时候需要平均分给每个 Segment。

loadFactor:负载因子,之前我们说了,Segment 数组不可以扩容,所以这个负载因子是给每个 Segment 内部使用的。

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
  int sshift = 0;
  int ssize = 1;
  // 计算并行级别 ssize,因为要保持并行级别是 2 的 n 次方
  while (ssize < concurrencyLevel) {
    ++sshift;
    ssize <<= 1;
  }
  // 我们这里先不要那么烧脑,用默认值,concurrencyLevel 为 16,sshift 为 4
  // 那么计算出 segmentShift 为 28,segmentMask 为 15,后面会用到这两个值
  this.segmentShift = 32 - sshift;
  this.segmentMask = ssize - 1;
  if (initialCapacity > MAXIMUM_CAPACITY)
    initialCapacity = MAXIMUM_CAPACITY;
  // initialCapacity 是设置整个 map 初始的大小,
  // 这里根据 initialCapacity 计算 Segment 数组中每个位置可以分到的大小
  // 如 initialCapacity 为 64,那么每个 Segment 或称之为"槽"可以分到 4 个
  int c = initialCapacity / ssize;
  if (c * ssize < initialCapacity)
    ++c;
  // 默认 MIN_SEGMENT_TABLE_CAPACITY 是 2,这个值也是有讲究的,因为这样的话,对于具体的槽上,
  // 插入一个元素不至于扩容,插入第二个的时候才会扩容
  int cap = MIN_SEGMENT_TABLE_CAPACITY; 
  while (cap < c)
    cap <<= 1;
  // 创建 Segment 数组,
  // 并创建数组的第一个元素 segment[0]
  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];
  // 往数组写入 segment[0]
  UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
  this.segments = ss;
}

初始化完成,我们得到了一个 Segment 数组。

我们就当是用 new ConcurrentHashMap() 无参构造函数进行初始化的,那么初始化完成后:

Segment

static class Segment<K,V> extends ReentrantLock implements Serializable {
  transient volatile HashEntry<K,V>[] table;
  transient int count;
  transient int modCount;
  
}

从上Segment的继承体系可以看出,Segment实现了ReentrantLock,也就带有锁的功能,table使用volatile修饰,保证了内存可见性。

put 过程分析

我们先看 put 的主流程,对于其中的一些关键细节操作,后面会进行详细介绍。

public V put(K key, V value) {
  Segment<K,V> s;
  if (value == null)
    throw new NullPointerException();
  // 1. 计算 key 的 hash 值
  int hash = hash(key);
  // 2. 根据 hash 值找到 Segment 数组中的位置 j
  //  hash 是 32 位,无符号右移 segmentShift(28) 位,剩下高 4 位,
  //  然后和 segmentMask(15) 做一次与操作,也就是说 j 是 hash 值的高 4 位,也就是槽的数组下标
  int j = (hash >>> segmentShift) & segmentMask;
  // 刚刚说了,初始化的时候初始化了 segment[0],但是其他位置还是 null,
  // ensureSegment(j) 对 segment[j] 进行初始化
  if ((s = (Segment<K,V>)UNSAFE.getObject     // nonvolatile; recheck
     (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
    s = ensureSegment(j);
  // 3. 插入新值到 槽 s 中
  return s.put(key, hash, value, false);
}

初始化槽: ensureSegment

ConcurrentHashMap 初始化的时候会初始化第一个槽 segment[0],对于其他槽来说,在插入第一个值的时候进行初始化。

这里需要考虑并发,因为很可能会有多个线程同时进来初始化同一个槽 segment[k],不过只要有一个成功了就可以。

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;
  if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
    // 这里看到为什么之前要初始化 segment[0] 了,
    // 使用当前 segment[0] 处的数组长度和负载因子来初始化 segment[k]
    // 为什么要用“当前”,因为 segment[0] 可能早就扩容过了
    Segment<K,V> proto = ss[0];
    int cap = proto.table.length;
    float lf = proto.loadFactor;
    int threshold = (int)(cap * lf);
    // 初始化 segment[k] 内部的数组
    HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
    if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
      == null) { // 再次检查一遍该槽是否被其他线程初始化了。
      Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
      // 使用 while 循环,内部用 CAS,当前线程成功设值或其他线程成功设值后,退出,如果其他线程成功设置后,这里获取到直接返回
      while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
          == null) {
        if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
          break;
      }
    }
  }
  return seg;
}

总的来说,ensureSegment(int k) 比较简单,对于并发操作使用 CAS 进行控制。

第一层很简单,根据 hash 值很快就能找到相应的 Segment,之后就是 Segment 内部的 put 操作了。

Segment 内部是由 数组+链表 组成的

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
  // 在往该 segment 写入前,需要先获取该 segment 的独占锁
  //  先看主流程,后面还会具体介绍这部分内容
  HashEntry<K,V> node = tryLock() ? null :
    scanAndLockForPut(key, hash, value);
  V oldValue;
  try {
    // 这个是 segment 内部的数组
    HashEntry<K,V>[] tab = table;
    // 再利用 hash 值,求应该放置的数组下标
    int index = (tab.length - 1) & hash;
    // first 是数组该位置处的链表的表头
    HashEntry<K,V> first = entryAt(tab, index);

    // 下面这串 for 循环虽然很长,不过也很好理解,想想该位置没有任何元素和已经存在一个链表这两种情况
    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,这个要看获取锁的过程,不过和这里都没有关系。
        // 如果不为 null,那就直接将它设置为链表表头;如果是null,初始化并设置为链表表头。
        if (node != null)
          node.setNext(first);
        else
          node = new HashEntry<K,V>(hash, key, value, first);

        int c = count + 1;
        // 如果超过了该 segment 的阈值,这个 segment 需要扩容
        if (c > threshold && tab.length < MAXIMUM_CAPACITY)
          rehash(node); // 扩容后面也会具体分析
        else
          // 没有达到阈值,将 node 放到数组 tab 的 index 位置,
          // 其实就是将新的节点设置成原链表的表头
          setEntryAt(tab, index, node);
        ++modCount;
        count = c;
        oldValue = null;
        break;
      }
    }
  } finally {
    // 解锁
    unlock();
  }
  return oldValue;
}

整体流程还是比较简单的,由于有独占锁的保护,所以 segment 内部的操作并不复杂。至于这里面的并发问题,我们稍后再进行介绍。

到这里 put 操作就结束了,接下来,我们说一说其中几步关键的操作。

获取写入锁: scanAndLockForPut

前面我们看到,在往某个 segment 中 put 的时候,首先会调用 node = tryLock() ? null : scanAndLockForPut(key, hash, value),也就是说先进行一次 tryLock() 快速获取该 segment 的独占锁,如果失败,那么进入到 scanAndLockForPut 这个方法来获取锁。

下面我们来具体分析这个方法中是怎么控制加锁的。

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) {
      if (e == null) {
        if (node == null) // speculatively create node
          // 进到这里说明数组该位置的链表是空的,没有任何元素
          // 当然,进到这里的另一个原因是 tryLock() 失败,所以该槽存在并发,不一定是该位置
          node = new HashEntry<K,V>(hash, key, value, null);
        retries = 0;
      }
      else if (key.equals(e.key))
        retries = 0;
      else
        // 顺着链表往下走
        e = e.next;
    }
    // 重试次数如果超过 MAX_SCAN_RETRIES(单核1多核64),那么不抢了,进入到阻塞队列等待锁
    //  lock() 是阻塞方法,直到获取锁后返回
    else if (++retries > MAX_SCAN_RETRIES) {
      lock();
      break;
    }
    else if ((retries & 1) == 0 &&
         // 这个时候是有大问题了,那就是有新的元素进到了链表,成为了新的表头
         //   所以这边的策略是,相当于重新走一遍这个 scanAndLockForPut 方法
         (f = entryForHash(this, hash)) != first) {
      e = first = f; // re-traverse if entry changed
      retries = -1;
    }
  }
  return node;
}

这个方法有两个出口,一个是 tryLock() 成功了,循环终止,另一个就是重试次数超过了 MAX_SCAN_RETRIES,进到 lock() 方法,此方法会阻塞等待,直到成功拿到独占锁。

这个方法就是看似复杂,但是其实就是做了一件事,那就是获取该 segment 的独占锁,如果需要的话顺便实例化了一下 node。

获取锁时,并不直接使用lock来获取,因为该方法获取锁失败时会挂起。事实上,它使用了自旋锁,如果tryLock获取锁失败,说明锁被其它线程占用,此时通过循环再次以tryLock的方式申请锁。如果在循环过程中该Key所对应的链表头被修改,则重置retry次数。如果retry次数超过一定值,则使用lock方法申请锁。

这里使用自旋锁是因为自旋锁的效率比较高,但是它消耗CPU资源比较多,因此在自旋次数超过阈值时切换为互斥锁。

扩容: rehash

重复一下,segment 数组不能扩容,扩容是 segment 数组某个位置内部的数组 HashEntry\<k,v>[] 进行扩容,扩容后,容量为原来的 2 倍。

首先,我们要回顾一下触发扩容的地方,put 的时候,如果判断该值的插入会导致该 segment 的元素个数超过阈值,那么先进行扩容,再插值,读者这个时候可以回去 put 方法看一眼。

该方法不需要考虑并发,因为到这里的时候,是持有该 segment 的独占锁的。

// 方法参数上的 node 是这次扩容后,需要添加到新的数组中的数据。
private void rehash(HashEntry<K,V> node) {
  HashEntry<K,V>[] oldTable = table;
  int oldCapacity = oldTable.length;
  // 2 倍
  int newCapacity = oldCapacity << 1;
  threshold = (int)(newCapacity * loadFactor);
  // 创建新数组
  HashEntry<K,V>[] newTable =
    (HashEntry<K,V>[]) new HashEntry[newCapacity];
  // 新的掩码,如从 16 扩容到 32,那么 sizeMask 为 31,对应二进制 ‘000...00011111'
  int sizeMask = newCapacity - 1;

  // 遍历原数组,老套路,将原数组位置 i 处的链表拆分到 新数组位置 i 和 i+oldCap 两个位置
  for (int i = 0; i < oldCapacity ; i++) {
    // e 是链表的第一个元素
    HashEntry<K,V> e = oldTable[i];
    if (e != null) {
      HashEntry<K,V> next = e.next;
      // 计算应该放置在新数组中的位置,
      // 假设原数组长度为 16,e 在 oldTable[3] 处,那么 idx 只可能是 3 或者是 3 + 16 = 19
      int idx = e.hash & sizeMask;
      if (next == null)  // 该位置处只有一个元素,那比较好办
        newTable[idx] = e;
      else { // Reuse consecutive sequence at same slot
        // e 是链表表头
        HashEntry<K,V> lastRun = e;
        // idx 是当前链表的头结点 e 的新位置
        int lastIdx = idx;

        // 下面这个 for 循环会找到一个 lastRun 节点,这个节点之后的所有元素是将要放到一起的
        for (HashEntry<K,V> last = next;
           last != null;
           last = last.next) {
          int k = last.hash & sizeMask;
          if (k != lastIdx) {
            lastIdx = k;
            lastRun = last;
          }
        }
        // 将 lastRun 及其之后的所有节点组成的这个链表放到 lastIdx 这个位置
        newTable[lastIdx] = lastRun;
        // 下面的操作是处理 lastRun 之前的节点,
        //  这些节点可能分配在另一个链表中,也可能分配到上面的那个链表中
        for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
          V v = p.value;
          int h = p.hash;
          int k = h & sizeMask;
          HashEntry<K,V> n = newTable[k];
          newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
        }
      }
    }
  }
  // 将新来的 node 放到新数组中刚刚的 两个链表之一 的 头部
  int nodeIndex = node.hash & sizeMask; // add the new node
  node.setNext(newTable[nodeIndex]);
  newTable[nodeIndex] = node;
  table = newTable;
}

总结一下put的流程:

当执行put操作时,会进行第一次key的hash来定位Segment的位置,如果该Segment还没有初始化,即通过CAS操作进行赋值,然后进行第二次hash操作,找到相应的HashEntry的位置,这里会利用继承过来的锁的特性,在将数据插入指定的HashEntry位置时(链表的尾端),会通过继承ReentrantLock的tryLock()方法尝试去获取锁,如果获取成功就直接插入相应的位置,如果已经有线程获取该Segment的锁,那当前线程会以自旋的方式去继续的调用tryLock()方法去获取锁,超过指定次数就挂起,等待唤醒。

get 过程分析

相对于 put 来说,get 真的不要太简单。

1.计算 hash 值,找到 segment 数组中的具体位置,或我们前面用的“槽”

2.槽中也是一个数组,根据 hash 找到数组中具体的位置

3.到这里是链表了,顺着链表进行查找即可

public V get(Object key) {
  Segment<K,V> s; // manually integrate access methods to reduce overhead
  HashEntry<K,V>[] tab;
  // 1. hash 值
  int h = hash(key);
  long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
  // 2. 根据 hash 找到对应的 segment
  if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
    (tab = s.table) != null) {
    // 3. 找到segment 内部数组相应位置的链表,遍历
    for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
         (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
       e != null; e = e.next) {
      K k;
      if ((k = e.key) == key || (e.hash == h && key.equals(k)))
        return e.value;
    }
  }
  return null;
}

size操作

put、remove和get操作只需要关心一个Segment,而size操作需要遍历所有的Segment才能算出整个Map的大小。一个简单的方案是,先锁住所有Sgment,计算完后再解锁。但这样做,在做size操作时,不仅无法对Map进行写操作,同时也无法进行读操作,不利于对Map的并行操作。

为更好支持并发操作,ConcurrentHashMap会在不上锁的前提逐个Segment计算3次size,如果某相邻两次计算获取的所有Segment的更新次数(每个Segment都与HashMap一样通过modCount跟踪自己的修改次数,Segment每修改一次其modCount加一)相等,说明这两次计算过程中无更新操作,则这两次计算出的总size相等,可直接作为最终结果返回。如果这三次计算过程中Map有更新,则对所有Segment加锁重新计算Size。该计算方法代码如下

public int size() {
 final Segment<K,V>[] segments = this.segments;
 int size;
 boolean overflow; // true if size overflows 32 bits
 long sum;     // sum of modCounts
 long last = 0L;  // previous sum
 int retries = -1; // first iteration isn't retry
 try {
  for (;;) {
   if (retries++ == RETRIES_BEFORE_LOCK) {
    for (int j = 0; j < segments.length; ++j)
     ensureSegment(j).lock(); // force creation
   }
   sum = 0L;
   size = 0;
   overflow = false;
   for (int j = 0; j < segments.length; ++j) {
    Segment<K,V> seg = segmentAt(segments, j);
    if (seg != null) {
     sum += seg.modCount;
     int c = seg.count;
     if (c < 0 || (size += c) < 0)
      overflow = true;
    }
   }
   if (sum == last)
    break;
   last = sum;
  }
 } finally {
  if (retries > RETRIES_BEFORE_LOCK) {
   for (int j = 0; j < segments.length; ++j)
    segmentAt(segments, j).unlock();
  }
 }
 return overflow ? Integer.MAX_VALUE : size;
}

ConcurrentHashMap的Size方法是一个嵌套循环,大体逻辑如下:

1.遍历所有的Segment。

2.把Segment的元素数量累加起来。

3.把Segment的修改次数累加起来。

4.判断所有Segment的总修改次数是否大于上一次的总修改次数。如果大于,说明统计过程中有修改,重新统计,尝试次数+1;如果不是。说明没有修改,统计结束。

5.如果尝试次数超过阈值,则对每一个Segment加锁,再重新统计。

6.再次判断所有Segment的总修改次数是否大于上一次的总修改次数。由于已经加锁,次数一定和上次相等。

7.释放锁,统计结束。

并发问题分析

现在我们已经说完了 put 过程和 get 过程,我们可以看到 get 过程中是没有加锁的,那自然我们就需要去考虑并发问题。

添加节点的操作 put 和删除节点的操作 remove 都是要加 segment 上的独占锁的,所以它们之间自然不会有问题,我们需要考虑的问题就是 get 的时候在同一个 segment 中发生了 put 或 remove 操作。

put 操作的线程安全性

remove 操作的线程安全性

最后我们来看看并发操作示意图

Case1:不同Segment的并发写入

不同Segment的写入是可以并发执行的。

Case2:同一Segment的一写一读

同一Segment的写和读是可以并发执行的。

Case3:同一Segment的并发写入

Segment的写入是需要上锁的,因此对同一Segment的并发写入会被阻塞。

由此可见,ConcurrentHashMap当中每个Segment各自持有一把锁。在保证线程安全的同时降低了锁的粒度,让并发操作效率更高。

您可能感兴趣的文章:

相关文章