1.线程安全集合类

1>.线程安全集合类可以分为三大类:

①.遗留的(/旧的)线程安全集合,如:Hashtable,Vector;

②.使用Collections装饰的线程安全集合,如:

• Collections.synchronizedCollection
• Collections.synchronizedList
• Collections.synchronizedMap
• Collections.synchronizedSet
• Collections.synchronizedNavigableMap
• Collections.synchronizedNavigableSet
• Collections.synchronizedSortedMap
• Collections.synchronizedSortedSet
  
  

③.java.util.concurrent.*;

2>.java.util.concurrent.*下的线程安全集合类,可以发现它们有规律,里面包含三类关键词:Blocking、CopyOnWrite、Concurrent:

①.Blocking大部分实现基于锁,并提供用来阻塞的方法;

②.CopyOnWrite之类容器修改开销相对较重;

③.Concurrent类型的容器:

• 内部很多操作使用CAS优化,一般可以提供较高吞吐量;
• 弱一致性;

• .遍历时弱一致性,例如,当利用迭代器遍历时,如果容器发生修改,迭代器仍然可以继续进行遍历,这时内容是旧的;
• 求大小弱一致性,size 操作未必是 100% 准确;
• 读取弱一致性;

2.ConcurrentHashMap

1>.案例: 单词计数

public class ConcurrentHashMapDemo {static final String ALPHA = "abcedfghijklmnopqrstuvwxyz";public static void main(String[] args) {generialData();// 这是错误的!!!// demo(//         // 创建 map 集合//         // 创建 ConcurrentHashMap对不对?//         // 不对,调用集合中单个线程安全的方法是没有问题的,如果调用了集合中多个线程安全的方法就会出现线程安全问题//         () -> new ConcurrentHashMap(),//         // 进行计数//         (map, words) -> {//             for (String word : words) {//                 Integer counter = map.get(word);  //方法1//                 int newValue = counter == null ? 1 : counter + 1;//                 map.put(word, newValue);  //方法2//             }//         }// );demo(// 创建 map 集合// 使用原子变量() -> new ConcurrentHashMap(),// 进行计数(map, words) -> {for (String word : words) {// 注意: 不能使用putIfAbsent,此方法返回的是上一次的value,首次调用返回null!map.computeIfAbsent(word, (key) -> new LongAdder()).increment();}});// // 或者// demo(//         () -> new ConcurrentHashMap(),//         (map, words) -> {//             for (String word : words) {//                 // 函数式编程,无需原子变量//                 map.merge(word, 1, Integer::sum);//             }//         }// );}// 生成测试数据private static void generialData() {int length = ALPHA.length();int count = 200;List list = new ArrayList<>(length * count);for (int i = 0; i < length; i++) {char ch = ALPHA.charAt(i);for (int j = 0; j < count; j++) {list.add(String.valueOf(ch));}}Collections.shuffle(list);for (int i = 0; i < 26; i++) {try (PrintWriter out = new PrintWriter(new OutputStreamWriter(new FileOutputStream("F://" + (i + 1) + ".txt")))) {String collect = list.subList(i * count, (i + 1) * count).stream().collect(Collectors.joining("\n"));out.print(collect);} catch (IOException e) {e.printStackTrace();}}}// 单词计数private static  void demo(Supplier> supplier, BiConsumer, List> consumer) {Map counterMap = supplier.get();List ts = new ArrayList<>();for (int i = 1; i <= 26; i++) {int idx = i;Thread thread = new Thread(() -> {List words = readFromFile(idx);consumer.accept(counterMap, words);});ts.add(thread);}ts.forEach(t -> t.start());ts.forEach(t -> {try {t.join();} catch (InterruptedException e) {e.printStackTrace();}});System.out.println(counterMap);}// 读取数据文件public static List readFromFile(int i) {ArrayList words = new ArrayList<>();try (BufferedReader in = new BufferedReader(new InputStreamReader(new FileInputStream("F://"+ i + ".txt")))) {while (true) {String word = in.readLine();if (word == null) {break;}words.add(word);}return words;} catch (IOException e) {throw new RuntimeException(e);}}
}

3.JDK7 HashMap并发死链

前言: JDK7中HashMap并发死链现象

①.JDK7中HashMap的底层数据结构为数组+(单)链表;

②.JDK7中HashMap里面后加入的元素总是会放在链表头部位置;

③.JDK7中HashMap的并发死链发生在扩容时

• JDK7中HashMap的扩容时机: 集合中元素个数超过了容量的3/4;
• JDK7中HashMap扩容的时链表中后加入的元素放入链表头,因此链表就倒过来了,某个线程完成扩容之后,其他线程扩容时又要将之前线程扩容之后的集合(链表)中的元素的存储顺序打乱了,造成了死循环;

HashMap 的并发死链源码:

// 将table迁移至newTable
void transfer(Entry[] newTable, boolean rehash) {int newCapacity = newTable.length;for (Entry e : table) {while(null != e) {Entry next = e.next;// 1 处if (rehash) {e.hash = null == e.key ? 0 : hash(e.key);}int i = indexFor(e.hash, newCapacity);// 2 处// 将新元素加入 newTable[i], 原 newTable[i] 作为新元素的 nexte.next = newTable[i];newTable[i] = e;e = next;}}}

分析:

①.原始链表,格式：[下标] (key,next),如:[1] (1,35)->(35,16)->(16,null);

②.线程a执行到1处,此时局部变量e为(1,35),而局部变量next为(35,16)线程a挂起;

③.线程b开始执行:

• 第一次循环

[1] (1,null)

• 第二次循环

[1] (35,1)->(1,null)

• 第三次循环

[1] (35,1)->(1,null)
[17] (16,null)

④.切换回线程a,此时局部变量e和next被恢复,引用没变但内容变了:e的内容被改为(1,null),而next的内容被改为(35,1)并链向(1,null);

• 第一次循环

[1] (1,null)

• 第二次循环,注意这时e是(35,1)并链向(1,null),所以next又是(1,null)

[1] (35,1)->(1,null)

• 第三次循环,e是(1,null),而next是null,但e被放入链表头,这样e.next变成了35(2 处)

[1] (1,35)->(35,1)->(1,35)

此时已经是死链了…

小结:

①.究其原因,是因为在多线程环境下使用了非线程安全的 map 集合;

②.JDK8虽然将扩容算法做了调整,不再将后加入的元素放到链表头(而是保持与扩容前一样的顺序),但仍不意味着能够在多线程环境下能够安全扩容,还会出现其它问题(如扩容丢数据);

4.JDK7中ConcurrentHashMap原理

4.1.概述

1>.ConcurrentHashMap维护了一个 segment数组,每个segment对应一把锁;

2>.优点:

如果多个线程访问不同的 segment,实际是没有冲突的;

3>.缺点:

Segments数组默认大小为16,这个容量初始化指定后就不能改变了,并且不是懒惰初始化;

4.2.构造器

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;// ssize 必须是2^n, 即2, 4, 8, 16 ... 表示了segments数组的大小int sshift = 0;int ssize = 1;while (ssize < concurrencyLevel) {++sshift;ssize <<= 1;}// segmentShift默认是32 - 4 = 28this.segmentShift = 32 - sshift;// segmentMask默认是15,即0000 0000 0000 1111this.segmentMask = ssize - 1;if (initialCapacity > MAXIMUM_CAPACITY)initialCapacity = MAXIMUM_CAPACITY;int c = initialCapacity / ssize;if (c * ssize < initialCapacity)++c;int cap = MIN_SEGMENT_TABLE_CAPACITY;while (cap < c)cap <<= 1;// 创建 segments and segments[0]Segment s0 = new Segment(loadFactor, (int)(cap * loadFactor), (HashEntry[])new HashEntry[cap]);Segment[] ss = (Segment[])new Segment[ssize];UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]this.segments = ss;
}

构造完成,如下图所示:

可以看到ConcurrentHashMap没有实现懒惰初始化,空间占用不友好;其中this.segmentShift和this.segmentMask的作用是决定将key的hash结果匹配到哪个segment;

例如,根据某一hash值求segment位置,先将高位向低位移动this.segmentShift位:

结果再与this.segmentMask做位于运算,最终得到1010即下标为10的segment:

4.3.put流程

public V put(K key, V value) {Segment s;if (value == null)throw new NullPointerException();int hash = hash(key);// 计算出 segment 下标int j = (hash >>> segmentShift) & segmentMask;// 获得 segment 对象, 判断是否为 null, 是则创建该 segmentif ((s = (Segment)UNSAFE.getObject(segments, (j << SSHIFT) + SBASE)) == null) {// 这时不能确定是否真的为 null, 因为其它线程也发现该 segment 为 null,// 因此在 ensureSegment 里用 cas 方式保证该 segment 安全性s = ensureSegment(j);}// 进入 segment 的put 流程return s.put(key, hash, value, false);}

segment继承了可重入锁(ReentrantLock),它的put方法如下:

final V put(K key, int hash, V value, boolean onlyIfAbsent) {// 尝试加锁// 如果不成功, 进入 scanAndLockForPut 流程// 如果是多核 cpu 最多 tryLock 64 次, 进入 lock 流程// 在尝试期间, 还可以顺便看该节点在链表中有没有, 如果没有顺便创建出来HashEntry node = tryLock() ? null : scanAndLockForPut(key, hash, value);// 执行到这里 segment 已经被成功加锁, 可以安全执行V oldValue;try {HashEntry[] tab = table;int index = (tab.length - 1) & hash;HashEntry first = entryAt(tab, index);for (HashEntry 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 {// 新增// 1) 之前等待锁时, node 已经被创建, next 指向链表头if (node != null)node.setNext(first);else// 2) 创建新 nodenode = new HashEntry(hash, key, value, first);int c = count + 1;// 3) 扩容if (c > threshold && tab.length < MAXIMUM_CAPACITY)rehash(node);else// 将 node 作为链表头setEntryAt(tab, index, node);++modCount;count = c;oldValue = null;break;}}} finally {unlock();}return oldValue;}

4.4.rehash流程

扩容操作发生在put中,因为此时已经获得了锁,因此rehash时不需要考虑线程安全;

private void rehash(HashEntry node) {HashEntry[] oldTable = table;int oldCapacity = oldTable.length;int newCapacity = oldCapacity << 1;threshold = (int)(newCapacity * loadFactor);HashEntry[] newTable =(HashEntry[]) new HashEntry[newCapacity];int sizeMask = newCapacity - 1;for (int i = 0; i < oldCapacity ; i++) {HashEntry e = oldTable[i];if (e != null) {HashEntry next = e.next;int idx = e.hash & sizeMask;if (next == null) // Single node on listnewTable[idx] = e;else { // Reuse consecutive sequence at same slotHashEntry lastRun = e;int lastIdx = idx;// 过一遍链表, 尽可能把 rehash 后 idx 不变的节点重用for (HashEntry last = next;last != null;last = last.next) {int k = last.hash & sizeMask;if (k != lastIdx) {lastIdx = k;lastRun = last;}}newTable[lastIdx] = lastRun;// 剩余节点需要新建for (HashEntry p = e; p != lastRun; p = p.next) {V v = p.value;int h = p.hash;int k = h & sizeMask;HashEntry n = newTable[k];newTable[k] = new HashEntry(h, p.key, v, n);}}}}// 扩容完成,才加入新的节点int nodeIndex = node.hash & sizeMask; // add the new nodenode.setNext(newTable[nodeIndex]);newTable[nodeIndex] = node;// 替换为新的 HashEntry tabletable = newTable;}

4.5.get流程

get时并未加锁,用了UNSAFE方法保证了可见性,扩容过程中,get先发生就从旧表取内容,get后发生就从新表取内容;

public V get(Object key) {Segment s; // manually integrate access methods to reduce overheadHashEntry[] tab;int h = hash(key);// u 为 segment 对象在数组中的偏移量long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;// s 即为 segmentif ((s = (Segment)UNSAFE.getObjectVolatile(segments, u)) != null &&(tab = s.table) != null) {for (HashEntry e = (HashEntry) 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;}

4.6.size计算流程

1>.计算元素个数前,先不加锁计算两次,如果前后两次结果如一样,认为个数正确返回;

2>.如果不一样,进行重试,重试超过3次,将所有segment锁住,重新计算个数返回;

public int size() {// Try a few times to get accurate count. On failure due to// continuous async changes in table, resort to locking.final Segment[] segments = this.segments;int size;boolean overflow; // true if size overflows 32 bitslong sum; // sum of modCountslong last = 0L; // previous sumint retries = -1; // first iteration isn't retrytry {for (; ; ) {if (retries++ == RETRIES_BEFORE_LOCK) {// 超过重试次数, 需要创建所有 segment 并加锁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 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;}