并发编程实战

如何解决

其他方法里面有线程sleep，也可能有较慢的I/O操作会影响性能

其他类的方法可能也会加锁，可能导致死锁

队列不空

队列不满

init

up

down

class ObjPool<T, R> {
 final List<T> pool;
 // 用信号量实现限流器
 final Semaphore sem;
 // 构造函数
 ObjPool(int size, T t){
 pool = new Vector<T>(){};
 for(int i=0; i<size; i++){
 pool.add(t);
 }
 sem = new Semaphore(size);
 }
 // 利用对象池的对象，调用 func
 R exec(Function<T,R> func) {
 T t = null;
 sem.acquire();
 try {
 t = pool.remove(0);
 return func.apply(t);
 } finally {
 pool.add(t);
 sem.release();
 }
 }
 }
 // 创建对象池
 ObjPool<Long, String> pool = 
 new ObjPool<Long, String>(10, 2);
 // 通过对象池获取 t，之后执行 
 pool.exec(t -> {
 System.out.println(t);
 return t.toString();
 });

class CachedData {
 Object data;
 volatile boolean cacheValid;
 final ReadWriteLock rwl =
 new ReentrantReadWriteLock();
 // 读锁 
 final Lock r = rwl.readLock();
 // 写锁
 final Lock w = rwl.writeLock();
 
 void processCachedData() {
 // 获取读锁
 r.lock();
 if (!cacheValid) {
 // 释放读锁，因为不允许读锁的升级
 r.unlock();
 // 获取写锁
 w.lock();
 try {
 // 再次检查状态 
 if (!cacheValid) {
 data = ...
 cacheValid = true;
 }
 // 释放写锁前，降级为读锁
 // 降级是可以的
 r.lock(); ①
 } finally {
 // 释放写锁
 w.unlock(); 
 }
 }
 // 此处仍然持有读锁
 try {use(data);} 
 finally {r.unlock();}
 }
 }

final StampedLock sl = 
 new StampedLock();
 
 // 获取 / 释放悲观读锁示意代码
 long stamp = sl.readLock();
 try {
 // 省略业务相关代码
 } finally {
 sl.unlockRead(stamp);
 }
 
 // 获取 / 释放写锁示意代码
 long stamp = sl.writeLock();
 try {
 // 省略业务相关代码
 } finally {
 sl.unlockWrite(stamp);
 }

无锁状态

class Point {
 private int x, y;
 final StampedLock sl = 
 new StampedLock();
 // 计算到原点的距离 
 int distanceFromOrigin() {
 // 乐观读
 long stamp = 
 sl.tryOptimisticRead();
 // 读入局部变量，
 // 读的过程数据可能被修改
 int curX = x, curY = y;
 // 判断执行读操作期间，
 // 是否存在写操作，如果存在，
 // 则 sl.validate 返回 false
 if (!sl.validate(stamp)){
 // 升级为悲观读锁
 stamp = sl.readLock();
 try {
 curX = x;
 curY = y;
 } finally {
 // 释放悲观读锁
 sl.unlockRead(stamp);
 }
 }
 return Math.sqrt(
 curX * curX + curY * curY);
 }
 }

final StampedLock sl = 
 new StampedLock();
 
 // 乐观读
 long stamp = 
 sl.tryOptimisticRead();
 // 读入方法局部变量
 ......
 // 校验 stamp
 if (!sl.validate(stamp)){
 // 升级为悲观读锁
 stamp = sl.readLock();
 try {
 // 读入方法局部变量
 .....
 } finally {
 // 释放悲观读锁
 sl.unlockRead(stamp);
 }
 }
 // 使用方法局部变量执行业务操作
 ......

long stamp = sl.writeLock();
 try {
 // 写共享变量
 ......
 } finally {
 sl.unlockWrite(stamp);
 }

while(存在未对账订单){
 // 查询未对账订单
 pos = getPOrders();
 // 查询派送单
 dos = getDOrders();
 // 执行对账操作
 diff = check(pos, dos);
 // 差异写入差异库
 save(diff);
 } 

while(存在未对账订单){
 // 查询未对账订单
 Thread T1 = new Thread(()->{
 pos = getPOrders();
 });
 T1.start();
 // 查询派送单
 Thread T2 = new Thread(()->{
 dos = getDOrders();
 });
 T2.start();
 // 等待 T1、T2 结束
 T1.join();
 T2.join();
 // 执行对账操作
 diff = check(pos, dos);
 // 差异写入差异库
 save(diff);
 } 

// 创建 2 个线程的线程池
 Executor executor = 
 Executors.newFixedThreadPool(2);
 while(存在未对账订单){
 // 查询未对账订单
 executor.execute(()-> {
 pos = getPOrders();
 });
 // 查询派送单
 executor.execute(()-> {
 dos = getDOrders();
 });
 
 /* ？？如何实现等待？？*/
 
 // 执行对账操作
 diff = check(pos, dos);
 // 差异写入差异库
 save(diff);
 } 

// 创建 2 个线程的线程池
 Executor executor = 
 Executors.newFixedThreadPool(2);
 while(存在未对账订单){
 // 计数器初始化为 2
 CountDownLatch latch = 
 new CountDownLatch(2);
 // 查询未对账订单
 executor.execute(()-> {
 pos = getPOrders();
 latch.countDown();
 });
 // 查询派送单
 executor.execute(()-> {
 dos = getDOrders();
 latch.countDown();
 });
 
 // 等待两个查询操作结束
 latch.await();
 
 // 执行对账操作
 diff = check(pos, dos);
 // 差异写入差异库
 save(diff);
 }

// 订单队列
 Vector<P> pos;
 // 派送单队列
 Vector<D> dos;
 // 执行回调的线程池 
 Executor executor = 
 Executors.newFixedThreadPool(1);
 final CyclicBarrier barrier =
 new CyclicBarrier(2, ()->{
 executor.execute(()->check());
 });
 
 void check(){
 P p = pos.remove(0);
 D d = dos.remove(0);
 // 执行对账操作
 diff = check(p, d);
 // 差异写入差异库
 save(diff);
 }
 
 void checkAll(){
 // 循环查询订单库
 Thread T1 = new Thread(()->{
 while(存在未对账订单){
 // 查询订单库
 pos.add(getPOrders());
 // 等待
 barrier.await();
 }
 });
 T1.start(); 
 // 循环查询运单库
 Thread T2 = new Thread(()->{
 while(存在未对账订单){
 // 查询运单库
 dos.add(getDOrders());
 // 等待
 barrier.await();
 }
 });
 T2.start();
 }

Stack

HashTable

Vector

CopyOnWriteArrayList

ConcurrentHashMap

ConcurrentSkipListMap

ConcurrentSkipListSet

CopyOnWriteArraySet

BlockingDeque

BlockingQueue

ConcurrentLinkedQueue

ConcurrentLinkedDeque

分类

CallerRunsPolicy若工作队列已满，则提交任务的线程自己执行该任务

AbortPolicy丢弃任务，并抛出 RejectedException,默认策略

DiscardPolicy直接丢弃任务，不抛出异常

DiscardOldestPolicy丢弃最老的任务

程序的顺序性规则

volatile 变量规则

传递性

管程中锁的规则

线程 start() 规则

线程 join() 规则

竞态条件，指的是程序的执行结果依赖线程执行的顺序

相互谦让

随机等待可解决

保证资源充足

公平地分配资源

避免持有锁的线程长时间执行

吞吐量

延迟

并发量

将共享变量及其对共享变量的操作统一封装起来

每个条件变量都对应有一个等待队列

有限使用成熟的工具类

迫不得已时才使用低级的同步原语

避免过早优化