Java并发编程

字数统计: 5.8k字   |   阅读时长≈ 22分

0x01 volatile 变量

volatile关键字的两层语义

　　一旦一个共享变量（类的成员变量、类的静态成员变量）被volatile修饰之后，那么就具备了两层语义：

　　1）保证了不同线程对这个变量进行操作时的可见性，即一个线程修改了某个变量的值，这新值对其他线程来说是立即可见的。

　　2）禁止进行指令重排序。

⚠️注意：volatile只能保证变量的可见性，但不保证修饰变量的原子性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public class Test {
    public volatile int inc = 0;
     
    public void increase() {
        inc++;
    }
     
    public static void main(String[] args) {
        final Test test = new Test();
        for(int i=0;i<10;i++){
            new Thread(){
                public void run() {
                    for(int j=0;j<1000;j++)
                        test.increase();
                };
            }.start();
        }
         
        while(Thread.activeCount()>1)  //保证前面的线程都执行完
            Thread.yield();
        System.out.println(test.inc);
    }
}

//上面👆这段代码最后print出的结果会是一个小于10000的任意一个数

Java中的双重检查（Double-Check）

Java在执行双重检查时会遇到指令重排序的问题

1
2
3
4
5
6
7
8
9
10
11
12
13
public class DoubleCheckedLocking {                 //1
    private static Instance instance;                    //2

    public static Instance getInstance() {               //3
        if (instance == null) {                          //4:第一次检查
            synchronized (DoubleCheckedLocking.class) {  //5:加锁
                if (instance == null)                    //6:第二次检查
                    instance = new Instance();           //7:问题的根源出在这里
            }                                            //8
        }                                                //9
        return instance;                                 //10
    }                                                    //11
}     																										//12

前面的双重检查锁定示例代码的第7行（instance = new Singleton();）创建一个对象。这一行代码可以分解为如下的三行伪代码：

1
2
3
memory = allocate();   //1：分配对象的内存空间
ctorInstance(memory);  //2：初始化对象
instance = memory;     //3：设置instance指向刚分配的内存地址

上面三行伪代码中的2和3之间，可能会被重排序（在一些JIT编译器上，这种重排序是真实发生的，详情见参考文献1的“Out-of-order writes”部分）。2和3之间重排序之后的执行时序如下：

1
2
3
4
memory = allocate();   //1：分配对象的内存空间
instance = memory;     //3：设置instance指向刚分配的内存地址
                       //注意，此时对象还没有被初始化！
ctorInstance(memory);  //2：初始化对象

所以解决的思路就是在Instance变量前添加volatile关键字，禁止new instance的指令重排列

双重检查锁定与延迟初始化

0x03 Java线程的6个状态

1
2
3
4
5
6
7
8
public enum State{
  NEW,  // 表示刚刚创建的线程，还没开始执行
  RUNNABLE, // 线程执行时处于RUNNABLE状态，表示线程所需的一切资源都已经准备好了
  BLOCKED, // 阻塞状态，线程会暂时停止执行，直到获得请求的锁
  WAITING, // 进入一个无时间限制的等待状态
  TIMED_WAITING, // 进入一个有时限的等待状态
  TERMINATED; // 线程的终止状态
}

0x04 Java 线程的run()和start()的区别

start() :

它的作用是启动一个新线程。
通过start()方法来启动的新线程，处于就绪（可运行）状态，并没有运行，一旦得到cpu时间片，就开始执行相应线程的run()方法，这里方法run()称为线程体，它包含了要执行的这个线程的内容，run方法运行结束，此线程随即终止。start()不能被重复调用。用start方法来启动线程，真正实现了多线程运行，即无需等待某个线程的run方法体代码执行完毕就直接继续执行下面的代码。这里无需等待run方法执行完毕，即可继续执行下面的代码，即进行了线程切换。

run() :

run()就和普通的成员方法一样，可以被重复调用。
如果直接调用run方法，并不会启动新线程！程序中依然只有主线程这一个线程，其程序执行路径还是只有一条，还是要顺序执行，还是要等待run方法体执行完毕后才可继续执行下面的代码，这样就没有达到多线程的目的。
总结：调用start方法方可启动线程，而run方法只是thread的一个普通方法调用，还是在主线程里执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public class Test {
    static void pong(){
        System.out.print("pong");
    }
    public static void main(String[] args) {
        Thread t=new Thread(){
            public void run(){
                pong();
            }
        };
        t.run();
        System.out.print("ping");       
    }
}
运行结果：
pongping

总结一下：

  1. start() 可以启动一个新线程，run()不能
  2. start()不能被重复调用，run()可以
  3. start()中的run代码可以不执行完就继续执行下面的代码，即进行了线程切换。直接调用run方法必须等待其代码全部执行完才能继续执行下面的代码。
  4. start() 实现了多线程，run()没有实现多线程。

0x05 Java 中的wait、notify、notifyAll

这三个方法是Object类的三个本地方法

wait：线程自动释放其占有的对象锁，并等待notify
notify：唤醒一个正在wait当前对象锁的线程，并让它拿到对象锁
notifyAll：唤醒所有正在wait前对象锁的线程

notify和notifyAll的最主要的区别是：notify只是唤醒一个正在wait当前对象锁的线程，而notifyAll唤醒所有。值得注意的是：notify是本地方法，具体唤醒哪一个线程由虚拟机控制；notifyAll后并不是所有的线程都能马上往下执行，它们只是跳出了wait状态，接下来它们还会是竞争对象锁。

详细demo参考：Java 中的wait、notify、notifyAll

0x06 Java 中的join和yield

join()方法表示无限等待，它会一直阻塞当前线程，直到目标线程执行完毕

yield()方法表示当前线程让出CPU，重新进行CPU资源的竞争

详细demo参考：Java多线程系列–“基础篇”08之 join()

0x07 一个错误的加锁示范

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
public class BadLockOnInteger implements Runnable {

    public static Integer i = 0;
    static BadLockOnInteger instance = new BadLockOnInteger();

    @Override
    public void run() {
        for (int j = 0; j < 10000000; j++) {
            synchronized (i) {
                i++;
            }
        }
    }


    public static void main(String[] args) throws InterruptedException {
        Thread t1 = new Thread(instance);
        Thread t2 = new Thread(instance);

        t1.start();
        t2.start();
        t1.join();
        t2.join();
        System.out.println(i);
    }
}

上面这段代码会返回一个比20000000小很多的数字，原因在于没有做到真正的线程安全。因为Integer属于不可变对象，一旦被创建，就不可能被修改。所以Integer的值被++只能是新建一个Integer，并让它表示为2。

所以上述代码线程每次加锁都加在了不同对象实例上，没有达到上锁的目的。

修改方法是是将synchronized(i) 改为synchronized(isntance)

0x08 ThreadPoolExector

参考：Java中的线程池——ThreadPoolExecutor的使用

corePoolSize:核心线程数，默认情况下核心线程会一直存活，即使处于闲置状态也不会受存keepAliveTime限制。除非将allowCoreThreadTimeOut设置为true。

maximumPoolSize:线程池所能容纳的最大线程数。超过这个数的线程将被阻塞。当任务队列为没有设置大小的LinkedBlockingDeque时，这个值无效。

keepAliveTime:非核心线程的闲置超时时间，超过这个时间就会被回收。

unit:指定keepAliveTime的单位，如TimeUnit.SECONDS。当将allowCoreThreadTimeOut设置为true时对corePoolSize生效。

workQueue:线程池中的任务队列.

常用的有三种队列，SynchronousQueue,LinkedBlockingDeque,ArrayBlockingQueue。

threadFactory:线程工厂，提供创建新线程的功能。ThreadFactory是一个接口，只有一个方法

原理

  1. 如果当前池大小 poolSize 小于 corePoolSize ，则创建新线程执行任务。
  2. 如果当前池大小 poolSize 大于 corePoolSize ，且等待队列未满，则进入等待队列
  3. 如果当前池大小 poolSize 大于 corePoolSize 且小于 maximumPoolSize ，且等待队列已满，则创建新线程执行任务。
  4. 如果当前池大小 poolSize 大于 corePoolSize 且大于 maximumPoolSize ，且等待队列已满，则调用拒绝策略来处理该任务。
  5. 线程池里的每个线程执行完任务后不会立刻退出，而是会去检查下等待队列里是否还有线程任务需要执行，如果在 keepAliveTime 里等不到新的任务了，那么线程就会退出。

0X09 CAS 技术

CAS是乐观锁技术，当多个线程尝试使用CAS同时更新同一个变量时，只有其中一个线程能更新变量的值，而其它线程都失败，失败的线程并不会被挂起，而是被告知这次竞争中失败，并可以再次尝试。CAS有3个操作数，内存值V，旧的预期值A，要修改的新值B。当且仅当预期值A和内存值V相同时，将内存值V修改为B，否则什么都不做。

0x10 Java中用到的线程调度算法

采用时间片轮转的方式。可以设置线程的优先级，会映射到下层的系统上面的优先级上，如非特别需要，尽量不要用，防止线程饥饿。

0x10 什么是线程组，为什么在Java中不推荐使用？

ThreadGroup类，可以把线程归属到某一个线程组中，线程组中可以有线程对象，也可以有线程组，组中还可以有线程，这样的组织结构有点类似于树的形式。

不推荐使用的原因：

  • 1.线程组ThreadGroup对象中比较有用的方法是stop、resume、suspend等方法，由于这几个方法会导致线程的安全问题（主要是死锁问题），已经被官方废弃掉了，所以线程组本身的应用价值就大打折扣了。
  • 2.线程组ThreadGroup不是线程安全的，这在使用过程中获取的信息并不全是及时有效的，这就降低了它的统计使用价值。

0x11 为什么使用Exector框架（可以理解为为什么要使用线程池）

Java并发——Executor框架详解（Executor框架结构与框架成员)

Exector框架是Java的线程池框架

  • 使用new Thread()来执行任务时需要不断的创建新进程是一个比较耗时和耗资源的。
  • 调用 new Thread()创建的线程缺乏管理，被称为野线程，而且可以无限制的创建，线程之间的相互竞争会导致过多占用系统资源而导致系统瘫痪，还有线程之间的频繁交替也会消耗很多系统资源。
  • 接使用new Thread() 启动的线程不利于扩展，比如定时执行、定期执行、定时定期执行、线程中断等都不便实现。

0x12 Java线程池—Executor框架源码深度解析

Java线程池—Executor框架源码深度解析

0x13什么是阻塞队列？阻塞队列的实现原理是什么？如何使用阻塞队列来实现生产者-消费者模型？

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。

这两个附加的操作是：在队列为空时，获取元素的线程会等待队列变为非空。当队列满时，存储元素的线程会等待队列可用。

阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。

0x14 在Java中CycliBarriar和CountdownLatch有什么区别？

CountdownLatch:

一个线程(或者多个)，等待另外N个线程完成某个事情之后才能执行。是并发包中提供的一个可用于控制多个线程同时开始某个动作的类，其采用的方法为减少计数的方式，当计数减至零时位于latch.Await()后的代码才会被执行，CountDownLatch是减计数方式，计数==0时释放所有等待的线程；CountDownLatch当计数到0时，计数无法被重置；

CyclicBarrier:

字面意思回环栅栏，通过它可以实现让一组线程等待至某个状态之后再全部同时执行。叫做回环是因为当所有等待线程都被释放以后，CyclicBarrier可以被重用。 即:N个线程相互等待，任何一个线程完成之前，所有的线程都必须等待。CyclicBarrier是当await的数量到达了设置的数量的时候，才会继续往下面执行，CyclicBarrier计数达到指定值时，计数置为0重新开始。

Java中CycliBarriar和CountdownLatch区别(附测试实例)

0x15 乐观锁和悲观锁的理解及如何实现，有哪些实现方式？

悲观锁：总是假设最坏的情况，每次去拿数据的时候都认为别人会修改，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会阻塞直到它拿到锁。传统的关系型数据库里边就用到了很多这种锁机制，比如行锁，表锁等，读锁，写锁等，都是在做操作之前先上锁。再比如Java里面的同步原语synchronized关键字的实现也是悲观锁。

乐观锁：顾名思义，就是很乐观，每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号等机制。乐观锁适用于多读的应用类型，这样可以提高吞吐量，像数据库提供的类似于write_condition机制，其实都是提供的乐观锁。在Java中java.util.concurrent.atomic包下面的原子变量类就是使用了乐观锁的一种实现方式CAS实现的。

乐观锁的实现方式：

  • 使用版本标识来确定读到的数据与提交时的数据是否一致。提交后修改版本标识，不一致时可以采取丢弃和再次尝试的策略。
  • java中的Compare and Swap即CAS ，当多个线程尝试使用CAS同时更新同一个变量时，只有其中一个线程能更新变量的值，而其它线程都失败，失败的线程并不会被挂起，而是被告知这次竞争中失败，并可以再次尝试。　CAS 操作中包含三个操作数 —— 需要读写的内存位置（V）、进行比较的预期原值（A）和拟写入的新值(B)。如果内存位置V的值与预期原值A相匹配，那么处理器会自动将该位置值更新为新值B。否则处理器不做任何操作。

CAS缺点：

  • ABA问题：

    比如说一个线程one从内存位置V中取出A，这时候另一个线程two也从内存中取出A，并且two进行了一些操作变成了B，然后two又将V位置的数据变成A，这时候线程one进行CAS操作发现内存中仍然是A，然后one操作成功。尽管线程one的CAS操作成功，但可能存在潜藏的问题。从Java1.5开始JDK的atomic包里提供了一个类AtomicStampedReference来解决ABA问题。

  • 循环时间长开销大：

    对于资源竞争严重（线程冲突严重）的情况，CAS自旋的概率会比较大，从而浪费更多的CPU资源，效率低于synchronized。

  • 只能保证一个共享变量的原子操作：

    当对一个共享变量执行操作时，我们可以使用循环CAS的方式来保证原子操作，但是对多个共享变量操作时，循环CAS就无法保证操作的原子性，这个时候就可以用锁

0x16 ConcurrentHashMap 是如何实现的？ 1.7、1.8 实现有何不同？为什么这么做？

参考：

1.Java的ConcurrentHashMap

2.ConcurrentHashMap揭秘-JDK1.7

1 ConcurrentHashMap1.8 同步原理

concurrentHashMap 在JDK1.8中采用了 CAS + synchronized 来保证并发安全性。至于如何实现，那我继续看一下put方法逻辑

  • putVal()方法逻辑
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    //1. 计算key的hash值
    int hash = spread(key.hashCode());
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        //2. 如果当前table还没有初始化先调用initTable方法将tab进行初始化
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        //3. tab中索引为i的位置的元素为null，则直接使用CAS将值插入即可
        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
        }
        //4. 当前正在扩容
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    //5. 当前为链表，在链表中插入新的键值对
                    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;
                            }
                        }
                    }
                    // 6.当前为红黑树，将新的键值对插入到红黑树中
                    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;
                        }
                    }
                }
            }
            // 7.插入完键值对后再根据实际大小看是否需要转换成红黑树
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    //8.对当前容量大小进行检查，如果超过了临界值（实际大小*加载因子）就需要扩容 
    addCount(1L, binCount);
    return null;
}

这个put的过程很清晰，对当前的table进行无条件自循环直到put成功，可以分成以下六步流程来概述：

1、判断Node[]数组是否初始化，没有则进行初始化操作
2、通过hash定位数组的索引坐标，是否有Node节点，如果没有则使用CAS进行添加（链表的头节点），添加失败则进入下次循环。
3、检查到内部正在扩容，就帮助它一块扩容。
4、如果f！=null，则使用synchronized锁住f元素（链表/红黑树的头元素）。如果是Node（链表结构）则执行链表的添加操作；如果是TreeNode（树型结构）则执行树添加操作。
5、判断链表长度已经达到临界值8（默认值），当节点超过这个值就需要把链表转换为树结构。
6、如果添加成功就调用addCount（）方法统计size，并且检查是否需要扩容

2 ConcurrentHashMap 1.7 同步原理

HashMap是根据散列值分段存储的，同步Map在同步的时候锁住了所有的段，而ConcurrentHashMap加锁的时候根据散列值锁住了散列值锁对应的那段，因此提高了并发性能。ConcurrentHashMap也增加了对常用复合操作的支持，比如”若没有则添加”：putIfAbsent()，替换：replace()。这2个操作都是原子操作。

ConcurrentHashMap可以做到读取数据不加锁，并且其内部的结构可以让其在进行写操作的时候能够将锁的粒度保持地尽量地小，不用对整个ConcurrentHashMap加锁。

image-20191226103855951

可以看到，ConcurrentHashMap内部采用了一种叫Segment的数据结构，很明显它就是一个数组table(哈希桶)，数据的元素就是大家熟悉的HashEntry(哈希链)。

简单来讲，就是ConcurrentHashMap比HashMap多了一次hash过程，第1次hash定位到Segment，第2次hash定位到HashEntry，然后链表搜索找到指定节点。

该种实现方式的缺点是hash过程比普通的HashMap要长，但是优点也很明显，在进行写操作时，只需锁住写元素所在的Segment即可，其他Segment无需加锁，提高了并发读写的效率。

  • ConcurrentHashMap1.7 中的put()方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
V put(K key, int hash, V value, boolean onlyIfAbsent) {
    // 加锁
    lock();
    try {
        int c = count;
        // 对c进行+1操作，获取新的数组容量
        // 如果新的数组容量高于阈值，则先进行扩容操作
        if (c++ > threshold) // ensure capacity
            rehash();
        // 获取Segment的数组table
        HashEntry<K,V>[] tab = table;
        // 确定在数组中的位置index，即第2次hash过程
        int index = hash & (tab.length - 1);
        // 获取index位置的头结点
        HashEntry<K,V> first = tab[index];
        HashEntry<K,V> e = first;
        // 沿链表遍历，直到找到与元素key或者hash值相同的节点
        while (e != null && (e.hash != hash || !key.equals(e.key)))
            e = e.next;
  
        V oldValue;
        // 若key或者hash值相同的节点存在，则进行更新操作
        if (e != null) {
            // value也是volatile修饰的，所以内存即时可见
            oldValue = e.value;
            if (!onlyIfAbsent)
                e.value = value;
        }
        // 若key或者hash值相同的节点不存在，则新建节点，追加到当前链表的头部(头插法)
        else {
            oldValue = null;
            // 更新modCount的值，记录对table的大小造成影响的操作的数量
            ++modCount;
            tab[index] = new HashEntry<K,V>(key, hash, first, value);
            // 更新count的值，内存即时可见
            count = c; // write-volatile
        }
        // 返回旧值
        return oldValue;
    } finally {
        // 释放锁，与synchronize关键字不同，Lock必须显示释放
        unlock();
    }
}

0x17 CopyOnWriteArrayList可以用于什么应用场景？

CopyOnWriteArrayList(免锁容器)的好处之一是当多个迭代器同时遍历和修改这个列表时，不会抛出ConcurrentModificationException。在CopyOnWriteArrayList中，写入将导致创建整个底层数组的副本，而源数组将保留在原地，使得复制的数组在被修改时，读取操作可以安全地执行。

  1. 由于写操作的时候，需要拷贝数组，会消耗内存，如果原数组的内容比较多的情况下，可能导致young gc或者full gc；
  2. 不能用于实时读的场景，像拷贝数组、新增元素都需要时间，所以调用一个set操作后，读取到数据可能还是旧的,虽然CopyOnWriteArrayList 能做到最终一致性,但是还是没法满足实时性要求；

CopyOnWriteArrayList透露的思想

  • 读写分离，读和写分开
  • 最终一致性
  • 使用另外开辟空间的思路，来解决并发冲突

0x18 为什么代码会重排序?

在执行程序时，为了提供性能，处理器和编译器常常会对指令进行重排序，但是不能随意重排序，不是你想怎么排序就怎么排序，它需要满足以下两个条件：

  • 在单线程环境下不能改变程序运行的结果；
  • 存在数据依赖关系的不允许重排序

需要注意的是：重排序不会影响单线程环境的执行结果，但是会破坏多线程的执行语义。

0x19 volatile 变量和 atomic 变量有什么不同？

Volatile变量可以确保先行关系，即写操作会发生在后续的读操作之前, 但它并不能保证原子性。例如用volatile修饰count变量那么 count++ 操作就不是原子性的。

而AtomicInteger类提供的atomic方法可以让这种操作具有原子性如getAndIncrement()方法会原子性的进行增量操作把当前值加一，其它数据类型和引用变量也可以进行相似操作。

赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

Sorry，you know nothing about me :)