多线程并发问题

作者: Android_冯星 | 来源:发表于2020-04-10 13:43 被阅读0次

多线程并发问题，基本是面试必问的。

大部分同学应该都知道Synchronized，Lock，部分同学能说到volatile、并发包，优秀的同学则能在前面的基础上，说出Synchronized、volatile的原理，以及并发包中常用的数据结构，例如ConcurrentHashMap的原理。

一、多线程为什么会有并发问题

为什么多线程同时访问（读写）同个变量，会有并发问题？

Java 内存模型规定了所有的变量都存储在主内存中，每条线程有自己的工作内存。
线程的工作内存中保存了该线程中用到的变量的主内存副本拷贝，线程对变量的所有操作都必须在工作内存中进行，而不能直接读写主内存。
线程访问一个变量，首先将变量从主内存拷贝到工作内存，对变量的写操作，不会马上同步到主内存。
不同的线程之间也无法直接访问对方工作内存中的变量，线程间变量的传递均需要自己的工作内存和主存之间进行数据同步进行。

二、Java 内存模型（JMM）

Java 内存模型(JMM) 作用于工作内存（本地内存）和主存之间数据同步过程，它规定了如何做数据同步以及什么时候做数据同步，如下图。

image.png

三、并发三要素

原子性：在一个操作中，CPU 不可以在中途暂停然后再调度，即不被中断操作，要么执行完成，要么就不执行。

可见性：多个线程访问同一个变量时，一个线程修改了这个变量的值，其他线程能够立即看得到修改的值。

有序性：程序执行的顺序按照代码的先后顺序执行。

四、怎么做，才能解决并发问题？（重点）

下面结合不同场景分析解决并发问题的处理方式。

4.1 volatile 关键字

4.1.1 volatile 特性

保证可见性，不保证原子性

当写一个volatile变量时，JVM会把本地内存的变量强制刷新到主内存中
这个写操作导致其他线程中的缓存无效，其他线程读，会从主内存读。volatile的写操作对其它线程实时可见。

禁止指令重排序 指令重排序是指编译器和处理器为了优化程序性能对指令进行排序的一种手段，需要遵守一定规则：

不会对存在依赖关系的指令重排序，例如 a = 1;b = a; a 和b存在依赖关系，不会被重排序
不能影响单线程下的执行结果。比如：a=1;b=2;c=a+b这三个操作,前两个操作可以重排序，但是c=a+b不会被重排序，因为要保证结果是3

4.1.2 使用场景

对于一个变量，只有一个线程执行写操作，其它线程都是读操作，这时候可以用 volatile 修饰这个变量。

4.1.3 单例双重锁为什么要用到volatile？

public class TestInstance {

private static volatile TestInstance mInstance;

public static TestInstance getInstance(){       //1
    if (mInstance == null){                     //2
        synchronized (TestInstance.class){      //3
            if (mInstance == null){             //4
                mInstance = new TestInstance(); //5
            }
        }
    }
    return mInstance;
}

假如没有用volatile，并发情况下会出现问题，线程A执行到注释5 new TestInstance() 的时候，分为如下几个几步操作：

分配内存
初始化对象
mInstance 指向内存

这时候如果发生指令重排，执行顺序是132，执行到第3的时候，线程B刚好进来了，并且执行到注释2，这时候判断mInstance 不为空，直接使用一个未初始化的对象。所以使用volatile关键字来禁止指令重排序。

4.1.4 volatile 原理

在JVM底层volatile是采用内存屏障来实现的，内存屏障会提供3个功能：

它确保指令重排序时不会把其后面的指令排到内存屏障之前的位置，也不会把前面的指令排到内存屏障的后面；即在执行到内存屏障这句指令时，在它前面的操作已经全部完成；
它会强制将缓存的修改操作立即写到主内存
写操作会导致其它CPU中的缓存行失效，写之后，其它线程的读操作会从主内存读。

4.1.5 volatile 的局限性

volatile 只能保证可见性，不能保证原子性写操作对其它线程可见，但是不能解决多个线程同时写的问题。

二、Synchronized

4.2.1 Synchronized 使用场景

多个线程同时写一个变量。

例如售票，余票是100张，窗口A和窗口B同时各卖出一张票，假如余票变量用 volatile 修饰，是有问题的。

A窗口获取余票是100，B窗口获取余票也是100，A卖出一张变成99，刷新回主内存，同时B卖出一张变成99，也刷新回主内存，会导致最终主内存余票是99而不是98。

前面说到 volatile 的局限性，就是多个线程同时写的情况，这种情况一般可以使用Synchronized。

Synchronized 可以保证同一时刻，只有一个线程可执行某个方法或某个代码块。

4.2.2 Synchronized 原理


public class SynchronizedTest {

public static void main(String[] args) {
    synchronized (SynchronizedTest.class) {
        System.out.println("123");
    }
    method();
}

private static void method() {
}
}

将这段代码先用javac命令编译，再java p -v SynchronizedTest.class命令查看字节码，部分字节码如下

public static void main(java.lang.String[]);
descriptor: ([Ljava/lang/String;)V
flags: ACC_PUBLIC, ACC_STATIC
Code:
  stack=2, locals=3, args_size=1
     0: ldc           #2                  // class com/lanshifu/opengldemo/test/SynchronizedTest
     2: dup
     3: astore_1
     4: monitorenter
     5: getstatic     #3                  // Field java/lang/System.out:Ljava/io/PrintStream;
     8: ldc           #4                  // String 123
    10: invokevirtual #5                  // Method java/io/PrintStream.println:(Ljava/lang/String;)V
    13: aload_1
    14: monitorexit
    15: goto          23
    18: astore_2
    19: aload_1
    20: monitorexit
    21: aload_2
    22: athrow
    23: invokestatic  #6                  // Method method:()V
    26: return

可以看到 4: monitorenter 和 14: monitorexit，中间是打印的语句。

执行同步代码块，首先会执行monitorenter指令，然后执行同步代码块中的代码，退出同步代码块的时候会执行monitorexit指令。

使用Synchronized进行同步，其关键就是必须要对对象的监视器monitor进行获取，当线程获取monitor后才能继续往下执行，否则就进入同步队列，线程状态变成BLOCK，同一时刻只有一个线程能够获取到monitor，当监听到monitorexit被调用，队列里就有一个线程出队，获取monitor。详情参考：www.jianshu.com/p/d53bf830f…

每个对象拥有一个计数器，当线程获取该对象锁后，计数器就会加一，释放锁后就会将计数器减一，所以只要这个锁的计数器大于0，其它线程访问就只能等待。

4.2.3 Synchronized 锁的升级

大家对Synchronized的理解可能就是重量级锁，但是Java1.6对 Synchronized 进行了各种优化之后，有些情况下它就并不那么重，Java1.6 中为了减少获得锁和释放锁带来的性能消耗而引入的偏向锁和轻量级锁。

偏向锁：大多数情况下，锁不仅不存在多线程竞争，而且总是由同一线程多次获得，为了让线程获得锁的代价更低而引入了偏向锁。

当一个线程A访问加了同步锁的代码块时，会在对象头中存储当前线程的id，后续这个线程进入和退出这段加了同步锁的代码块时，不需要再次加锁和释放锁。

轻量级锁：在偏向锁情况下，如果线程B也访问了同步代码块，比较对象头的线程id不一样，会升级为轻量级锁，并且通过自旋的方式来获取轻量级锁。

重量级锁：如果线程A和线程B同时访问同步代码块，则轻量级锁会升级为重量级锁，线程A获取到重量级锁的情况下，线程B只能入队等待，进入BLOCK状态。

4.2.4 Synchronized 缺点

不能设置锁超时时间
不能通过代码释放锁
容易造成死锁

三、ReentrantLock

上面说到Synchronized的缺点，不能设置锁超时时间和不能通过代码释放锁，ReentranLock就可以解决这个问题。

在多个条件变量和高度竞争锁的地方，用ReentrantLock更合适，ReentrantLock还提供了Condition，对线程的等待和唤醒等操作更加灵活，一个ReentrantLock可以有多个Condition实例，所以更有扩展性。

4.3.1 ReentrantLock 的使用

lock 和 unlock

 ReentrantLock reentrantLock = new ReentrantLock();
        System.out.println("reentrantLock->lock");
        reentrantLock.lock();
        try {
            
            System.out.println("睡眠2秒...");
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            reentrantLock.unlock();
            System.out.println("reentrantLock->unlock");
        }

实现可定时的锁请求：tryLock

    public static void main(String[] args) {
        ReentrantLock reentrantLock = new ReentrantLock();
        Thread thread1 = new Thread_tryLock(reentrantLock);
        thread1.setName("thread1");
        thread1.start();
        Thread thread2 = new Thread_tryLock(reentrantLock);
        thread2.setName("thread2");
        thread2.start();
}


    static class Thread_tryLock extends Thread {
        ReentrantLock reentrantLock;

        public Thread_tryLock(ReentrantLock reentrantLock) {
            this.reentrantLock = reentrantLock;
        }

        @Override
        public void run() {
            try {
                System.out.println("try lock:" + Thread.currentThread().getName());
                boolean tryLock = reentrantLock.tryLock(3, TimeUnit.SECONDS);
                if (tryLock) {
                    System.out.println("try lock success :" + Thread.currentThread().getName());
                    System.out.println("睡眠一下：" + Thread.currentThread().getName());
                    Thread.sleep(5000);
                    System.out.println("醒了：" + Thread.currentThread().getName());
                } else {
                    System.out.println("try lock 超时 :" + Thread.currentThread().getName());
                }

            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.println("unlock:" + Thread.currentThread().getName());
                reentrantLock.unlock();
            }
        }
    }

打印的日志：

try lock:thread1
try lock:thread2
try lock success :thread2
睡眠一下：thread2
try lock 超时 :thread1
unlock:thread1
Exception in thread "thread1" java.lang.IllegalMonitorStateException
    at java.util.concurrent.locks.ReentrantLock$Sync.tryRelease(ReentrantLock.java:151)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.release(AbstractQueuedSynchronizer.java:1261)
    at java.util.concurrent.locks.ReentrantLock.unlock(ReentrantLock.java:457)
    at com.lanshifu.demo_module.test.lock.ReentranLockTest$Thread_tryLock.run(ReentranLockTest.java:60)
醒了：thread2
unlock:thread2

上面演示了trtLock的使用，trtLock设置获取锁的等待时间，超过3秒直接返回失败，可以从日志中看到结果。有异常是因为thread1获取锁失败，不应该调用unlock。

4.3.2 Condition 条件

public static void main(String[] args) {

        Thread_Condition thread_condition = new Thread_Condition();
        thread_condition.setName("测试Condition的线程");
        thread_condition.start();
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        thread_condition.singal();

    }


static class Thread_Condition extends Thread {

        @Override
        public void run() {
            await();
        }

        private ReentrantLock lock = new ReentrantLock();
        public Condition condition = lock.newCondition();

        public void await() {
            try {
                System.out.println("lock");
                lock.lock();
                System.out.println(Thread.currentThread().getName() + ":我在等待通知的到来...");
                condition.await();//await 和 signal 对应
                //condition.await(2, TimeUnit.SECONDS); //设置等待超时时间
                System.out.println(Thread.currentThread().getName() + ":等到通知了，我继续执行>>>");
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                System.out.println("unlock");
                lock.unlock();
            }
        }

        public void singal() {
            try {
                System.out.println("lock");
                lock.lock();
                System.out.println("我要通知在等待的线程，condition.signal()");
                condition.signal();//await 和 signal 对应
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.println("unlock");
                lock.unlock();
            }
        }
    }

运行打印日志

lock
测试Condition的线程:我在等待通知的到来...
lock
我要通知在等待的线程，condition.signal()
unlock
测试Condition的线程:等到通知了，我继续执行>>>
unlock

上面演示了Condition的 await 和 signal 使用，前提要先lock。

4.3.3 公平锁与非公平锁

ReentrantLock 构造函数传true表示公平锁。

公平锁表示线程获取锁的顺序是按照线程加锁的顺序来分配的，即先来先得的顺序。而非公平锁就是一种锁的抢占机制，是随机获得锁的，可能会导致某些线程一致拿不到锁，所以是不公平的。

4.3.4 ReentrantLock 注意点

ReentrantLock使用lock和unlock来获得锁和释放锁
unlock要放在finally中，这样正常运行或者异常都会释放锁
使用condition的await和signal方法之前，必须调用lock方法获得对象监视器

四、并发包

通过上面分析，并发严重的情况下，使用锁显然效率低下，因为同一时刻只能有一个线程可以获得锁，其它线程只能乖乖等待。

Java提供了并发包解决这个问题，接下来介绍并发包里一些常用的数据结构。

我们都知道HashMap是线程不安全的数据结构，HashTable则在HashMap基础上，get方法和put方法加上Synchronized修饰变成线程安全，不过在高并发情况下效率底下，最终被ConcurrentHashMap替代。

ConcurrentHashMap 采用分段锁，内部默认有16个桶，get和put操作，首先将key计算hashcode，然后跟16取余，落到16个桶中的一个，然后每个桶中都加了锁（ReentrantLock），桶中是HashMap结构（数组加链表，链表过长转红黑树）。

所以理论上最多支持16个线程同时访问。

4.4.2 LinkBlockingQueue

链表结构的阻塞队列，内部使用多个ReentrantLock

    /** Lock held by take, poll, etc */
    private final ReentrantLock takeLock = new ReentrantLock();

    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();

    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();

    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();

private void signalNotEmpty() {
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
    }

    /**
     * Signals a waiting put. Called only from take/poll.
     */
    private void signalNotFull() {
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            notFull.signal();
        } finally {
            putLock.unlock();
        }
    }

源码不贴太多，简单说一下LinkBlockingQueue 的逻辑：

从队列获取数据，如果队列中没有数据，会调用notEmpty.await();进入等待。
在放数据进去队列的时候会调用notEmpty.signal();，通知消费者，1中的等待结束，唤醒继续执行。
从队列里取到数据的时候会调用notFull.signal();，通知生产者继续生产。
在put数据进入队列的时候，如果判断队列中的数据达到最大值，那么会调用notFull.await();，等待消费者消费掉，也就是等待3去取数据并且发出notFull.signal();，这时候生产者才能继续生产。

LinkBlockingQueue 是典型的生产者消费者模式，源码细节就不多说。

4.4.3 原子操作类：AtomicInteger

内部采用CAS（compare and swap）保证原子性

举一个int自增的例子

        AtomicInteger atomicInteger = new AtomicInteger(0);
        atomicInteger.incrementAndGet();//自增

五、总结

1. 当只有一个线程写，其它线程都是读的时候，可以用volatile修饰变量
1. 当多个线程写，那么一般情况下并发不严重的话可以用Synchronized，Synchronized并不是一开始就是重量级锁，在并发不严重的时候，比如只有一个线程访问的时候，是偏向锁；当多个线程访问，但不是同时访问，这时候锁升级为轻量级锁；当多个线程同时访问，这时候升级为重量级锁。所以在并发不是很严重的情况下，使用Synchronized是可以的。不过Synchronized有局限性，比如不能设置锁超时，不能通过代码释放锁。
1. ReentranLock 可以通过代码释放锁，可以设置锁超时。
4.高并发下，Synchronized、ReentranLock 效率低，因为同一时刻只有一个线程能进入同步代码块，如果同时有很多线程访问，那么其它线程就都在等待锁。这个时候可以使用并发包下的数据结构，例如ConcurrentHashMap，LinkBlockingQueue，以及原子性的数据结构如：AtomicInteger。

面试的时候按照上面总结的这个思路回答基本就ok了。既然说到并发包，那么除了ConcurrentHashMap，其它一些常用的数据结构的原理也需要去了解下，例如HashMap、HashTable、TreeMap原理，ArrayList、LinkedList对比，这些都是老生常谈的，自己去看源码或者一些博客。