Java Util Concurrent并发编程（二）集合类的

2020-11-08 本文已影响0人唯有努力不欺人丶

我们背书大多数都背过，甚至HashMap，ArrayList不安全人人都能说，但是具体为什么呢？不安全的点在哪里？怎么解决这个问题？下面一一详细说。

集合类不安全

其实我们稍微了解基础知识的，都应该知道HashMap,HashSet,ArrayList是线程不安全的。然后更扎实一点的还能背出来，HashTable，Vector，ConcurrentHashMap是线程安全的（我也能背出来，手动滑稽）
但是说真的。这些都是死记硬背的知识点。我们真正去用代码证明过么？

ArrayList

下面的代码可以看出ArrayList在多线程下不安全:

public class D {

    public static void main(String[] args) {
        List<String> list = new ArrayList<String>();
        for(int i = 0;i<100;i++) {          
            new Thread(()->{
                list.add(UUID.randomUUID().toString().substring(0, 10));
                System.out.println(list);
            }).start();
        }
    }

}

很简单的代码逻辑，一个ArrayList。100个线程往里添加元素的同时输出这个集合（这个输出代码一定要写）。然后回报错如下图:

ArrayList不安全
其实这个错误大家也可以好好看看，强烈建议大家一定去看官方手册来找到答案，不要随便百度搜索，看到一个解释就完了。毕竟你不知道写这个解释的到底是什么水平。

对这个异常的描述
虽然解决这个问题最简单的方法就是把ArrayList改成Vector。但是！Vector比ArrayList出现的还要早。如果Vector真的那么好干嘛还要后发明一个ArrayList呢？这个巨大的性能差异还是不能避免的。更好的做法是让ArrayList变得安全。说到这就不得不提一下集合工具类Collections啦。看下图：

Collections工具
所以我们这里用Collections,如下代码:

image.png
当然了，上面两个和JUC关系不大。我们这里主要讲JUC的办法，继续去官方手册找。然后找到下图：

JUC中线程安全的ArrayList
所以我们用这个类也是可以的。

image.png
为什么这个CopyOnWrite就是安全的呢？这个是一个计算机程序设计领域的一个优化策略，简称COW（写入时复制）。在add方法中拿过来的时候就复制一个（长度+1），然后添加操作完成后，再把整个数组替换回去。
下面的CopyOnWriteArrayList的add代码：

注意全程用Lock锁
其实我们觉得CopyOnWriteArrayList比Vector好的原因也是如此，我们再去看看Vector的源码:

Vector用synchronized锁
这个也是这两个的最大区别。虽然都是线程安全的，但是实现的方式却不一样。（好多材料说现在synchronized现在性能已经很好了。但是当年synchronized确实曾经是最笨重的锁）。

HashSet

说完了List然后Set也是必不可少的。但是List和Set到底是什么关系呢？光说没用。我们得从源码上看。附上集合类关系图。

List，Set，Queue是同等级的
然后我们把刚刚对ArrayList做的操作同样对HashSet也实现一遍（这里是因为Set中HashSet是比较常用的）。

    public static void main(String[] args) {
        Set<String> set = new HashSet<String>();
        for(int i = 0;i<100;i++) {          
            new Thread(()->{                
                    set.add(UUID.randomUUID().toString().substring(0, 10));
                    System.out.println(set);

            }).start();
        }
    }

运行结果图

就连报错都和ArrayList的问题是一样的。所以就不多说了。但是具体的解决办法是什么呢？思路也可以参考上面的思路：

找别的类型代替，勉强实现（但是这个很可惜，set家族没有线程安全的。所以这个思路pass）
Collections工具类中让其变为线程安全的。（如果你ArrayList是照着上面都敲了一遍的，那么在Collections.synchronized的时候绝对是能看到Set也是有这样的方法的。）
JUC的方式实现。

上面说了，思路1在set中不存在，所以直接从思路2开始：

Collections工具类解决Set线程安全问题

接下来JUC的方式，如果刚刚你看手册但凡用点心，其实也能看到的CopyOnWriteArraySet的，附上截图：

Set的JUC形式

代码实现

到这里Set其实就可以过了。甚至于Map大家应该也都知道了。而且HashSet的底层是HashMap大家应该都知道吧，毕竟名字都差不多。如果不知道的其实去看一下HashSet的源码就可以了。附上代码截图。
HashSet的无参构造函数里面就只做了一件事。创建了一个HashMap。

HashSet无参构造函数
而HashSet中add的本质就是往这个HashMap中添加了一个key为给定元素的kv对。（感兴趣的小朋友可以去看下HashSet的源码，一共才三百多行。除了注解啥的，几乎就是对HashMap的简单操作。）

HashSet中add

HashMap

刚刚大家也都看到了，HashSet的本质就是HashMap。所以HashMap要专门讲讲，而且还挺有意思的。

Map家族。

Map家族概览

上面的图片是所以实现了Map接口的类或接口。我这里用不同颜色的框框框起来了。注意其中好多莫名其妙工具包里的实现类，我们暂时不要去专研它。比较有意思的是我用黄色框起来的哪个类，是JSONObject类，这个是fastjson工具包中的，算是乱入吧。不过JSON的本质就是kv对，这个其实也好理解。
继续说，其实这里注意一下，Map中是有线程安全的Map的。比如HashTable，或者说SynchronizedMap。不过总而言之还是HashMap用的最多。所以这里着重讲一下HashMap。
下面看一下HashMap的无参构造函数：

HashMap的无参构造

大家可以简单看一下（反正简单看完也看不懂），这个涉及到Hash的一些计算。而这个默认的Load_factor是加载因子。感兴趣的可以去看看Hash的原理或者专门讲解HashMap的，毕竟这个不是一句两句能说清的。我们要知道下面两行代码是等价的：

Map<String, String> map = new HashMap<String, String>();
Map<String, String> map = new HashMap<String, String>(16，0.75);

然后我们继续说HashMap的不安全：

HashMap多线程时报错

这个问题和上面两个一模一样，所以不多说了，开始分析解决办法：

别的数据结构代替（我们知道HashTable是线程安全的，可以用这个代替）

HashTable线程安全，不报错
Collections工具类中的办法（这个我们在上面看Map家族的时候其实就看到过这里有个SynchronizedMap。所以也不用多说）

Collections中线程安全的list
这个才是今天的重点，JUC中的线程安全的Map！这里很重要，不要被上面的欺骗了，就死脑筋去找CopyOnWriteMap。我们刚刚其实看Map家族的时候是看到了没有这个类的。而JUC中的Map，我用紫色框框框起来的。叫做ConcurrentHashMap。（所以上面那一个Map家族的截图是有多重要。自己仔细看看其实能知道好多东西）
至于ConcurrentHashMap为什么是线程安全的。按照我们的思路应该去官方手册和源码上找答案了（因为这个比较长，所以用代码的形式贴过来）：

    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) {
        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();
            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
            }
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            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) {
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        addCount(1L, binCount);
        return null;
    }

其实这个ConcurrentHashMap是调用了一个putVal的方法。
我们一行代码一行代码的分析：其实第一行代码就能看出来，ConcurrentHashMap的key和value都不是能null。不然直接报错了。
剩下的代码其实挺生涩的，我反正自认也算是学过hash源码也看的贼吃力。不过看看又不要钱，尤其是看到不懂的变量，点进去肯定是必然操作啊。第一个要查看的就是table，居然是volatile的，有点小收获啊：

table是volatile的

这个table是volatile的知道了，继续往下走代码,额，一个判空判断：tab是空则初始化。这个应该是能理解的。至于为什么设计在put里初始化（准确的说是第一次put初始化）咱也不知道，不过如果多个线程同时put那么这个初始化不就不安全了么？所以这个初始化应该也不是这么简单吧？去瞅瞅怎么初始化的：
这个还有点小复杂。但是都是用字符写的，只要肯用心分析还是能有那么点收获的。这个代码中有一个Thread.yield(); 注意:JUC本身就是多线程中的安全。所以涉及到线程的操作都要注意一下。为啥就让出CPU时间片了？就因为sizeCtl这个值小于0？仔细瞅瞅，原来当有代码进入到下面了就自动把sizeCtl改成-1了。所以当sizeCtl小于0说明有别的线程已经进入了，那确实这么处理没啥问题了。（这块我是对着代码一行一行扣的，如果理解有误欢迎指出！）
反正代码看的糊里糊涂，理解个大概。如果走到初始化说明是第一个元素，根本不存在并发问题，所以没啥说的。

初始化代码
如果是已经初始化完了的继续走if else。第二个分支是node里面是空，那么直接正常插入，也没啥好说的。第三个分支是不是在扩容。第四个分支是里面有元素，没扩容的正差插入。
这里我们就很容易看懂了，既然是想做到线程安全。锁绝对是不可避免的啊，粗略扫一下这个代码。那么大的synchronized (f) 就明晃晃的挂在那里。所以线程安全是必然了吧？
至此，我们算是勉强理解了ConcurrentHashMap是线程安全的。

Callable

这是一个JUC中的接口，想去了解他最好的途径就是官方手册+源码。我们先去手册中找到他：

Callable接口文档

通过以上的官方文档中的叙述，我们可以得到一下三点：

可以有返回值
可以抛出异常
方法是call（）而不是run（）。

下面我们用代码去更加了解它:

callable源码

这个callable的源码有点简单啊。有个泛型。泛型类型是call（）方法的返回值。上面说了callable的特点就是可以有返回值。现在我们理出来了，返回值的类型是我们自己定义的。而且Callable本身也不是线程，和Runnable和Thread没有直接关系。要怎么使用呢？我们得一层一层找关系。
中间么找的不说了，我最近特别喜欢翻各种类/接口的api文档，指不定就发现什么小秘密了呢。这里话题转过来，直接跳到一个类里：

FutureTask
这个类名字翻译过来也挺有意思，未来任务。就是可以创建任务的？不知道是不是可以这么理解，主要是看这个类能把runnable和callable挂上关系。其实这个可以理解为适配模式。这个FutureTask就是一个适配类。
下面的代码：

image.png

值得说一下的这个是get方法。这个方法可以有参数，是超时时间，就是多久没获取到这个值就报超时错误！get的值是callable中call方法返回的值。如果在这个方法执行时间很长的话，最后把get放到最后。

get方法
正常返回的代码（call先打印，过两秒钟打印ok）

public class D {

    public static void main(String[] args) throws Exception {
        FutureTask<String> futureTask = new FutureTask<String>(new myThread());
        new Thread(futureTask).start();
        System.out.println(futureTask.get());
    }

}
class myThread implements Callable<String>{
    @Override
    public String call() throws Exception {     
        System.out.println("call");
        TimeUnit.SECONDS.sleep(2);
        return "ok";
    }   
}

get超时报错

还有一点，这个Future是有缓存的。附上测试截图：

结果不变直接去缓存

JUC中的辅助类

CountDownLatch

老规矩，学习要么对手册要么对源码。我们先去手册上找到这个类，如下图：

CountDownLatch

我圈起来是我我个人看后觉得应该记住的点，首先这个类就是一个计数器。其次是一次性的。甚至这里提到了另一个类我们以后再说。而且这个类的api少的离谱，就是一个等待，设置时间的等待，获取当前计数。计数-1几个方法:

CountDownLatch的api
看上去也不难，让我们在代码中去使用它。下面是测试代码:

    public static void main(String[] args) throws Exception {
            CountDownLatch countDownLatch = new CountDownLatch(5);
            for(int i = 0;i<5;i++) {
                new Thread(()->{                    
                    System.out.println(Thread.currentThread().getName()+"执行了");
                    countDownLatch.countDown();
                },""+i).start();;
            }
//          countDownLatch.await();
            System.out.println("计数器归零！");
    }

如果我放开被注释掉的代码，这个计数器归零的语句一定是最后打印的。因为只有五个线程都执行完了才会往下执行这个输出语句。但是我如果注释掉这句那么就不会等线程都输出完了。这个结果就是随机的了，一切皆有可能。感兴趣的可以自己去敲一下，跑一跑。

CyclicBarrier

这个类怎么说呢，但凡上面你认真点看了，就会发现这个类出现在了CountDownLatch的介绍中了，我当时还提了一下。
官方手册里的介绍比较晦涩。简单来说CountDownLatch如果是减法计数器，可以把CyclicBarrier看做是加法计数器。大概思路就是每有一个线程等待计数器+1.当等待线程达到给定的线程数后，可以执行传入CyclicBarrier中的Run方法了。而且这个计数器是可重置的，下面的代码测试：

    public static void main(String[] args) throws Exception {
            CyclicBarrier cyclicBarrier = new CyclicBarrier(5,()->{
                System.out.println("5个等待线程唤醒了这个方法！");
            });
            for(int i = 0;i<5;i++) {
                final Integer temp = i;
                new Thread(()->{                                                    
                    System.out.println(Thread.currentThread().getName()+"执行了,当前syclicBarrier的线程等待数："+temp);
                    try {
                        cyclicBarrier.await();
                    } catch (Exception e) {
                        e.printStackTrace();
                    } 
                    
                },""+i).start();;
            }
    }

如上代码。因为5个线程在等待了会触发那个唤醒方法。下面是运行结果：

运行结果

这里像不像集齐七颗龙珠召唤神龙啊，哈哈。至于上面Demo中的temp变量是为了在lambda中可以用i这个值。所以用final属性修饰了temp。这个是作用域问题。也没啥好说的。当然了这个await也是有等待时间的。就跟拼夕夕规定时间拼不到人会失败一下。感兴趣的自己去看api文档。继续往下说。

Semaphore

这个类也是一个工具类。其实有一个观点;工具类不可能很复杂的。哪怕实现逻辑很复杂，但是在使用上也都会尽量做的简单易懂。说实话我觉得有些东西我们看了很难懂，是语言表达的晦涩而不是工具本身。就比如上面的CyclicBarrier,破壁啊，聚会人数啊种种表述，本质上不还是计数么？这里不得不说很多时候外文翻译过来也就这样了，还是中国人自己做的东西用起来比较顺手。哈哈，继续往下说Semphore:

官方手册中的介绍

这么长的一段话。大概就是说这个Semphore可以初始化一定数量的许可证。一堆线程可以选择获取这个许可证进行一些操作。操作完要还回许可证。如果没获取到许可证的会一直等着。比如我们去饭店吃饭。饭店就只有20个包厢。每波客人去都会用一个包厢。吃完走了就会空出包厢。如果包厢满了再去饭店的客人只能在门口/大厅等着有空出来的包厢才能进去。
而且我们可以看这个构造方法:

Semphore构造方法
这个其实挺好理解的：一个参数是饭店中有的包厢数。还可以设定这个饭店是什么模式？当一堆人等待的时候是按时间顺序先排队的先进去（公平锁）还是说有包厢空出来了一群人打架抢这个包厢，谁赢了谁进去（非公平锁）。
当这个类的概念我们明白了剩下的就是去代码中使用它（我直接贴demo代码再一一解释）：

    public static void main(String[] args) throws Exception {
            Semaphore  semaphore = new Semaphore(3);
            for(int i = 0;i<6;i++) {
                final int temp = i;
                new Thread(()->{
                    try {
                        TimeUnit.SECONDS.sleep(temp);
                        semaphore.acquire();
                        System.out.println(Thread.currentThread().getName()+"号客人进入包厢。");
                        TimeUnit.SECONDS.sleep(5);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }finally {
                        System.out.println(Thread.currentThread().getName()+"号客人离开包厢。");
                        semaphore.release();
                    }

                },"t"+i).start();
            }
    }

这里各种睡是为了让我们可以更好的看时间顺序：

执行结果

首先进来就睡i秒是为了让客人陆陆续续的来。其次吃饭5秒中省的进来和出去看不清。一开始我没各种限制条件会直接先显示进来四个再出去一个。这个不是说饭店三个包厢能同时吃四波人，而是可以理解上一桌客人吃完了出了包厢还没走出大门新的一波人就进来了而已。所以我现在这么睡了以后就比较容易看出来运行的结果了。
和预料的是一样的。当满了以后所有人都只能等着。有人空出来了就会有人补进去。最终到没有人排队。
Semphore需要记住的就几点：

两个重要方法等待获取锁和释放锁。
- 获取锁acquire();
- 释放锁release();
它的作用可以是限流。比如高并发来了，我们可以一个个处理。甚至秒杀的话可以直接设置给定商品数量的信号量。实用还是比较多的。

这篇笔记也暂时就到这里，如果稍微帮到你了记得点个喜欢点个关注。另外文中大部分都是我自己看手册和源码自己组织语言写下来的（我觉得单纯把老师讲的一字一字打出来没什么意义），所以如果有说的不准确或者有问题欢迎指出！另外也希望大家都可以工作顺顺利利！周末愉快！

Java Util Concurrent并发编程（二）集合类的

集合类不安全

ArrayList

HashSet

HashMap

Callable

JUC中的辅助类

CountDownLatch

CyclicBarrier

Semaphore

猜你喜欢

热点阅读

Java Util Concurrent并发编程（二） 集合类的

集合类不安全

ArrayList

HashSet

HashMap

Callable

JUC中的辅助类

CountDownLatch

CyclicBarrier

Semaphore

猜你喜欢

热点阅读

Java Util Concurrent并发编程（二）集合类的