java并发编程

Java 内存模型规范了 JVM 如何提供按需禁用缓存和编译优化的方法

方法包括 volatile、synchronized 和 final 三个关键字

六项 Happens-Before 规则

程序的顺序性规则

volatile 变量规则

传递性

管程中锁的规则

线程 start() 规则

线程 join() 规则

操作的中间状态对外不可见

是要保证中间状态对外不可见

静态方法：类锁

实例方法：实例锁

代码块：指定对象锁

如果一个资源对应多个锁，没有形成互斥关系，可能会导致并发问题

受保护资源和锁之间合理的关联关系应该是 N:1 的关系

用不同的锁对受保护资源进行精细化管理，能够提升性能

使用同一把锁

锁能覆盖所有受保护资源

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

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

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

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

破坏占用且等待条件

破坏不可抢占条件

破坏循环等待条件

都可以暂停线程的执行

wait、notify、notifyAll只能在同步代码块或同步方法中使用，而sleep不受限制

wait方法释放锁，而sleep没有

sleep来自Thread类，wait来自Object类

sleep必须捕捉异常，wait、notify、notifyAll不需要

数据竞争（Data Race）：存在共享数据并且该数据会发生变化，通俗地讲就是有多个线程会同时读写同一数据

竞态条件（Race Condition）：程序的执行结果依赖线程执行的顺序

互斥

锁

有时线程虽然没有发生阻塞，但仍然会存在执行不下去的情况，这就是所谓的“活锁”

比如，甲乙同时相互谦让，结果又相撞

方案

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

方案

减少锁带来的串行比

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

减少锁持有时间

吞吐量：指的是单位时间内能处理的请求数量

延迟：指的是从发出请求到收到响应的时间

并发量

Hasen 模型

Hoare 模型

MESA 模型

互斥

同步

通用生命周期：五态模型

java中线程的生命周期

目的：提升程序性能，主要是降低延迟，提高吞吐量

在并发编程领域，提升性能本质上就是提升硬件的利用率，
再具体点来说，就是提升 I/O 的利用率和 CPU 的利用率

我们需要解决 CPU 和 I/O 设备综合利用率的问题

CPU 密集型的计算场景

I/O 密集型计算场景

每个方法在调用栈里都有自己的独立空间，称为栈帧，
每个栈帧里都有对应方法需要的参数和返回地址

栈帧和方法是同生共死的

局部变量就是放到了调用栈里

每个线程都有自己独立的调用栈

线程封闭：仅在单线程内访问数据

原因分析

所有的递归算法都可以用非递归算法实现？

将属性和实现细节封装在对象内部，外界对象只能通过目标对象提供的公共方法来间接访问这些内部属性

将共享变量作为对象属性封装在内部，对所有公共方法制定并发访问策略

对于这些不会发生变化的共享变量，建议你用 final 关键字来修饰

一定要识别出所有共享变量之间的约束条件，
如果约束条件识别不足，很可能导致制定的并发访问策略南辕北辙

一定要特别注意竞态条件

三件事

三条原则

一个合理的受保护资源与锁之间的关联关系应该是 N:1

锁，应是私有的、不可变的、不可重用的

// 普通对象锁
private final Object lock = new Object();
// 静态对象锁
private static final Object lock = new Object(); 

logger.debug("The var1：" +
var1 + ", var2:" + var2);

如果日志级别设置为 INFO，虽然这行代码不会写日志，
但是会计算"The var1：" + var1 + ", var2:" + var2的值，
因为方法调用前会先计算参数

使用{}占位符是写日志的一个良好习惯

在触发 InterruptedException 异常的同时，JVM 会同时把线程的中断标志位清除

经验值为“最佳线程 =2 * CPU 的核数 + 1”

最佳线程数最终还是靠压测来确定的，
实际工作中大家面临的系统，“I/O 耗时 / CPU 耗时”往往都大于 1，
所以基本上都是在这个初始值的基础上增加

一般来讲，随着线程数的增加，吞吐量会增加，延迟也会缓慢增加；
但是当线程数增加到一定程度，吞吐量就会开始下降，延迟会迅速增加。
这个时候基本上就是线程能够设置的最大值了

不同的 I/O 模型对最佳线程数的影响非常大
例如大名鼎鼎的 Nginx 用的是非阻塞 I/O，
采用的是多进程单线程结构，
Nginx 本来是一个 I/O 密集型系统，
但是最佳进程数设置的却是 CPU 的核数，
完全参考的是 CPU 密集型的算法

Java SDK 并发包通过 Lock 和 Condition 两个接口来实现管程，
其中 Lock 用于解决互斥问题，Condition 用于解决同步问题

破坏死锁中的不可抢占条件的方案

Lock接口

Lock如何保证可见性

可重入锁-ReentrantLock

可重入函数

公平锁与非公平锁

用锁的最佳实践

Condition 实现了管程模型里面的条件变量

利用两个条件变量快速实现阻塞队列

同步与异步

Dubbo 源码分析

一个计数器，一个等待队列，三个方法

init()：设置计数器的初始值

down()：计数器的值减 1；如果此时计数器的值小于 0，则当前线程将被阻塞，否则当前线程可以继续执行

up()：计数器的值加 1；如果此时计数器的值小于或者等于 0，则唤醒等待队列中的一个线程，并将其从等待队列中移除

down()、up() 这两个操作历史上最早称为 P 操作和 V 操作，所以信号量模型也被称为 PV 原语

acquire()

release()

Semaphore 可以允许多个线程访问一个临界区

对象池

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

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

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

ReentrantReadWriteLock

final Map<K, V> m =
new HashMap<>();
final ReadWriteLock rwl =
new ReentrantReadWriteLock();
// 读锁
final Lock r = rwl.readLock();
// 写锁
final Lock w = rwl.writeLock();

当获取不到缓存时，从数据库获取

// 读锁查缓存
// 判断缓存为null
// 写锁
// 然后再次检查缓存是否为null
// 查数据库并设置

升级

降级

超时机制

写锁

悲观读锁

乐观读

“乐观读”这个词，而不是“乐观读锁”，是要提醒你，乐观读这个操作是无锁的

StampedLock 提供的乐观读，是允许一个线程获取写锁的，也就是说不是所有的写操作都被阻塞

tryOptimisticRead()，获取乐观读，返回stamp

validate(stamp)，检验stamp是否被改动（期间是否存在写操作）

如果存在写操作，升级为悲观读锁

和数据库的乐观锁有异曲同工之妙

StampedLock 不支持重入

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

如果线程阻塞在 StampedLock 的 readLock() 或者 writeLock() 上时，
此时调用该阻塞线程的 interrupt() 方法，会导致 CPU 飙升

使用 StampedLock 一定不要调用中断操作，
如果需要支持中断功能，
一定使用可中断的悲观读锁 readLockInterruptibly() 和写锁 writeLockInterruptibly()

getPOrders()/getDOrders()，查询订单和派送单

check()，对账

new Thread()

join()

join()方法失效

CountDownLatch latch =
new CountDownLatch(2);

latch.countDown();

// 等待两个查询操作结束
latch.await();

对账的时候，可以同时查询下一批数据

查询相当于数据的生产者，对账是消费者

创建两个队列分别存储订单和派送单

T1（查询订单）和T2（查询psd）要保持步调一致，
获得数据后，告知T3（对账）

一个是线程 T1 和 T2 要做到步调一致

另一个是要能够通知到线程 T3

依然可以使用计数器来实现，但有更优雅可靠的实现

final CyclicBarrier barrier =
new CyclicBarrier(2, ()->{
executor.execute(()->check());
});

// 等待
barrier.await();

CountDownLatch 主要用来解决一个线程等待多个线程的场景

CyclicBarrier 是一组线程之间互相等待

CountDownLatch 的计数器是不能循环利用的，也就是说一旦计数器减到 0，再有线程调用 await()，该线程会直接通过

CyclicBarrier 的计数器是可以循环利用的，而且具备自动重置的功能，一旦计数器减到 0 会自动重置到你设置的初始值。
除此之外，CyclicBarrier 还可以设置回调函数，可以说是功能丰富

Colletions类中的线程安全包装方法

Vector

Stack

HashTable

List

Map

Set

Queue

加锁、解锁操作本身就消耗性能

同时拿不到锁的线程还会进入阻塞状态，
进而触发线程切换，线程切换对性能的消耗也很大

CPU 为了解决并发问题，提供了 CAS 指令
（CAS，全称是 Compare And Swap，即“比较并交换”）

CAS 指令包含 3 个参数：共享变量的内存地址 A、用于比较的值 B 和共享变量的新值 C；
并且只有当内存中地址 A 处的值等于 B 时，才能将内存中地址 A 处的值更新为新值 C

作为一条 CPU 指令，CAS 指令本身是能够保证原子性的

只有当目前 count 的值和期望值 expect 相等时，才会将 count 更新为 newValue

所谓自旋，其实就是循环尝试

while(count != cas(count,newValue))

cas(count,newValue) 返回的值等于count，
但不能够认为 count 的值没有被其他线程更新过

大多数情况下我们并不关心 ABA 问题，例如数值的原子递增

原子化的更新对象很可能就需要关心 ABA 问题，因为两个 A 虽然相等，但是第二个 A 的属性可能已经发生变化

使用 CAS 方案的时候，一定要考虑ABA问题

AtomicLong.getAndIncrement()

转调unsafe.getAndAddLong(
Object o, long offset, long delta)

do {
// 获取当前值
oldV = xxxx；
// 根据当前值计算新值
newV = ...oldV...
}while(!compareAndSet(oldV,newV);

基本数据类型

引用类型

数组

对象属性更新器

累加器

总结

为什么不采用一般的池化资源的设计方法

Thread 对象的所有方法，都不存在类似 execute(Runnable target) 这样的公共方法

ThreadPoolExecutor(
int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) 

ThreadPoolExecutor

不建议使用 Executors

强烈建议使用有界队列

默认拒绝策略要慎重使用

注意异常处理的问题

ThreadPoolExecutor 提供的 3 个 submit() 方法

FutureTask

编写并发程序，首先要做的就是分工，
所谓分工指的是如何高效地拆解任务并分配给线程

用两个线程 T1 和 T2 来完成烧水泡茶程序，
T1 负责洗水壶、烧开水、泡茶这三道工序，
T2 负责洗茶壶、洗茶杯、拿茶叶三道工序，
其中 T1 在执行泡茶这道工序时需要等待 T2 完成拿茶叶的工序

任务之间有依赖关系，
比如当前任务依赖前一个任务的执行结果，
这种问题基本上都可以用 Future 来解决

在分析这种问题的过程中，
建议你用有向图描述一下任务之间的依赖关系，
同时将线程的分工也做好

无需手工维护线程

语义更清晰

代码更简练并且专注于业务逻辑

// 使用默认线程池
static CompletableFuture<Void> runAsync(Runnable runnable);
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier);
// 可以指定线程池
static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor);
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)  ;

CompletionStage 接口可以清晰地描述任务之间的这种时序关系

描述串行关系

描述 AND 汇聚关系

描述 OR 汇聚关系

异常处理

ReactiveX的发展（Java 语言的实现版本是 RxJava），
解决“回调地狱”的问题

注意捕获异常

解决先获取到的报价先保存到数据库的问题

ExecutorCompletionService(Executor executor)；
ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue)

Future<V> submit(Callable<V> task);
Future<V> submit(Runnable task, V result);
// 如果阻塞队列是空的，那么调用 take() 方法的线程会被阻塞
Future<V> take() throws InterruptedException;
// 如果阻塞队列是空的，那么调用 poll() 方法会返回 null 值
Future<V> poll();
// 支持以超时的方式获取并移除阻塞队列头部的一个元素，
//如果等待了 timeout unit 时间，阻塞队列还是空的，那么该方法会返回 null 值
Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;

Forking 的集群模式，支持并行地调用多个查询服务，
只要有一个成功返回结果，整个服务就可以返回了

// 创建线程池
// 创建 CompletionService
// 用于保存 Future 对象
List<Future<Integer>> futures =new ArrayList<>(3);
// 提交异步任务，并保存 future 到 futures
// 获取最快返回的任务执行结果
Integer r = 0;
try {
// 只要有一个成功返回，则 break
for (int i = 0; i < 3; ++i) {
r = cs.take().get();
// 简单地通过判空来检查是否成功返回
if (r != null) {
break;
}
}
} finally {
// 取消所有任务
for(Future<Integer> f : futures)
f.cancel(true);
}
// 返回结果
return r;

当需要批量提交异步任务的时候建议使用 CompletionService

CompletionService 将线程池 Executor 和阻塞队列 BlockingQueue 的功能融合在了一起，
能够让批量异步任务的管理更简单

CompletionService 能够让异步任务的执行结果有序化，先执行完的先进入阻塞队列

Fork/Join 是一个并行计算的框架，主要就是用来支持分治任务模型的

Fork 对应的是分治任务模型里的任务分解，
Join 对应的是结果合并

Fork/Join 计算框架主要包含两部分，
一部分是分治任务的线程池 ForkJoinPool，
另一部分是分治任务 ForkJoinTask

ForkJoinTask

注意事项

生产者 - 消费者模式

多个任务队列

支持一种叫做“任务窃取”的机制

分治、递归，
注意控制基准条件，
避免调用次数过多，
导致栈溢出

Fork/Join 并行计算框架主要解决的是分治任务

compute方法（递归），注意栈溢出的问题

Java 1.8 提供的 Stream API 里面并行流也是以 ForkJoinPool 为基础的。
默认情况下所有的并行流计算都共享一个 ForkJoinPool

建议用不同的 ForkJoinPool 执行不同类型的计算任务

所谓不变性，简单来讲，就是对象一旦被创建之后，状态就不再发生变化

类final

属性final

属性类型是不变性类

只允许只读方法
或者写方法返回新对象

利用享元模式可以减少创建对象的数量，从而减少内存占用

Java 语言里面 Long、Integer、Short、Byte 等这些基本数据类型的包装类都用到了享元模式
只缓存范围：-128,127

享元模式本质上其实就是一个对象池

对象的所有属性都是 final 的，并不能保证不可变性

不可变对象也需要正确发布

在使用 Immutability 模式的时候一定要确认保持不变性的边界在哪里，
是否要求属性对象也具备不可变性

最大应用领域：函数式编程领域

操作系统领域（父子进程fork）

java容器

Docker容器镜像设计

git设计思想

static final AtomicLong nextId=new AtomicLong(0);
// 定义 ThreadLocal 变量
static final ThreadLocal<Long> tl=ThreadLocal.withInitial(
()->nextId.getAndIncrement());
// 此方法会为每个线程分配一个唯一的 Id
static long get(){
return tl.get();
}

可以通过ThreadLocal，让对象线程私有

Thread 这个类内部有一个私有属性 threadLocals，
其类型就是 ThreadLocalMap，
ThreadLocalMap 的 Key 是 ThreadLocal

ThreadLocal是代理类

在线程池中使用 ThreadLocal可能导致内存泄露

线程池中线程的存活时间太长，往往都是和程序同生共死的，
这就意味着 Thread 持有的 ThreadLocalMap 一直都不会被回收

再加上 ThreadLocalMap 中的 Entry 对 ThreadLocal 是弱引用（WeakReference），
所以只要 ThreadLocal 结束了自己的生命周期是可以被回收掉的

但是 Entry 中的 Value 却是被 Entry 强引用的，
所以即便 Value 的生命周期结束了，
Value 也是无法被回收的，从而导致内存泄露

try{}finally{}方案，手动释放资源

管程的一个经典用法

ReentrantLock

get() 方法通过条件变量的 await() 方法实现等待，
onChanged() 方法通过条件变量的 signalAll() 方法实现唤醒功能

GuardedObject 内部维护了一个 Map，
其 Key 是 MQ 消息 id，而 Value 是 GuardedObject 对象实例

同时增加了静态方法 create() 和 fireEvent()；
create() 方法用来创建一个 GuardedObject 对象实例，并根据 key 值将其加入到 Map 中，
而 fireEvent() 方法则是模拟的大堂经理根据包间找就餐人的逻辑

synchronied

使用 volatile 的前提是对原子性没有要求

频繁创建、销毁 Java 线程的成本有点高

而且无限制地创建线程还可能导致应用 OOM

轻量级线程被叫做Fiber

用创建有界的队列来接收任务

在创建线程池时，清晰地指明拒绝策略

在实际工作中给线程赋予一个业务相关的名字

为不同的任务创建不同的线程池

提交到相同线程池中的任务一定是相互独立的，否则就一定要慎重

两阶段

interrupt() 方法和线程终止的标志位

shutdown()

shutdownNow()

解耦

支持异步

并且能够平衡生产者和消费者的速度差异

批量执行任务

使用 Immutability 模式需要注意对象属性的不可变性

使用 Copy-on-Write 模式需要注意性能问题

使用线程本地存储模式需要注意异步执行问题

Guarded Suspension 模式的经典实现是使用管程

 Balking 模式最容易忽视的就是竞态条件问题

Thread-Per-Message 模式在实现的时候需要注意是否存在线程的频繁创建、销毁以及是否可能导致 OOM

Worker Thread 模式的实现，需要注意潜在的线程死锁问题

生产者 - 消费者模式

核心是要想通过限流器，必须拿到令牌。
也就是说，只要我们能够限制发放令牌的速率，
那么就能控制流速

1、令牌以固定的速率添加到令牌桶中，假设限流的速率是 r/ 秒，则令牌每 1/r 秒会添加一个；
2、假设令牌桶的容量是 b ，如果令牌桶已满，则新的令牌会被丢弃；
3、请求能够通过限流器的前提是令牌桶中有令牌。

b 其实是 burst 的简写，意义是限流器允许的最大突发流量

生产者 - 消费者模式

一个生产者线程定时向阻塞队列中添加令牌，
而试图通过限流器的线程则作为消费者线程，
只有从阻塞队列中获取到令牌，才允许通过限流器

高并发场景，而且系统压力已经临近极限了，此时这个实现就有问题

问题就出在定时器上，在高并发场景下，
当系统压力已经临近极限的时候，定时器的精度误差会非常大，
同时定时器本身会创建调度线程，也会对系统的性能产生影响

只需要记录一个下一令牌产生的时间，
并动态更新它，就能够轻松完成限流功能

burst（最大突发流量/令牌桶的容量）为1的情况

burst（最大突发流量/令牌桶的容量）为n的情况

经典的限流算法有两个，
一个是令牌桶算法（Token Bucket），
另一个是漏桶算法（Leaky Bucket）

使用 BIO 模型，一般都会为每个 socket 分配一个独立的线程

Java 里还提供了非阻塞式（NIO）API，利用非阻塞式 API 就能够实现一个线程处理多个连接

Reactor模式类结构图

Reactor 模式的核心自然是Reactor 这个类，
其中 register_handler() 和 remove_handler() 这两个方法可以注册和删除一个事件处理器；
handle_events() 方式是核心，也是 Reactor 模式的发动机，这个方法的核心逻辑如下：
首先通过同步事件多路选择器提供的 select() 方法监听网络事件，
当有网络事件就绪后，就遍历事件处理器来处理该网络事件

Netty 中最核心的概念是事件循环（EventLoop），
其实也就是 Reactor 模式中的 Reactor，
负责监听网络事件并调用事件处理器进行处理

还有一个核心概念是EventLoopGroup

每个网络连接都关联到了一个线程上，这样做的好处是：
对于一个网络连接，读写操作都是单线程执行的，从而避免了并发程序的各种问题

在 Disruptor 中，生产者生产的对象（也就是消费者消费的对象）称为 Event，
使用 Disruptor 必须自定义 Event，例如示例代码的自定义 Event 是 LongEvent；

构建 Disruptor 对象除了要指定队列大小外，还需要传入一个 EventFactory，
示例代码中传入的是LongEvent::new；

消费 Disruptor 中的 Event 需要通过 handleEventsWith() 方法注册一个事件处理器，
发布 Event 则需要通过 publishEvent() 方法

Java SDK 中 ArrayBlockingQueue 使用数组作为底层的数据存储，
而 Disruptor 是使用RingBuffer作为数据存储

RingBuffer 本质上也是数组

程序的局部性原理指的是在一段时间内程序的执行会限定在一个局部范围内

CPU 从内存中加载数据 X 时，会将数据 X 缓存在高速缓存 Cache 中，
实际上 CPU 缓存 X 的同时，还缓存了 X 周围的数据，
因为根据程序具备局部性原理，X 周围的数据也很有可能被访问

Disruptor 内部的 RingBuffer 也是用数组实现的，
但是这个数组中的所有元素在初始化时是一次性全部创建的，
所以这些元素的内存地址大概率是连续的

消费者线程在消费的时候，是遵循空间局部性原理的，
消费完第 1 个元素，很快就会消费第 2 个元素；
当消费第 1 个元素 E1 的时候，CPU 会把内存中 E1 后面的数据也加载进 Cache，
如果 E1 和 E2 在内存中的地址是连续的，那么 E2 也就会被加载进 Cache 中，
然后当消费第 2 个元素的时候，由于 E2 已经在 Cache 中了，所以就不需要从内存中加载了，
这样就能大大提升性能

除此之外，在 Disruptor 中，生产者线程通过 publishEvent() 发布 Event 的时候，
并不是创建一个新的 Event，而是通过 event.set() 方法修改 Event，
也就是说 RingBuffer 创建的 Event 是可以循环利用的，
这样还能避免频繁创建、删除 Event 导致的频繁 GC 问题

伪共享和 CPU 内部的 Cache 有关，Cache 内部是按照缓存行（Cache Line）管理的，缓存行的大小通常是 64 个字节；
CPU 从内存中加载数据 X，会同时加载 X 后面（64-size(X)）个字节的数据

伪共享指的是由于共享缓存行导致缓存无效的场景

方案很简单，每个变量独占一个缓存行、不共享缓存行就可以了，具体技术是缓存行填充

Disruptor 中的 RingBuffer 维护了入队索引，但是并没有维护出队索引，
这是因为在 Disruptor 中多个消费者可以同时消费，每个消费者都会有一个出队索引，所以 RingBuffer 的出队索引是所有消费者里面最小的那一个

入队操作的逻辑：如果没有足够的空余位置，就出让 CPU 使用权，然后重新计算；反之则用 CAS 设置入队索引

remove(Object element) 方法的查找顺序变成了逆序查找

get(int index) 方法没有对 index 参数进行越界检查，
HiKariCP 能保证不会越界，所以不用每次都进行越界检查

用于存储所有的数据库连接的共享队列 sharedList

线程本地存储 threadList

等待数据库连接的线程数 waiters

分配数据库连接的工具 handoffQueue

处理能力，处理接收到的消息

存储能力，Actor 可以存储自己的内部状态，
并且内部状态在不同 Actor 之间是绝对隔离的

通信能力，Actor 可以和其他 Actor 之间通信