首页  思维导图  详情

Java并发体系

2018-08-17 17:33:10   51  举报





AI智能生成

Java并发体系，详细分析Java并发

作者其他创作

大纲/内容

Java并发集合

ConcurrentHashMap

CAS + Synchronized 来保证并发更新的安全，底层采用数组+链表/红黑树的存储结构

重要内部类

Node

key-value键值对

TreeNode

红黑树节点

TreeBin

就相当于一颗红黑树，其构造方法其实就是构造红黑树的过程

ForwardingNode

辅助节点，用于ConcurrentHashMap扩容操作

sizeCtl

控制标识符，用来控制table初始化和扩容操作的

含义

负数代表正在进行初始化或扩容操作

-1代表正在初始化

-N 表示有N-1个线程正在进行扩容操作

正数或0代表hash表还没有被初始化，这个数值表示初始化或下一次进行扩容的大小

重要操作

initTable

ConcurrentHashMap初始化方法

只能有一个线程参与初始化过程，其他线程必须挂起

构造函数不做初始化过程，初始化真正是在put操作触发

步骤

sizeCtl < 0 表示正在进行初始化，线程挂起

初始化步骤完成后，设置sizeCtl = 0.75 * n（下一次扩容阈值），表示下一次扩容的大小

put

核心思想

根据hash值计算节点插入在table的位置，如果该位置为空，则直接插入，否则插入到链表或者树中

真实情况较为复杂

table为null，线程进入初始化步骤，如果有其他线程正在初始化，该线程挂起

如果插入的当前 i 位置为null，说明该位置是第一次插入，利用CAS插入节点即可，插入成功，则调用addCount判断是否需要扩容。若插入失败，则继续匹配（自旋）

若该节点的hash == MOVED（-1），表示有线程正在进行扩容，则进入扩容进程中

其余情况就是按照链表或者红黑树结构插入节点，但是这个过程需要加锁（synchronized）

get

table ==null ；return null

从链表/红黑树节点获取

扩容

多线程扩容

构建一个nextTable，其大小为原来大小的两倍，这个步骤是在单线程环境下完成的

将原来table里面的内容复制到nextTable中，这个步骤是允许多线程操作

链表转换为红黑树过程

所在链表的元素个数达到了阈值 8，则将链表转换为红黑树

红黑树算法

1.8 与 1.7的区别

ConcurrentLinkedQueue

基于链接节点的无边界的线程安全队列，采用FIFO原则对元素进行排序，内部采用CAS算法实现

不变性

在入队的最后一个元素的next为null

队列中所有未删除的节点的item都不能为null且都能从head节点遍历到

对于要删除的节点，不是直接将其设置为null，而是先将其item域设置为null（迭代器会跳过item为null的节点）

允许head和tail更新滞后。这是什么意思呢？意思就说是head、tail不总是指向第一个元素和最后一个元素（后面阐述）

head的不变性和可变性

tail的不变性和可变性

精妙之处：利用CAS来完成数据操作，同时允许队列的不一致性，弱一致性表现淋漓尽致

ConcurrentSkipListMap

第三种key-value数据结构：SkipList（跳表）

SkipList

平衡二叉树结构

Skip list让已排序的数据分布在多层链表中，以0-1随机数决定一个数据的向上攀升与否，通过“空间来换取时间”的一个算法，在每个节点中增加了向前的指针，在插入、删除、查找时可以忽略一些不可能涉及到的结点，从而提高了效率

特性

由很多层结构组成，level是通过一定的概率随机产生的

每一层都是一个有序的链表，默认是升序，也可以根据创建映射时所提供的Comparator进行排序，具体取决于使用的构造方法

最底层(Level 1)的链表包含所有元素

如果一个元素出现在Level i 的链表中，则它在Level i 之下的链表也都会出

每个节点包含两个指针，一个指向同一链表中的下一个元素，一个指向下面一层的元素

查找、删除、添加

ConcurrentSkipListSet

内部采用ConcurrentSkipListMap实现

atomic

基本类型类

用于通过原子的方式更新基本类型

AtomicBoolean

原子更新布尔类型

AtomicInteger

原子更新整型

AtomicLong

原子更新长整型

数组

通过原子的方式更新数组里的某个元素

AtomicIntegerArray

原子更新整型数组里的元素

AtomicLongArray

原子更新长整型数组里的元素

AtomicReferenceArray

原子更新引用类型数组里的元素

引用类型

如果要原子的更新多个变量，就需要使用这个原子更新引用类型提供的类

AtomicReference

原子更新引用类型

AtomicReferenceFieldUpdater

原子更新引用类型里的字段

AtomicMarkableReference

原子更新带有标记位的引用类型

字段类

如果我们只需要某个类里的某个字段，那么就需要使用原子更新字段类

AtomicIntegerFieldUpdater

原子更新整型的字段的更新器

AtomicLongFieldUpdater

原子更新长整型字段的更新器

AtomicStampedReference

原子更新带有版本号的引用类型

阻塞队列

ArrayBlockingQueue

一个由数组实现的FIFO有界阻塞队列

ArrayBlockingQueue有界且固定，在构造函数时确认大小，确认后不支持改变

在多线程环境下不保证“公平性”

实现

ReentrantLock

Condition

LinkedBlockingQueue

基于链接，无界的FIFO阻塞队列

PriorityBlockingQueue

支持优先级的无界阻塞队列

默认情况下元素采用自然顺序升序排序，可以通过指定Comparator来对元素进行排序

二叉堆

分类

最大堆

父节点的键值总是大于或等于任何一个子节点的键值

最小堆

父节点的键值总是小于或等于任何一个子节点的键值

添加操作则是不断“上冒”，而删除操作则是不断“下掉”

ReentrantLock + Condition

DelayQueue

支持延时获取元素的无界阻塞队列

应用

缓存：清掉缓存中超时的缓存数据

任务超时处理

根据Delay时间排序的优先级队列：PriorityQueue

Delayed接口

用来标记那些应该在给定延迟时间之后执行的对象

该接口要求实现它的实现类必须定义一个compareTo 方法，该方法提供与此接口的 getDelay 方法一致的排序。

SynchronousQueue

一个没有容量的阻塞队列

交换工作，生产者的线程和消费者的线程同步以传递某些信息、事件或者任务

难搞懂，与Exchanger 有一拼

LinkedTransferQueue

链表组成的的无界阻塞队列

相当于ConcurrentLinkedQueue、SynchronousQueue (公平模式下)、无界的LinkedBlockingQueues等的超集

预占模式

有就直接拿走，没有就占着这个位置直到拿到或者超时或者中断

LinkedBlockingDeque

由链表组成的双向阻塞队列

容量可选，在初始化时可以设置容量防止其过度膨胀，如果不设置，默认容量大小为Integer.MAX_VALUE

运用

“工作窃取”模式

线程池

好处

降低资源消耗

通过重复利用已创建的线程降低线程创建和销毁造成的消耗

提高响应速度

当任务到达时，任务可以不需要等到线程创建就能立即执行

提高线程的可管理性

进行统一分配、调优和监控\u00A0\u00A0

Executor

Executors

静态工厂类，提供了Executor、ExecutorService、ScheduledExecutorService、ThreadFactory 、Callable 等类的静态工厂方法

ThreadPoolExecutor

参数含义

corePoolSize

线程池中核心线程的数量

maximumPoolSize

线程池中允许的最大线程数

keepAliveTime

线程空闲的时间

unit

keepAliveTime的单位

workQueue

用来保存等待执行的任务的阻塞队列

使用的阻塞队列

threadFactory

用于设置创建线程的工厂

DefaultThreadFactory

handler

RejectedExecutionHandler，线程池的拒绝策略

AbortPolicy：直接抛出异常，默认策略

CallerRunsPolicy：用调用者所在的线程来执行任务

DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务

DiscardPolicy：直接丢弃任务

线程池分类

newFixedThreadPool

可重用固定线程数的线程池

分析

corePoolSize和maximumPoolSize一致

使用“无界”队列LinkedBlockingQueue

maximumPoolSize、keepAliveTime、RejectedExecutionHandler 无效

newCachedThreadPool

使用单个worker线程的Executor

corePoolSize和maximumPoolSize被设置为1

使用LinkedBlockingQueue作为workerQueue

newSingleThreadExecutor

会根据需要创建新线程的线程池\u00A0\u00A0

corePoolSize被设置为0

maximumPoolSize被设置为Integer.MAX_VALUE

SynchronousQueue 作为WorkerQueue

如果主线程提交任务的速度高于maximumPool中线程处理任务的速度时，CachedThreadPool会不断创建新线程，可能会耗尽CPU和内存资源

任务提交

Executor.execute()

ExecutorService.submit()

任务执行

执行流程

线程池调优

两种模型

线程池监控

ScheduledThreadPoolExecutor

继承自ThreadPoolExecutor

给定的延迟之后运行任务，或者定期执行任务

内部使用DelayQueue来实现，会把调度的任务放入DelayQueue中。DelayQueue内部封装PriorityQueue，这个PriorityQueue会对队列中的ScheduledFutureTask进行排序

Future

异步计算

Future

提供操作

执行任务的取消

查询任务是否完成

获取任务的执行结果

FutureTask

实现RunnableFuture接口，既可以作为Runnable被执行，也可以作为Future得到Callable的返回值

内部基于AQS实现

Java并发体系

Java内存模型（JMM）

线程通信机制

内存共享

Java采用

消息传递

内存模型

重排序

为了程序的性能，处理器、编译器都会对程序进行重排序处理

条件

在单线程环境下不能改变程序运行的结果

存在数据依赖关系的不允许重排序

问题

重排序在多线程环境下可能会导致数据不安全

顺序一致性

多线程环境下的理论参考模型

为程序提供了极强的内存可见性保证

一个线程中的所有操作必须按照程序的顺序来执行

所有线程都只能看到一个单一的操作执行顺序，不管程序是否同步

每个操作都必须原子执行且立刻对所有线程可见

happens-before

JMM中最核心的理论，保证内存可见性

在JMM中，如果一个操作执行的结果需要对另一个操作可见，那么这两个操作之间必须存在happens-before关系。

理论

如果一个操作happens-before另一个操作，那么第一个操作的执行结果将对第二个操作可见，而且第一个操作的执行顺序排在第二个操作之前

两个操作之间存在happens-before关系，并不意味着一定要按照happens-before原则制定的顺序来执行。如果重排序之后的执行结果与按照happens-before关系来执行的结果一致，那么这种重排序并不非法

as-if-serial

所有的操作均可以为了优化而被重排序，但是你必须要保证重排序后执行的结果不能被改变

synchronized

同步、重量级锁

原理

synchronized可以保证方法或者代码块在运行时，同一时刻只有一个方法可以进入到临界区，同时它还可以保证共享变量的内存可见性

锁对象

普通同步方法，锁是当前实例对象

静态同步方法，锁是当前类的class对象

同步方法块，锁是括号里面的对象

实现机制

Java对象头

synchronized的锁就是保存在Java对象头中的

包括两部分数据

Mark Word（标记字段）

Mark Word被设计成一个非固定的数据结构以便在极小的空间内存存储尽量多的数据，它会根据对象的状态复用自己的存储空间

包括

哈希码（HashCode）、GC分代年龄、锁状态标志、线程持有的锁、偏向线程 ID、偏向时间戳

Klass Pointer（类型指针）

monitor

Owner

初始时为NULL表示当前没有任何线程拥有该monitor record，当线程成功拥有该锁后保存线程唯一标识，当锁被释放时又设置为NULL

锁优化

自旋锁

该线程等待一段时间，不会被立即挂起，看持有锁的线程是否会很快释放锁（循环方式）

自旋字数较难控制（-XX:preBlockSpin）

存在理论：线程的频繁挂起、唤醒负担较重，可以认为每个线程占有锁的时间很短，线程挂起再唤醒得不偿失

缺点

自旋次数无法确定

适应性自旋锁

自旋的次数不再是固定的，它是由前一次在同一个锁上的自旋时间及锁的拥有者的状态来决定

自旋成功，则可以增加自旋次数，如果获取锁经常失败，那么自旋次数会减少

锁消除

若不存在数据竞争的情况，JVM会消除锁机制

判断依据

变量逃逸

锁粗化

将多个连续的加锁、解锁操作连接在一起，扩展成一个范围更大的锁。例如for循环内部获取锁

轻量级锁

在没有多线程竞争的前提下，减少传统的重量级锁使用操作系统互斥量产生的性能消耗

通过CAS来获取锁和释放锁

性能依据

对于绝大部分的锁，在整个生命周期内都是不会存在竞争的

在多线程环境下，其运行效率比重量级锁还会慢

偏向锁

为了在无多线程竞争的情况下尽量减少不必要的轻量级锁执行路径

主要尽可能避免不必须要的CAS操作，如果竞争锁失败，则升级为轻量级锁

volatile

volatile可见性：对一个volatile的读，总可以看到对这个变量最终的写

volatile原子性：volatile对单个读/写具有原子性（32位Long、Double），但是复合操作除外，例如i++;

内存屏障

内存语义

当写一个volatile变量时，JMM会把该线程对应的本地内存中的共享变量值立即刷新到主内存中

当读一个volatile变量时，JMM会把该线程对应的本地内存设置为无效，直接从主内存中读取共享变量

操作系统语义

主存、高速缓存（线程私有）缓存一致？

解决方案

通过在总线加LOCK#锁的方式

通过缓存一致性协议（MESI协议）

DCL

单例模式

happens-beofre

volatile方案

禁止重排序

基于类初始化的解决方案

利用classloder的机制来保证初始化instance时只有一个线程。JVM在类初始化阶段会获取一个锁，这个锁可以同步多个线程对同一个类的初始化

并发基础

AQS

AbstractQueuedSynchronizer，同步器，实现JUC核心基础组件

解决了子类实现同步器时涉及的大量细节问题，例如获取同步状态、FIFO同步队列

采用模板方法模式，AQS实现大量通用方法，子类通过继承方式实现其抽象方法来管理同步状态

CLH同步队列

FIFO双向队列，AQS依赖它来解决同步状态的管理问题

首节点唤醒，等待队列加入到CLH同步队列的尾部

同步状态获取与释放

独占式

获取锁

获取同步状态：acquire

响应中断：acquireInterruptibly

超时获取:tryAcquireNanos

释放锁

release

共享式

acquireShared

releaseShared

线程阻塞和唤醒

当有线程获取锁了，其他再次获取时需要阻塞，当线程释放锁后，AQS负责唤醒线程

LockSupport

是用来创建锁和其他同步类的基本线程阻塞原语

每个使用LockSupport的线程都会与一个许可关联，如果该许可可用，并且可在进程中使用，则调用park()将会立即返回，否则可能阻塞。如果许可尚不可用，则可以调用 unpark 使其可用

park()、unpark()

CAS

Compare And Swap，整个JUC体系最核心、最基础理论

内存值V、旧的预期值A、要更新的值B，当且仅当内存值V的值等于旧的预期值A时才会将内存值V的值修改为B，否则什么都不干

native中存在四个参数

缺陷

循环时间太长

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

ABA问题

版本号

锁

可重入锁，是一种递归无阻塞的同步机制

比synchronized更强大、灵活的锁机制，可以减少死锁发生的概率

分为公平锁、非公平锁

底层采用AQS实现，通过内部Sync继承AQS

ReentrantReadWriteLock

读写锁，两把锁：共享锁：读锁、排他锁：写锁

支持公平性、非公平性，可重入和锁降级

锁降级：遵循获取写锁、获取读锁在释放写锁的次序，写锁能够降级成为读锁

Lock 提供条件Condition，对线程的等待、唤醒操作更加详细和灵活

内部维护一个Condition队列。当前线程调用await()方法，将会以当前线程构造成一个节点（Node），并将节点加入到该队列的尾部

并发工具类

CyclicBarrier

它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)

通俗讲：让一组线程到达一个屏障时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续干活

底层采用ReentrantLock + Condition实现

应用场景

多线程结果合并的操作，用于多线程计算数据，最后合并计算结果的应用场景

CountDownLatch

在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待

用给定的计数初始化 CountDownLatch。由于调用了 countDown() 方法，所以在当前计数到达零之前，await 方法会一直受阻塞。之后，会释放所有等待的线程，await 的所有后续调用都将立即返回。这种现象只出现一次——计数无法被重置。如果需要重置计数，请考虑使用 CyclicBarrier。

与CyclicBarrier区别

CountDownLatch的作用是允许1或N个线程等待其他线程完成执行；而CyclicBarrier则是允许N个线程相互等待

CountDownLatch的计数器无法被重置；CyclicBarrier的计数器可以被重置后使用，因此它被称为是循环的barrier

内部采用共享锁来实现

Semaphore

信号量

一个控制访问多个共享资源的计数器

从概念上讲，信号量维护了一个许可集。如有必要，在许可可用前会阻塞每一个 acquire()，然后再获取该许可。每个 release() 添加一个许可，从而可能释放一个正在阻塞的获取者。但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动

信号量Semaphore是一个非负整数（>=1）。当一个线程想要访问某个共享资源时，它必须要先获取Semaphore，当Semaphore >0时，获取该资源并使Semaphore – 1。如果Semaphore值 = 0，则表示全部的共享资源已经被其他线程全部占用，线程必须要等待其他线程释放资源。当线程释放资源时，Semaphore则+1

通常用于限制可以访问某些资源（物理或逻辑的）的线程数目

内部采用共享锁实现

Exchanger

可以在对中对元素进行配对和交换的线程的同步点

允许在并发任务之间交换数据。具体来说，Exchanger类允许在两个线程之间定义同步点。当两个线程都到达同步点时，他们交换数据结构，因此第一个线程的数据结构进入到第二个线程中，第二个线程的数据结构进入到第一个线程中

其他

ThreadLocal

一种解决多线程环境下成员变量的问题的方案，但是与线程同步无关。其思路是为每一个线程创建一个单独的变量副本，从而每个线程都可以独立地改变自己所拥有的变量副本，而不会影响其他线程所对应的副本

ThreadLocal 不是用于解决共享变量的问题的，也不是为了协调线程同步而存在，而是为了方便每个线程处理自己的状态而引入的一个机制

四个方法

get()：返回此线程局部变量的当前线程副本中的值

initialValue()：返回此线程局部变量的当前线程的“初始值”