并发编程
2022-03-02 21:18:04 2 举报
AI智能生成
并发编程
作者其他创作
大纲/内容
JMM内存模型
定义主存(物理内存)和线程工作内存(CPU缓存),底层对应CPU寄存器、缓存、硬件内存、CPU指令优化等
可见性
保证指令执行不受CPU缓存影响
JMM Volatile 缓存一致性协议
多线程程序因为可见性问题可能会导致其他线程读取到了旧数据(脏数据)
原子性
保证不受线程上下文切换影响
锁 - 锁具有排它性,保证共享变量在某一时刻只能被一个线程访问
CAS(Compare and Swap)指令 - CAS指令直接在硬件(处理器和内存)层次上实现,看作是硬件锁
有序性
保证不受CPU指令并行优化影响
指令重排
JIT编译器、处理器可能执行指令重排序,程序顺序与执行顺序不一样
javac编译器一般不会执行指令重排序
指令重排序的对象是指令
指令重排不会对单线程程序的结果正确性产生影响,可能导致多线程程序出现非预期的结果
存储子系统重排序
由高速缓存(CPU Cache)、写缓冲器(Store buffer, Write buffer)引起的, 感知顺序与执行顺序不一致
重排序对象是内存操作的结果,没有真正的对指令执行顺序进行调整,而是造成一种指令执行顺序被调整的现象
LoadLoad、StoreStore、LoadStore、StoreLoad
内存重排序与具体的处理器微架构有关,不同架构的处理器所允 许的内存重排序不同
两个操作指令之间存在数据依赖关系(Data dependency)不允许重排序
synchronized关键字实现有序性
volatile 可实现有序性
volatile
内存屏障(Memary Fence)
写屏障
对volatile变量的写指令后加入写屏障
保证写屏障之前的对共享变量的改动同步到主存中 - 可见性
保证写屏障之前的代码不会重排到屏障之后 - 有序性
读屏障
对volatile变量的读指令前加入读屏障
保证读屏障之后的对共享变量的读取从主存中最新数据- 可见性
保证读屏障之后的代码不会重排到屏障之前- 有序性
不能解决原子性
happens-beofre
as-if-serial
线程
工具
jps
jstack <PID>
jconsole
创建线程
Thread - new Thread(){run(){}} : 运行Thread子类run()方法
Runable - new Thread(new Runable(){run()}) : 运行Runable实现类run()方法
FutureTask - 实现Runable,在run方法中运行Callable接口实现类call()方法, 通过Future接口get()方法阻塞获取执行结果
状态
NEW - 新建线程对象且未启动
RUNNABLE
RUNNING:线程正在执行
Thread.yield()方法可能把线程由RUNNING状态转换为READY状态
READY:线程可以被线程调度器进行调度使它处于RUNNING状态
BLOCKED
线程发起阻塞I/O操作或者申请由其他线程占用的独占资源时线程会转换为BLOCKED阻塞状态
阻塞状态线程不占用CPU资源. 当阻塞I/O操作执行完或线程获得其申请的资源时线程可以转换为RUNNABLE
WAITING
object.wait()方法将线程转为WAITING状态(线程时入等待队列以待被换醒),通过object.notify()、object.notifyAll()换醒至锁池等待竞争锁(线程状态转为BLOCKED)
thread.join()方法将当前线程(非thread)转为WAITING状态,待thread执行完成后被唤醒
TIMED_WAITING
object.wait(times)、thread.join(times)方法将线程转为TIMED_WAITING状态
thread.join(times)方法将当前线程(非thread)转为TIMED_WAITING状态
Thread.sleep(times)方法将当前线程转为TIMED_WAITING状态,通过Thread.interrupt()方法换醒将线程转为RUNNABLE
TERMINATED
方法
thread.start()
thread.run()
Thread.sleep()
thread.interrupt()
thread.isInterrupted()
Thread.interrupted()
Thread.yield()
thread.join()
thread.stop()
obj.wait()、obj.notify()、obj.notifyAll()
LockSupport.park()、LockSupport.unPark()
并发同步处理
Synchronized
原理
膨胀升级
锁分类
乐观锁(无锁)、悲观锁(有锁)
Synchroniced 悲观锁
CAS + volatile 乐观锁 cas自旋实现
重入锁(可多次加锁)、不可重入锁(只能加一次锁)
ReentrantLock 可重入锁
公平锁(线程排队)、非公平锁(线程竞争)
ReentrantLock.FairSync 公平锁
ReentrantLock.NonFairSync 非公平锁
AbstractQueuedSynchronizer(AQS)
1. 相关阻塞锁和同步器工具框架
2. state变量定义资源状态,控制锁的获取与释放;
3. FIFO队列为线程等待队列,类似Monitor的EntryList
子类实现tryAcquire()、tryRelease()等方法实现锁
ReentrantLock
自定义锁
Condition 条件队列 同步队列
ReentrantLock
借助AQS实现
可重入锁
ReentrantLock.FairSync 公平锁 - 新线程进队
ReentrantLock.NonFairSync 非公平锁 - 新线程与队列第二个节点(由head哨兵节点唤醒)线程竞争
newCondition() 支持条件队列
AQS之独占锁实现
ReentrantReadWriteLock 读写锁
实现ReadWriteLock接口
rrwlock.readLock().lock()/unlock() - 不支持条件变量
rrwlock.writeLock().lock()/unlock()
r-r锁可并发
AQS之共享锁实现
w-r/w-w锁互斥
r->w锁重入升级会导致写锁永久等待
w->r锁重入可降级,即写锁释放前加读锁可避免写锁释放后被其它写锁干扰
应用 - 单机缓存读写之线程安全实现
StampedLock 读写锁(Jdk1.8)
long stamp =new StampedLock().readLock().lock()/unlock(stamp)
long stamp = new StampedLock().writeLock().lock()/unlock(stamp)
锁优化: 乐观读(r-r并发)
1. long stamp = new StampedLock().tryOptimisticRead()
2. xxx其它操作
3. lock.validate(stamp) 验戳
成功 -> 读取数据
失败(它线程获得写锁)-> 可锁升级后再读取数据 lock.readLock().lock()
不支持条件变量、不支持锁重入
J.O.C
Semaphore 限流器
限制线程竞争有限资源
new Semaphore(信号量N).require()/release()
CountDownLatch
减计数、为0时释放所有等待的线程、计数为0时,无法重置; countDown()计数减一
new CountDownLatch
cdl.countDown()-递减1
cdl.await()阻塞-等待递减为0后唤醒
应用:多线程执行并行任务并等待完成
CyclicBarrier
加计数方式、计数达到指定值时释放所有等待线程、计数达到指定值时,计数置为0重新开始
new Cycli8cBarrier().await() - 计数加1,若加1后的值不等于构造方法的值,则线程阻塞
Atomic原子操作
CAS
CompareAndSet
CAS操作须使用volatile(共享变量可见性特性)支持
无锁(乐观锁)实现方式之一, 实现无锁并发或无阻塞并发
原子类
原子整数
AtomicInteger、AtomicLong、AtomicBoolean
例: new AtomicInteger(10)
原子引用
AtomicReference(ABA问题)、AtomicMarkableReference、AtomicStampReference
例: new AtomicReference(object)
原子数组
AtomicIntegerArray、AtomicLongArray、AtomicReferenceArray
例: new AtomicIntegerArray(10)
字段更新器
AtomicIntegerFieldUpdater、AtomicLongFieldUpdater、AtomicReferenceFieldUpdater
例: AtomicIntegerFieldUpdater.newUpdater(Xxxx.class, "Integer类型字段名称")
原子累加器
LongAdder: 通过多个累加单元实现快速累加, 效率比AtomicLong高
缓存行伪共享 @sun.misc.Contended
Unsafe
原子类、LockSupport底层实现
不可变类之线程安全
String、DateTimeFormatter
保护性拷贝
享元模式
包装类:Integer、Long、Character、Boolean、Byte、Short ;valueOf()方法
String串池
BigInteger、BigDecimal
无状态类本身也是线程安全的
线程池
ThreadPoolExecutor
Executors.newFixedThreadPool()
指定大小核心线程(可修改),无急救线程
LinkedBlockingQueue 无界缓存等待队列,存放任意数量任务
返回ThreadPoolExecutor线程池对象
适用于任务量已知且任务执行相对耗时的情形
Executors.newCachedThreadPool()
核心线程数为0,最大急救线程数Integer.MAX_VALUE;急救线程空闲60s释放
SynchronousQueue 无缓冲等待队列,队列不存储任务元素直接将任务交给消费者
返回ThreadPoolExecutor线程池对象
适用于任务量密集且任务执行时间短的情形
Executors.newSingleThreadExecutor()
一个核心线程,且不会释放,线程异常结束会重新创建
LinkedBlockingQueue 无界缓存等待队列
返回FinalizableDelegatedExecutorService线程池对象,只暴露ExecutorService接口方法
适用于串行执行任务的情形
提交任务方法
void execute(Runnable task)
无返回
<T> Future<T> submit(Callable<T> task)
返回Future对象,可获取执行结果
<T> List<Future<T>> invokeAll(Collections<? extends Callable<T>> tasks)
返回Future对象,可获取执行结果
可携带超时设置,超时任务取消
<T> T invokeAny(Collections<? extends Callable<T>> tasks)
返回第一个执行完毕的任务结果,其它任务取消
定时任务调度
Timer实现定时任务调度
只一个线程运行,若任务异常结束则会影响后续任务调度
ScheduledExecutorService
Executors.newScheduledThreadPool(corePoolSize)
DelayQueue 延时无界队列
方法
schedule(Runnable/Callable task,delay,TimeUnit) - 延时
scheduleAtFixedRate(Runnable task, delayTime, period, TimeUnit) - 固定速率定时执行任务
scheduleWithFixedDelay(Runnable task, delayTime, period, TimeUnit)- 固定间隔定时执行任务
Futrue/Callable
FutureTask 封状Callable任务
实现Runnable和Futur两个接口
future.get()方法阻塞获取结果result, run()执行完成后保存结果至result后唤醒阻塞的线程
ForkJoin线程池
原理
分治、任务拆分; 拆分任务fork然后join归并,底层实现工作窃取的算法来提高效率
默认的线程池数量等价于CPU数量
使用
ForkJoinPool
new ForkJoinPool().submit(ForkJoinTask<T> task)
ForkJoinTask
RecursiveAction:没有返回值的任务
RecursiveTask:携带返回值的任务
fork() join()
应用
parallelStream并行流
应用 - tomcat线程池
并发Map、List与Set详解
HashMap与ConcurrentHashMap源码剖析
ArrayList、LinkedList与CopyOnWriteArrayList详解
Set与CopyOnWriteArraySet详解
阻塞队列BlockingQueue
ArrayBlockingQueue 数组有界队列
LinkedBlockingQueue 支持有界队列
ConcurrentLinkedQueue 链表有界队列
PriorityBlockingQueue 优先级排序无界队列
DelayQueue 延时无界队列
SynchronousQueue 无缓冲等待队列
线程安全集合
HashTable、Vector 不推介使用
Collections
Collection.synchronizedCollection()
Collection.synchronizedList()
Collection.synchronizedMap()
Collection.synchronizedSet()
Collection.synchronizedSortedMap()
Collection.synchronizedSortedSet()
JUC安全集合
Blocking
阻塞队列,提供阻塞方法
CopyOnWrite
Concurrent集合
弱一致性
Iterator时弱一致
size() 弱一致
读取弱一致
非安全容器使用fail-fast机制抛出ConcurrentModificationException异常终止遍历
ConcurrentHashMap
使用细粒度锁(jdk1.7使用分段锁Segment数组、1.8使用table头元素做为锁),volatile+ CAS + Synchronized
computeIfAbsent()
computeIfPresent()
HashMap在jdk1.7环境下有多线程死链问题
jdk1.7新加元素在链表头部, 1.8改为尾部
table扩容时产生死链
HashMap在jdk1.8环境下有多线程虽然无死链问题,但扩容时会丢失数据
应用
效率
限流
同步异步
缓存
队列
分治
统筹
0 条评论
下一页