多线程(进阶)

1. 常见的锁策略

1.1 乐观锁 悲观锁

乐观锁 : 总是假设最好的情况,每次去拿数据的时候都认为别人不会修改数据,但是在对数据提交更新的时候,再去判断这个数据在这个期间是否有别人对这个数据进行了修改.

悲观锁 : 总是假设最坏的情况,每次去拿数据的时候都认为别人会修改数据,每次在拿数据的时候都会上锁,当别人想去拿这个数据的时候就会阻塞直到它拿到这个锁…

乐观锁的一个使用方法—版本号机制
引入一个版本号 version,初值为1, 并且规定只有提交的版本大于记录当前版本才能执行

图解版本号机制:

1.2 读写锁

线程对于数据的访问,主要存在两种操作: 读操作 和 写操作

• 当两个线程都是读操作的时候,此时线程没有安全的问题.不必互斥.
• 当两个线程都是写操作的时候,此时线程有安全问题. 必须互斥
• 当一个线程是读操作 一个线程是写操作,此时线程有安全问题,必须互斥

但是在第三种情况的时候,两个线程用同一把锁就会产生性能损耗.

读写锁就能解决这个性能消耗问题.也就是把读操作和写操作区别对待.
Java标准库中提供了ReentrantReadWriteLock 类,实现了读写锁.

• ReentrantReadWriteLock.ReadLock类表示一个读锁. 提供了一个lock/unlock方法进行加锁解锁.
• ReetrantReadWriteLock.WriteLock类表示一个写锁.提供了一个lock/unlock方法进行加锁解锁.

读写锁适合"读多写少"的场景

1.3 重量级锁 轻量级锁

CPU提供了原子操作指令,操作系统对这些指令封装了一层,提供了一个mutex互斥锁,JVM相对于操作系统提供的mutex互斥锁,又封装了一层,实现了synchronized这样的锁
重量级锁: 加锁机制重度依赖了OS提供的mutex,加锁的开销很大.通常是使用内核来完成的.
轻量级锁 加锁机制尽可能不使用mutex,加锁的开销更小.通常是在用户态来完成的.

1.4 自旋锁(Spin Lock) 挂起等待锁

挂起等待锁: 线程在枪锁失败了之后会进入阻塞等待的状态,结束阻塞需要看操作系统具体的调度时间.当线程挂起来的时候,不占用CPU;

自旋锁: 在线程枪锁失败后,不是阻塞等待,而是快速的再循环一次,一旦锁被其他线程释放,就能第一时间获取到锁.

自旋锁的优点 : 没有放弃CPU,不涉及线程阻塞和调度,一旦锁被释放,能第一时间获取到锁.
自旋锁的缺点 : 如果其他的线程一直使用这个锁,长时间没有释放,那么CPU的消耗就很大.

图解自旋锁 与 挂起等待锁

1.5 公平锁 非公平锁

公平锁: 遵循 “先来后到” 原则,如果线程2比线程3先来,当线程1释放锁之后,线程2就能在线程3之前先获取到锁

非公平锁: 不遵循 “先来后到” 原则,如果线程1释放锁之后,线程2和线程3都可能获取到锁.

公平锁和非公平锁 图解:

1.6 可重入锁 不可重入锁

可重入锁: 允许同一个线程多次获取同一把锁.
不可重入锁 不允许同一个线程多次获取同一把锁.

例如:一个递归函数里面有加锁操作,如果在递归的过程中,多次获取了同一把锁,如果这个锁不会阻塞自己,就是可重入锁(也叫递归锁)

Java中 只要以Reettrant开头的锁都是可重入锁.JDK提供的所有现成的Lock实现类,包括synchronized关键字锁都是可重入锁.

2. CAS

2.1 什么是CAS

CAS 英文是 compare and swap 顾名思义就是.“比较并交换”.CAS可以视为一种乐观锁
一个CAS涉及到以下操作:
假设内存中的数据为A,旧的预期数据为B,需要修改的值为C.

1. 比较 A 和 B 是否相等.
2. 如果相等,将C写入A中(交换的步骤)
3. 返回操作是true/false;

这个操作是一个原子性的.CPU提供了一组CAS相关的指令

一个CAS的伪代码,辅助理解CAS的工作流程

boolean CAS(value,oldvalue,expectvalue){
        if(value == oldvalue){
            value = expectvalue;
            return true;
        }
        return false;
    }

当多个线程同时对某一个资源进行CAS操作,只能有一个线程操作成功,其他的线程并不会阻塞,而是收到操作失败的信号.

2.2 CAS 实现了原子类

标准库中提供了 java.util.concurrent.atomic 包,这里面的类都是基于这种方式来实现的.
典型的 AtomicInteger类,其中getAndIncrement相当于 i++操作

AtomicInteger atomicInteger = new AtomicInteger();
	  
	atomicInteger.getAndIncrement();//相当于 i++

一个AtomicInteger的伪代码

class AtomicInteger{
        private int value;
        
        public int getAndIncrement() {
            int oldValue = value;
            while ( CAS(value,oldValue,oldValue+1) != true){
                oldValue = value;
            }
            return oldValue;
        }
    }

图解:
两个线程调用这个getAndIncrement方法

此时不需要使用重量级锁,就能高效的完成多线程的自增操作

2.3 CAS 实现自旋锁

一个自旋锁的伪代码

public class SpinLock {
    private Thread owner = null;
    
    public void lock() {
        while(!CAS(this.owner, null , Thread.currentThread())){
            
        }
    }
    
    public void unlock(){
        this.owner = null;
    }
}

2.4 CAS 的 ABA 问题

2.4.1 什么是 ABA 问题

2.4.2 ABA问题的解决办法

引入版本号,在 CAS 比较数据当前值 和 旧值的同时,也要比较版本号是否符合规则(如果当前版本号 和 读取到的版本号相同,则修改数据并把版本号+1,反之则操作失败).

图解:

3. Synchronized 原理

3.1 Synchronized 的特点

根据锁策略总结了Synchronized的特点,在 JDK 8 的情况下

• 开始的时候是乐观锁,如果锁冲突频繁,就转换成了悲观锁
• 开始的时候是轻量级锁,如果锁被持有的时间较长,就转换成重量级锁
• 实现轻量级锁的时候,大概率用到了自旋锁策略
• 是一种不公平锁
• 是一种可重入锁
• 不是读写锁

3.2 Synchronized 锁加工的过程

JVM 将 Synchronized 锁分为 无锁 -> 偏向锁 -> 自旋锁 -> 重量级锁. 会根据情况依次升级. 也叫 锁膨胀

偏向锁

偏向锁,也是一种乐观锁.认为没有线程会去竞争锁,但是会去给该线程做一个"偏向锁的标记",记录这个锁是属于哪一个线程的.

• 如果当前其他的线程来竞争当前锁,那么就不会真的加锁.(避免了加锁解锁的开销)
• 如果有其他的线程来竞争当前锁,那么会根据标记,让锁优先被属于他的线程拿到,然后其他的线程进入等待,这里也就取消了偏向锁的状态,进入轻量级锁的状态.

自旋锁

当偏向锁状态随着其他的线程的竞争,偏向锁的状态被消除门进入了轻量级锁状态,也就是自适应的自旋锁.

此时就是通过 CAS来实现的.
是在用户态完成的操作,但是自旋操作会浪费CPU.
这里的自旋不会一直进行下去,当达到一定时间或者一定的次数的时候,就不会自旋了.

重量级锁

当此时的冲突概率比较大,锁的竞争更激烈,自旋锁不能快速获取到锁的状态,就会继续膨胀为重量级锁.
这里的重量级锁就用到了内核提供的mutex

3.3 其他的优化操作.

锁消除

通过 编译器 和 JVM 来判断锁是否可以消除.如果可以消除那么就直接消除锁.

例如StringBuffer中,很多方法用到了Synchronized .但是在单线程下,就算有加锁,编译器和JVM都会消除锁.

StringBuffer sb = new StringBuffer();
sb.append("a");
sb.append("c");

这里的append的调用就会频繁的涉及到加锁和解锁,但是在单线程下,这些加锁和解锁的操作都是没必要的.

锁粗化

一段逻辑中如果多次出现加锁解锁,编译器 + JVM 就会自动进行锁的粗化

4. Callable 接口

Callable也是一个创建线程的方法,

一种多线程计算 1~1000的代码

public class ThreadDemo1 {
    static class Result{
        public int sum=0;
        public Object lock = new Object();
    }

    public static void main(String[] args) throws InterruptedException {
        Result result = new Result();
        Thread t1 = new Thread(){
            @Override
            public void run() {
                int sum = 0;
                for (int i = 0; i <= 1000; i++) {
                    sum += i;
                }
                synchronized (result.lock){
                    result.sum = sum;
                    result.lock.notify();
                }
            }
        };
        t1.start();

        synchronized (result.lock){
            if(result.sum == 0){
                result.lock.wait();
            }
            System.out.println(result.sum);
        }
    }
}

这个代码运用到了一个辅助类 Result 和 wait notify 操作.

这里使用Callable 来解决

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class ThreadDemo2 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Callable<Integer> callable = new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                int sum = 0;
                for (int i = 1; i <= 1000 ; i++) {
                    sum+=i;
                }
                return sum;
            }
        };
        FutureTask<Integer> futureTask = new FutureTask<>(callable);
        Thread t = new Thread(futureTask);
        t.start();

        int result = futureTask.get();
        System.out.println(result);
    }
}

这里就简化了很多.
还可以写成匿名类部类的形式

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class ThreadDemo3 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        FutureTask<Integer> futureTask = new FutureTask<>(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                int sum=0;
                for (int i = 0; i <= 1000 ; i++) {
                    sum+=i;
                }
                return sum;
            }
        });
        Thread t = new Thread(futureTask);
        t.start();
        System.out.println(futureTask.get());
    }
}

Callable 相比于 Runnable ,都是描述一个"任务"的,Callable有返回值.Runnable没有.Callable中是call方法 Runnale 是run方法.

5.JUC(java.util.concurrent)的常见类

5.1 ReentrantLock

RenntrantLock 是一个可重入的互斥锁.

RenntrantLock 把加锁和解锁的操作分开了
提供了几个方法:

• lock(): 加锁,如果获取不到锁就死等;
• trylock(超时时间); 加锁,如果获取不到锁,等到一定时间就放弃加锁.
• unlock(): 解锁

ReentrantLock 和 Synchronized 的区别

1. ReentrantLock 把加锁和解锁分成了两个步骤,虽然存在遗忘解锁的可能,但是可以让加锁和解锁的代码更灵活
2. synchronized 在申请锁失败时,会死等.而RenntrantLock 可以通过trylock方法等待一段时间就放弃
3. synchronized 是一个非公平锁,ReentrantLock默认是非公平锁,可以通过自带的构造方法传入一个true来实现公平锁模式

1. ReentrantLock 提供了一个更强大的唤醒机制.synchronized 是通过wait/notify ReentrantLock搭配了一个Condition类实现等待-唤醒,可以更精确的控制唤醒某个指定线程.

5.2 原子类

原子类 内部是使用 CAS 来实现的.所以性能比加锁实现 i++ 高很多.
原子类有:

AtomicBoolean
AtomicInteger
AtomicIntegerArray
AtomicLong
Atomicreference
AtomicStampedReference

一个使用示例:

public static void main(String[] args) {
        AtomicInteger atomicInteger = new AtomicInteger(0);
        atomicInteger.addAndGet(2); // i += delta
        atomicInteger.decrementAndGet(); //--i
        atomicInteger.getAndDecrement(); //i--
        atomicInteger.incrementAndGet(); //++i
        atomicInteger.getAndIncrement(); //i++
    }

5.3 线程池

虽然线程相比于进程,创建和销毁更轻量.但频繁的创建和销毁的时候还是会比较低效.为了解决这个问题就引入了线程池.如果某个线程不使用了,就将该线程放到池子里,需要使用的时候,也不需要创建,而是从池子里拿.

ExecutorService 和 Executors

• ExecutorService 表示一个线程池实例
• Executors 是一个工厂类,能够创建出几种不同风格的线程池
• ExecutorService 的 submit 方法能够向线程池 中提交若干任务.

使用示例:

ExecutorService pool = Executors.newFixedThreadPool(10);
        pool.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("hello");
            }
        });

Executors 创建线程池的几种方式

• newFixedThreadPool: 创建固定线程数的线程池
• newCachedThreadPool: 创建线程数目动态增长的线程池.
• newSingleThreadExecutor: 创建只包含单个线程的线程池.
• newScheduledThreadPool: 设定 延迟时间后执行命令，或者定期执行命令. 是进阶版的 Timer.

Executors 本质上是 ThreadPoolExecutor 类的封装.

ThreadPoolExecutor

ThreadPoolExecutor 的构造方法

理解 ThreadPoolExecutor 构造方法的参数
把创建一个线程池想象成开个公司. 每个员工相当于一个线程.

• corePoolSize: 正式员工的数量. (正式员工, 一旦录用, 永不辞退)
• maximumPoolSize: 正式员工 + 临时工的数目. (临时工: 一段时间不干活, 就被辞退).
• keepAliveTime: 临时工允许的空闲时间.
• unit: keepaliveTime 的时间单位, 是秒, 分钟, 还是其他值. workQueue: 传递任务的阻塞队列
• threadFactory: 创建线程的工厂, 参与具体的创建线程工作.
• RejectedExecutionHandler: 拒绝策略,如果任务量超出公司的负荷了接下来怎么处理.
  AbortPolicy(): 超过负荷, 直接抛出异常.
  CallerRunsPolicy(): 调用者负责处理
  DiscardOldestPolicy(): 丢弃队列中最老的任务.
  DiscardPolicy(): 丢弃新来的任务.

ExecutorService pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS,
                new SynchronousQueue<Runnable>(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
        for(int i=0;i<3;i++) {
            pool.submit(new Runnable() {
                @Override
                public void run() {
                    System.out.println("hello");
                }
            });
        }

5.4 信号量

信号量,是用来表示"可用资源的个数".本质上是一个计数器.
例子: 信号量就相当于是你进停车场的时候,外面会有一个计数牌子,表示当前可用停车位还有多少个.

• 当有一辆车子进去的时候,相当于申请了一个可用资源,可用车位就-1;(相当于信号量的P操作)
• 当有一辆车子出来的时候,相当于释放了一个可用资源,可用车位就+1;(相当于信号量的V操作)
• 如果当前计数牌子为0,还进去了一辆车,那么就会阻塞等待,直到其他释放资源.

代码示例:

public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(4);//这里的4是初始化可用资源数目
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println("申请资源成功");
                    semaphore.acquire();// 申请资源
                    System.out.println("获取到了资源");
                    Thread.sleep(1000);
                    semaphore.release();// 释放资源
                    System.out.println("释放了资源");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        for (int i = 0; i < 20; i++) {
            Thread t = new Thread(runnable);
            t.start();
        }
    }

观察结果:

前四次的获取资源会很快,然后就会等待其他线程释放资源后进行获取.

5.5 CountDownLatch

CountDownLatch 同时等待N个任务执行结束.
相当于比赛的时候,等全部比完,才会公布成绩.

import java.util.concurrent.CountDownLatch;

public class ThreadDemo7 {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(10);
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep((long) Math.random() * 10000);
                    countDownLatch.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        for (int i = 0; i < 10; i++) {
            Thread t = new Thread(runnable);
            t.start();
        }
        countDownLatch.await();
        System.out.println(" 结束 ");
    }
}

6. HashTable 和 ConcurrentHashMap

HashMap

我们知道 HashMap 是线程不安全的,相比于HashTable 和 ConcurrentHashMap比起来,他的key值是允许为空的.

HashTable

HashTable 是线程安全的,把关键方法都加了 synchronized 关键字
一个HashTable 实例,只有一把锁.
当有多个线程并发去修改这个 HashTable 实例的时候,此时这多个线程就会同时竞争一把锁(锁的冲突概率非常高)
当HashTable 扩容的时候,是创建一个更大的内存,然后把数据全部复制过去,这个时候这个插入操作就非常的复杂.

ConcurrentHashMap

ConcurrentHashMap也是线程安全的.但是是相当于针对每一个哈希桶来加锁的.

1. 针对修改操作,使用的是颗粒更小的锁,针对每个哈希桶来分别设定锁,大大降低了锁冲突的概率
2. 针对读操作,没有加锁,而是直接使用了volatile关键字
3. 充分的利用了 CAS 特性.比如获取/修改 size 属性
4. 更优化的扩容
   a) 如果某个操作触发了扩容,就会创建一个新的数组,同时只搬几个元素过去
   b) 这个搬运的过程,新数组和老数组都存在
   c) 每次操作 ConcurrentHashMap线程的时候,都会搬运一部分
   d) 搬完最后一个元素之后就会把老数组删除掉
   e) 这个期间,插入操作只往新数组插入
   f) 这个祺姐,查找操作就需要同时查新数组和老数组

图解

7. 死锁

7.1 什么是死锁

死锁是这样一种情形：多个线程同时被阻塞，它们中的一个或者全部都在等待某个资源被释放。由于线程被无限期地阻塞，因此程序不可能正常终止

情形一: 一个线程针对一把锁,连续加锁两次或两次以上,并且该锁是不可重入锁.

lock void func1(){
}
lock void func2(){
    func1()
}

这里当第一个线程还没释放锁的时候又进行加锁,如果是不可重入锁那么就是死锁.

情形二: 有两个线程 有两个锁

这两个线程同时并发执行就可能出现死锁.

7.2 哲学家就餐问题

7.3 如何避免死锁

死锁产生的四个必要条件：

1. 互斥使用: 即当资源被一个线程使用(占有)时，别的线程不能使用
2. 不可抢占: 资源请求者不能强制从资源占有者手中夺取资源，资源只能由资源占有者主动释放。
3. 请求和保持: 即当资源请求者在请求其他的资源的同时保持对原有资源的占有。
4. 循环等待: 即存在一个等待队列：P1占有P2的资源，P2占有P3的资源，P3占有P1的资源。这样就形成了一个等待环路

当上述四个条件都成立的时候，便形成死锁。当然，死锁的情况下如果打破上述任何一个条件，便可让死锁消失。