JUC(七)

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<String, Integer>(),
        //         // 进行计数
        //         (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<String, LongAdder>(),
                // 进行计数
                (map, words) -> {
                    for (String word : words) {
                        // 注意: 不能使用putIfAbsent,此方法返回的是上一次的value,首次调用返回null!
                        map.computeIfAbsent(word, (key) -> new LongAdder()).increment();
                    }
                }
        );

        // // 或者
        // demo(
        //         () -> new ConcurrentHashMap<String, Integer>(),
        //         (map, words) -> {
        //             for (String word : words) {
        //                 // 函数式编程,无需原子变量
        //                 map.merge(word, 1, Integer::sum);
        //             }
        //         }
        // );
    }

    // 生成测试数据
    private static void generialData() {
        int length = ALPHA.length();
        int count = 200;

        List<String> 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 <V> void demo(Supplier<Map<String, V>> supplier, BiConsumer<Map<String, V>, List<String>> consumer) {
        Map<String, V> counterMap = supplier.get();
        List<Thread> ts = new ArrayList<>();
        for (int i = 1; i <= 26; i++) {
            int idx = i;
            Thread thread = new Thread(() -> {
                List<String> 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<String> readFromFile(int i) {
        ArrayList<String> 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<K,V> e : table) {
            while(null != e) {
                Entry<K,V> 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] 作为新元素的 next
                e.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 = 28
        this.segmentShift = 32 - sshift;
        // segmentMask默认是15,即0000 0000 0000 1111
        this.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<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];
        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<K,V> s;
        if (value == null)
            throw new NullPointerException();
        int hash = hash(key);
        // 计算出 segment 下标
        int j = (hash >>> segmentShift) & segmentMask;

        // 获得 segment 对象, 判断是否为 null, 是则创建该 segment
        if ((s = (Segment<K,V>)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<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value);

        // 执行到这里 segment 已经被成功加锁, 可以安全执行
        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;
                    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) 创建新 node
                        node = new HashEntry<K,V>(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<K,V> node) {
        HashEntry<K,V>[] oldTable = table;
        int oldCapacity = oldTable.length;
        int newCapacity = oldCapacity << 1;
        threshold = (int)(newCapacity * loadFactor);
        HashEntry<K,V>[] newTable =
                (HashEntry<K,V>[]) new HashEntry[newCapacity];
        int sizeMask = newCapacity - 1;
        for (int i = 0; i < oldCapacity ; i++) {
            HashEntry<K,V> e = oldTable[i];
            if (e != null) {
                HashEntry<K,V> next = e.next;
                int idx = e.hash & sizeMask;
                if (next == null) // Single node on list
                    newTable[idx] = e;
                else { // Reuse consecutive sequence at same slot
                    HashEntry<K,V> lastRun = e;
                    int lastIdx = idx;
                    // 过一遍链表, 尽可能把 rehash 后 idx 不变的节点重用
                    for (HashEntry<K,V> last = next;
                         last != null;
                         last = last.next) {
                        int k = last.hash & sizeMask;
                        if (k != lastIdx) {
                            lastIdx = k;
                            lastRun = last;
                        }
                    }
                    newTable[lastIdx] = 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);
                    }
                }
            }
        }

        // 扩容完成,才加入新的节点
        int nodeIndex = node.hash & sizeMask; // add the new node
        node.setNext(newTable[nodeIndex]);
        newTable[nodeIndex] = node;

        // 替换为新的 HashEntry table
        table = newTable;
    }

4.5.get流程

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

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

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<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) {
                    // 超过重试次数, 需要创建所有 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<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;
    }