首页  思维导图  详情

并发编程（王宝令）

2022-01-12 17:31:43   24  举报





AI智能生成

并发编程（王宝令）学习总结

java

并发编程

王宝令

作者其他创作

大纲/内容

1. 可见性、原子性和有序性问题：并发编程Bug的源头

1. cpu缓存导致的可见性问题

2. 线程切换带来的原子性问题

3. 有序性问题

java编译优化

cup处理器指令优化（如指令重排）

2. Java内存模型：看Java如何解决可见性和有序性问题

Happens-Before 原则（即前面一个操作的结果对后续操作是可见的）

1. 程序的顺序性规则

2. volatile变量规则

3. 传递性

4. 管程中锁的规则

一个线程在解锁前的操作对后面获取该锁的线程来说是可见的

5. 线程 start() 规则

线程A通过start()方法启动线程B，则线程A执行start()前的操作对线程B是可见的

6. 线程 join() 规则

在线程A中调用线程B#join方法，则线程B中的操作对B.join之后的代码是可见的(注意B#start要在B#join前面)

3. 互斥锁（上）：解决原子性问题

锁和资源间的对应关系

受保护资源和锁之间的关联关系是N:1的关系，一把锁能锁住多个资源，但一个资源不能被多个锁锁住，这样是不安全的

synchronized对原子性的支持

synchronized能保证同一时刻只有一个线程执行，从而保证原子性
再根据happen before第4条，管程中锁的规则：对一个锁的解锁 Happens-Before 于后续对这个锁的加锁。所以也能保证可见性

4. 互斥锁（下）：如何用一把锁保护多个资源？

1. 保护没有关联关系的多个资源

2. 保护有关联关系的多个资源

错误示例

正确示例1

但有性能问题，转账操作会锁住所有的账号

正确示例2

但有性能问题，转账操作会锁住所有的账号

5. 一不小心就死锁了怎么办？

1. 通过细粒度的锁解决转账过程中的性能问题

引生出死锁问题

2. 死锁产生的条件

1. 互斥，共享资源X和Y都只能被一个线程占用；

2. 占有且等待，线程T1已经取得共享资源X，在等待共享资源Y的时候，不释放共享资源X；

3. 不可抢占，其他线程不能强行抢占线程T1占有的资源；

4. 循环等待，线程T1等待线程T2占有的资源，线程T2等待线程T1占有的资源，就是循环等待。

3. 如何预防死锁

1. 破坏占用且等待条件（同时申请两个锁）

2. 破坏不可抢占条件

java.util.concurrent这个包下面提供的Lock可以轻松解决这个问题

3. 破坏循环等待条件（按顺序进行加锁）

6. 用“等待-通知”机制优化循环等待

使用 wait()和notifyAll()优化#5.3.3中的示例

注意：Allocator中释放资源的时候使用notifyAll，原因见Allocator类注解

7. 安全性、活跃性以及性能问题

1. 安全性

即程序按照我们期望的执行，产生线程安全问题的条件是存在共享数据并且该数据会发生变化，通俗地讲就是有多个线程会同时读写同一数据

2. 活跃性

所谓活跃性问题，指的是某个操作无法执行下去
我们常见的“死锁”就是一种典型的活跃性问题
除了死锁外，还有两种情况，分别是“活锁”和“饥饿”

活锁

指两个线程之间相互”谦让“，成为没有发生阻塞但依然执行不下去的“活锁”
解决方案是让线程等待一个随机时间

饥饿

“饥饿”指的是线程因无法访问所需资源而无法执行下去的情

3. 性能

解决性能问题

1. 使用无锁的算法和数据结构了

2. 减少锁持有的时间。互斥锁本质上是将并行的程序串行化，所以要增加并行度，一定要减少持有锁的时间

性能方面的度量指标

1. 吞吐量

指的是单位时间内能处理的请求数量。吞吐量越高，说明性能越好

2. 延迟

指的是从发出请求到收到响应的时间。延迟越小，说明性能越好

3. 并发量

指的是能同时处理的请求数量，一般来说随着并发量的增加、延迟也会增加。所以延迟这个指标
一般都会是基于并发量来说的。例如并发量是1000的时候，延迟是50毫秒。

8. 管程：并发编程的万能钥匙

管程模型

1. Hasen模型

2. Hoare模型

3. MESA模型（java基于该模型实现）

单条件变量的MESA模型（如：synchronized）
属于jvm内置的实现

synchronized管程（对象监视器）模型原理

synchronized锁升级过程

wait方法使用范式

需要在while循环里面调用wait，因为wait的线程被notify
通知后会进入锁等待队列里面重新获取锁，并不会被立即执行
所以可能会被其他线程拿到锁，需要重新判断是否要再次wait
原理可查看上面的模型原理图

wait正确使用实现简单队列的示例

wait不被立即执行示例

notify()何时可以使用

1. 所有等待线程拥有相同的等待条件（#6示例中不满足该条件）

2. 所有等待线程被唤醒后，执行相同的操作

3. 只需要唤醒一个线程

join 实现原理

其实就是让main线程wait在join的线程对象上，在join的对象结束前会调用notifyAll方法

实现原理示例

关于String和Integer类型和对象属性字段不适合做锁对象

首先对象中的属性是可能会被改变，一旦被改变就无法保证互斥性了
而String,Integer,Boolean对象在JVM里面是可能被重用的。
如果其他代码 synchronized(你的锁)，而且不释放，那你的程序就永远拿不到锁

属性被改示例代码

被重用示例代码

多条件变量的MESA模型（如：ReentrantLock）
属于接口层面的实现

和单条件变量模型原理一样，只是支持多条件变量

9. Java线程（上）：Java线程的生命周期

Java中线程的生命周期

NEW（初始化状态）

RUNNABLE（可运行/运行状态，io阻塞时也为该状态）

BLOCKED（阻塞状态）

WAITING（无时限等待）

TIMED_WAITING（有时限等待）

TERMINATED（终止状态）

状态间的转换

RUNNABLE与BLOCKED的状态转换

有一种场景会触发这种转换，就是线程等待synchronized的隐式锁

RUNNABLE与WAITING的状态转换

1. 获得synchronized隐式锁的线程，调用无参数的Object.wait()方法

2. 第二种场景，调用无参数的Thread.join()方法，其实join方法内部也是通过wait()实现的

3. 第三种场景，调用LockSupport.park()方法

RUNNABLE与TIMED_WAITING的状态转换

RUNNABLE与WAITING的状态转换中的场景带上超时时间+Thread.sleep(long millis)

从NEW到RUNNABLE状态

调用线程的start()方法

从RUNNABLE到TERMINATED状态

1. run方法执行完

2. run方法执出现异常

3. 调用stop方法（已标记为@Deprecated()，调用会立即杀死线程，建议使用interrupt()方法代替）

interrupt方法在不同线程不同状态下调用

1. 当线程处于WAITING、TIMED_WAITING状态时，分两种情况

1. 通过Object.wait()方法进入该状态
线程返回到RUNNABLE状态，同时会触发InterruptedException异常，并在异常代码执行前会重置中断状态为false

示例代码

2. 通过LockSupport.park()进入该状态
线程返回到RUNNABLE状态，不抛出异常，且不会重置中断状态

2. 当线程处于RUNNABLE状态且阻塞在java.nio.channels.InterruptibleChannel上时

线程会触发java.nio.channels.ClosedByInterruptException这个异常

3. 当线程处于RUNNABLE状态且阻塞在java.nio.channels.Selector上时

线程会触发java.nio.channels.ClosedByInterruptException这个异常

4. 当线程处于RUNNABLE状态并且没有阻塞在某个I/O操作上时

线程可以通过isInterrupted()方法，检测是不是自己被中断了
注意使用Thread.interrupted();也可以返回中断状态，不过它返回的同时会重置中断状态

引起上下文切换的原因

1. 时间片用完，CPU正常调度下一个任务

演示时间片用完导致上下文切换

2. 被其他优先级更高的任务抢占

3. 执行任务碰到IO阻塞，调度器挂起当前任务，切换执行下一个任务

4. 用户代码主动挂起当前任务让出CPU时间

演示线程频繁挂起导致线程上下文频繁切换

5. 多任务抢占资源，由于没有抢到被挂起

6. 硬件中断

10. Java线程（中）：创建多少线程才是合适的？

延迟与吞吐量

1. 将方法中能够并行的操作使用多线程异步处理减少方法的执行时间（延迟降低，提升吞吐）

2. 使用多线程将单机性能发挥到极致（延迟不变，提升吞吐）

3. 单机性能瓶颈时集群部署（提升吞吐，延迟不变）

针对不同场景的线程设置

纯cup计算

理论上“设置线程的数量=CPU核数”，不过在工程上，线程的数量一般会设置为“CPU核数+1”

CPU计算和I/O操作的耗时是1:1

单核下2个线程是最合适的，当一个线程执行cup计算时，另外一个线程可以执行io操作（如数据库查询操作）

CPU计算和I/O操作的耗时是1:2

在io不存在瓶颈的情况下，单核下3个线程是最合适的

14. Lock和Condition（上）：隐藏在并发包中的管程

Lock的实现类ReentrantLock

ReentrantLock类关系图

ReentrantLock公平锁流程图及示例

Lock 通过”破坏不可抢占条件“解决synchronizatized无法解决的死锁的问题

3. boolean tryLock(); // 支持非阻塞获取锁的API

ReentrantLock中的实现：通过cas操作进行获取锁，获取失败不会阻塞线程

2. boolean tryLock(long time, TimeUnit unit) //支持超时的API

ReentrantLock中的实现：和ReentrantLock#lock方法类似，只不过阻塞线程会通过LockSupport.parkNanos(this, nanosTimeout);设置超时时间
超时后tryLock返回false，还有一点不同的是阻塞期间如果中断线程会直接抛出InterruptedException异常

1. void lockInterruptibly() //支持中断的API

ReentrantLock中的实现：和ReentrantLock#lock方法类似，阻塞期间如果中断线程会直接抛出InterruptedException异常

java中的各种锁

已获取锁的线程是否可以再次进入该锁

可重入锁

synchronized

在对象监视器中实现锁重入，具体原理查看“synchronized锁升级过程”

重入示例

ReentrantLock

在获取锁时如果当前线程已经获取锁会对state进行+1，具体查看ReentrantLock内部类FairSync或NonfairSync对AQS#tryAcquire方法的重写实现

重入示例

不可重入锁

多线程竞争同一把锁时会不会锁住同步资源

乐观锁

CAS自旋

悲观锁

synchronized和ReentrantLock都是悲观锁，在存在锁竞争时会锁住同步资源

多个线程是否可以获取同一把锁

共享锁

排他锁

先进入的线程是否优先获取锁

公平锁

ReentrantLock#FairSync，在实现AQS#tryAcquire方法时会先调用hasQueuedPredecessors方法判断队列中是否有其他线程等待获取锁，在进行cas获取锁操作，防止新进入的线程抢走队列中其他等待线程的锁

非公平锁

ReentrantLock#NonfairSync

synchronized锁升级过程中涉及到的锁

无锁

偏向锁

轻量级锁

重量级锁

15. Lock和Condition（下）：Dubbo如何用管程实现异步转同步？

Condition流程图及示例分析

synchronized单条件变量示例

ReentrantLock和Condition实现缓冲区

16. Semaphore：如何快速实现一个限流器？

Semaphore#acquire流程图

Semaphore和ReentrantLock获取锁的主要区别

Semaphore

取锁时会对state进行减操作，如果减后的值大于等于0则当前线程获取锁成功（多个线程可以同时获取锁），否则获取锁失败会添加到QAS队列中等待其他线程释放锁

ReentrantLock

取锁时是对state进行加操作，而且只有当state为0或是持有锁的线程重入才能进行该操作，否则取锁失败会添加到QAS队列中等待持有锁的线程释放锁

信号量实现限流器

17. ReadWriteLock：如何快速实现一个完备的缓存？

什么是读写锁

1. 允许多个线程同时读共享变量；

2. 只允许一个线程写共享变量；

3. 如果一个写线程正在执行写操作，此时禁止读线程读共享变量。

实现一个缓存示例

实现缓存的按需加载示例

ReadWriteLock注意事项

支持锁降级，但不支持锁升级，如果进行锁升级，可能会导致写锁永久等待

18. StampedLock：有没有比读写锁更快的锁？

StampedLock支持的三种锁模式

1. 写锁，相当于ReadWriteLock的写锁

2. 悲观读锁，相当于ReadWriteLock的读锁

3. 乐观读，这个操作是无锁的，所以相比较ReadWriteLock的读锁，乐观读的性能更好一些

乐观读的正确使用方式及基本实现原理

StampedLock使用注意事项

1. StampedLock是不可重入锁

2. StampedLock的悲观读锁、写锁都不支持条件变量

3. 如果线程阻塞在StampedLock的readLock()或者writeLock()上时，此时调用该阻塞线程的interrupt()方法，会导致CPU飙升。
如果需要支持中断功能，一定使用可中断的悲观读锁readLockInterruptibly()和写锁writeLockInterruptibly()

19. CountDownLatch和CyclicBarrier：如何让多线程步调一致？

CountDownLatch执行流程图

CyclicBarrier#await执行流程

异常订单优化案例

Reconciliation0（最原始的做法，串行化）

Reconciliation1（对Reconciliation0进行优化,将查询未对账订单和查询派送单同时进行）

Reconciliation2（使用线程池和wait+notify 优化Reconciliation1）

Reconciliation3（使用线程池和CountDownLatch 优化Reconciliation2）

Reconciliation4（使用ReentrantLock和Condition优化Reconciliation3，将查询操作和对账保存操作也并行）

Reconciliation5（通过CyclicBarrier对 Reconciliation4进行优化,将查询操作和对账保存操作也并行）

20. 并发容器：都有哪些“坑”需要我们填？

容器相关

21. 原子类：无锁工具类的典范

底层主要是Unsafe类提供的本地方法调用一条cas指令进行实现

jdk提供的原子类

原子化的基本数据类型

AtomicInteger

AtomicInteger#incrementAndGet源码

使用示例

和使用synchronized性能对比

AtomicLong

AtomicBoolean

原子化的对象引用类型，用来修改引用类型

AtomicReference

可查看AbstractQueuedSynchronizer#addWaiter方法在设置新入节点为尾节点时有使用该方法

AtomicStampedReference

使用示例

AtomicMarkableReference

原子化数组

AtomicIntegerArray

使用示例

AtomicLongArray

AtomicReferenceArray

原子化对象属性更新器

AtomicIntegerFieldUpdater

使用示例

AtomicLongFieldUpdater

AtomicReferenceFieldUpdater

原子化的累加器

DoubleAccumulator

DoubleAdder

LongAccumulator

LongAdder

22. Executor与线程池：如何创建正确的线程池？

模拟线程池示例

线程池的使用示例

线程池参数

corePoolSize

表示线程池保有的最小线程数。有些项目很闲，但是也不能把人都撤了，至少要留corePoolSize个人坚守阵地

maximumPoolSize

表示线程池创建的最大线程数。当项目很忙时，就需要加人，加人的前提是corePoolSize线程都在忙碌且工作队列满了（需是有界队列）
那后面每來一个任务加一个线程，但是也不能无限制地加，最多就加到maximumPoolSize个人。
当项目闲下来时，就要撤人了，最多能撤到corePoolSize个人

keepAliveTime & unit

如果一个线程空闲了keepAliveTime unit(keepAliveTime的单位)这么久，而且线程池的线程数大于 corePoolSize
那么这个空闲的线程就要被回收了

workQueue

工作队列，和MyThreadPool示例代码的工作队列同义

threadFactory

通过这个参数你可以自定义如何创建线程，例如你可以给线程指定一个有意义的名字。不传默认是pool-1-thread-* 格式

handler

如果线程池中所有的线程都在忙碌，并且工作队列也满了（前提是工作队列是有界队列）
那么此时提交任务，线程池就会拒绝接收

4种策略

1. CallerRunsPolicy：提交任务的线程自己去执行该任务。

2. AbortPolicy：默认的拒绝策略，会throws RejectedExecutionException。

3. DiscardPolicy：直接丢弃任务，没有任何异常抛出。

4. DiscardOldestPolicy：丢弃最老的任务，其实就是把最早进入工作队列的任务丢弃，然后把新任务加入到工作队列。

executors

ScheduledThreadPoolExecutor

示例

23. Future：如何用多线程实现最优的“烧水泡茶”程序

FutureTask类关系

FutureTask原理分析

线程池提供的三个submit方法

submit(Callable<T> task)

这个方法的参数是一个Callable对象，内部默认通过FutureTask#run（该run方法为线程池执行的run方法）进行调用Callable对象的call方法并将结果设置到FutureTask的内部属性中，可以通过submit返回的Future#get获取返回的结果

使用示例

submit(Runnable task, T result)

其实这种调用方式最终会通过RunnableAdapter将task和result适配成Callable，call方法中会调用task.run并返回result

使用示例

submit(Runnable task)

这种方式和上面的调用没啥区别，只是result传参被null代替

使用示例

24. CompletableFuture：异步编程没那么难

相对于“线程池+Future”的方式，CompletableFuture主要用于解决任务之间有聚合关系的场景

CompletableFuture执行异步任务的静态方法

1. runAsync(Runnable runnable) //用于执行没有返回值的异步任务，默认使用ForkJoinPool线程池

2. supplyAsync(Supplier<U> supplier) //用于执行有返回值的异步任务，默认使用ForkJoinPool线程池

3. runAsync(Runnable runnable, Executor executor) //和1一样，只是指定线程池

4. supplyAsync(Supplier<U> supplier, Executor executor) //和2一样，只是指定线程池

CompletableFuture接口实现了CompletionStage接口，用于描述多个任务之间的关系，包括

1. 串行关系

CompletionStage<R> thenApply(fn);
CompletionStage<R> thenApplyAsync(fn);
CompletionStage<Void> thenAccept(consumer);
CompletionStage<Void> thenAcceptAsync(consumer);
CompletionStage<Void> thenRun(action);
CompletionStage<Void> thenRunAsync(action);
CompletionStage<R> thenCompose(fn);
CompletionStage<R> thenComposeAsync(fn);

示例CompletionStageTest#test1

2. AND汇聚关系

CompletionStage<R> thenCombine(other, fn);
CompletionStage<R> thenCombineAsync(other, fn);
CompletionStage<Void> thenAcceptBoth(other, consumer);
CompletionStage<Void> thenAcceptBothAsync(other, consumer);
CompletionStage<Void> runAfterBoth(other, action);
CompletionStage<Void> runAfterBothAsync(other, action);

示例CompletionStageTest#test2

3. OR汇聚关系

CompletionStage applyToEither(other, fn);
CompletionStage applyToEitherAsync(other, fn);
CompletionStage acceptEither(other, consumer);
CompletionStage acceptEitherAsync(other, consumer);
CompletionStage runAfterEither(other, action);
CompletionStage runAfterEitherAsync(other, action);

示例CompletionStageTest#test3

4. 异常处理

CompletionStage exceptionally(fn);
CompletionStage<R> whenComplete(consumer);
CompletionStage<R> whenCompleteAsync(consumer);
CompletionStage<R> handle(fn);
CompletionStage<R> handleAsync(fn);

示例CompletionStageTest#test4

25. CompletionService：如何批量执行异步任务？

相对于CompletableFuture，CompletionService主要用于批量并行任务结果处理

使用示例

26. Fork/Join：单机版的MapReduce

把一个复杂的问题分解成多个相似的子问题，然后再把子问题分解成更小的子问题，直到子问题简单到可以直接求解

使用示例

执行原理

比较复杂，后续有时间再研究

28. Immutability模式：如何利用不变性解决并发问题？

将一个类所有的属性都设置成final的，并且只允许存在只读方法，那么这个类基本上就具备不可变性了，如String，Integer类

29. Copy-on-Write模式：不是延时策略的COW