【Java进阶】并发编程

PS：整理自极客时间《Java并发编程》

1. 概述

• 三种性质
  • 可见性：一个线程对共享变量的修改，另外一个线程能马上看到。缓存可致使可见性问题。
  • 原子性：一个或多个CPU执行操做不被中断。线程切换可致使原子性问题。
  • 有序性：编译器优化可能致使指令顺序发生改变。编译器优化可能致使有序性问题。
• 三个问题
  • 安全性问题：线程安全
  • 活跃性问题：死锁、活锁、饥饿
  • 性能问题：
    • 使用无锁结构：TLS，Copy-On-Write，乐观锁；Java的原子类，Disruptor无锁队列
    • 减小锁的持有时间：让锁细粒度。如ConcurrentHashmap；再如读写锁，读无锁写有锁

2. Java内存模型

• volatile
  
  • C语言中的原意：禁用CPU缓存，从内存中读出和写入。
  • Java语言的引伸义：
    • Java会将变量马上写入内存，其余线程读取时直接从内存读（普通变量改变后，何时写入内存是不必定的）
    • 禁止指令重排序
  • 解决问题：
    • 保证可见性
    • 保证有序性
    • 不能保证原子性
• Happens-Before规则（H-B）
  • 程序顺序性规则：前面执行的语句对后面语句可见
  • volatile变量规则：volatile变量的写操做对后续的读操做可见
  • 传递性规则：A H-B B，B H-B C，那么A H-B C
  • 管程中锁的规则：对一个锁的解锁 H-B于 后续对这个锁的加锁

3. 互斥锁sychronized

• 锁对象：非静态this，静态Class，括号Object参数
• 预防死锁：
  • 互斥：不能破坏
  • 占有且等待：同时申请全部资源
  • 不可抢占：sychronized解决不了，Lock能够解决
  • 循环等待：给资源设置id字段，每次都是按顺序申请锁
• 等待通知机制：
  • wait、notify、notifyAll

class Allocator {
  private List<Object> als;
  // 一次性申请全部资源
  synchronized void apply(
    Object from, Object to){
    // 经典写法
    while(als.contains(from) ||
         als.contains(to)){
      try{
        wait();
      }catch(Exception e){
      }   
    } 
    als.add(from);
    als.add(to);  
  }
  // 归还资源
  synchronized void free(
    Object from, Object to){
    als.remove(from);
    als.remove(to);
    notifyAll();
  }
}

 

4. 线程的生命周期

• 通用线程的生命周期：
• Java线程的生命周期：
• 状态流转：
  • RUNNABLE -- BLOCKED：线程获取和等待sychronized隐式锁
    • ps：调用阻塞式API时，不会进入BLOCKED状态，但对于操做系统而言，线程实际上进入了休眠态，只不过JVM不关心。
  • RUNNABLE -- WAITING：
    • Object.wait()
    • Thread.join()
    • LockSupport.park()
  • RUNNABLE -- TIMED-WAITING：调用各类带超时参数的线程方法
  • NEW -- RUNNABLE：Thread.start()
  • RUNNABLE -- TERMINATED：线程运行完毕，有异常抛出，或手动调用线程stop()

6. 线程的性能指标

• 延迟：发出请求到收到响应
• 吞吐量：单位时间内处理的请求数量
• 最佳线程数：
  • CPU密集型：线程数 = CPU核数 + 1
  • IO密集型：线程数 = （IO耗时/CPU耗时 + 1）* CPU核数

7. JDK并发包

• Lock：lock、unlock
  • 互斥锁，和sychronized同样的功能，里面能保证可见性
• Condition：await、signal
  • 条件，相比于sychronized的Object.wait，Condition能够实现多条件唤醒等待机制
• Semaphore：acquire、release
  • 信号量，能够用来实现多个线程访问一个临界区，如实现对象池设计中的限流器
• ReadWriteLock：readLock、writeLock
  • 写锁、读锁，容许多线程读，一个线程写，写锁持有时全部读锁和写锁的获取都阻塞（写锁的获取要等全部读写锁释放）
  • 适用于读多写少的场景
• StampedLock：tryOptimisticRead、validate
  • 写锁、读锁（分悲观读锁、乐观读锁）：
• 线程同步：
  • CountDownLatch：一个线程等待多个线程
    • 初始化 --> countDown（减1） --> await（等待为0）
  • CyclicBarrier：一组线程之间相互等待
    • 初始化 --> 设置回调函数（为0时执行，并返回原始值） -->  await（减1并等待为0）
• 并发容器：
  • List：
    • CopyOnWriteArrayList：适用写少的场景，要容忍可能的读不一致
  • Map：
    • ConcurrentHashMap：分段锁
    • ConcurrentSkipListMap：跳表
  • Set：
    • CopyOnWriteArraySet：同上
    • ConcurrentSkipListSet：同上
  • Queue：
    • 分类：阻塞Blocking、单端Queue、双端Deque
    • 单端阻塞（BlockingQueue）：Array～、Linked～、Sychronized～、LinkedTransfer～、Priority～、Delay～
    • 双端阻塞（BlockingDeque）：Linked～
    • 单端非阻塞（Queue）：ConcurrentLinked～
    • 双端非阻塞（Deque）：ConcurrentLinked～
• 原子类：
  • 无锁方案原理：增长了硬件支持，即CPU的CAS指令
  • ABA问题：有解决ABA问题的需求时，增长一个递增的版本号纬度化解
  • 分类：原子化基本数据类型，原子化引用类型、原子化数组、原子化对象属性更新器、原子化累加器
• Future：
  • Future：cancel、isCanceled、isDone、get
  • FutureTask：实现了Runnable和Future接口
• 强大工具类
  • CompletableFuture：一个强大的异步编程工具类（任务之间有聚合关系），暂时略
  • CompletionService：批量并行任务，暂时略

8. 线程池

• 设计原理：
  • 用生产者消费者模型，线程池是消费者，调用者是生产者。
  • 线程池对象里维护一个阻塞队列，一个已经跑起来的工做线程组ThreadsList
  • ThreadList里面循环从队列中去Runnable任务，并调用run方法

  •  1 // 简化的线程池，仅用来讲明工做原理
     2 class MyThreadPool{
     3   // 利用阻塞队列实现生产者 - 消费者模式
     4   BlockingQueue<Runnable> workQueue;
     5   // 保存内部工做线程
     6   List<WorkerThread> threads 
     7     = new ArrayList<>();
     8   // 构造方法
     9   MyThreadPool(int poolSize, 
    10     BlockingQueue<Runnable> workQueue){
    11     this.workQueue = workQueue;
    12     // 建立工做线程
    13     for(int idx=0; idx<poolSize; idx++){
    14       WorkerThread work = new WorkerThread();
    15       work.start();
    16       threads.add(work);
    17     }
    18   }
    19   // 提交任务
    20   void execute(Runnable command){
    21     workQueue.put(command);
    22   }
    23   // 工做线程负责消费任务，并执行任务
    24   class WorkerThread extends Thread{
    25     public void run() {
    26       // 循环取任务并执行
    27       while(true){ ①
    28         Runnable task = workQueue.take();
    29         task.run();
    30       } 
    31     }
    32   }  
    33 }
    34 
    35 /** 下面是使用示例 **/
    36 // 建立有界阻塞队列
    37 BlockingQueue<Runnable> workQueue = 
    38   new LinkedBlockingQueue<>(2);
    39 // 建立线程池  
    40 MyThreadPool pool = new MyThreadPool(
    41   10, workQueue);
    42 // 提交任务  
    43 pool.execute(()->{
    44     System.out.println("hello");
    45 });

• ThreadPoolExcutor
  • 参数
    • corePoolSize：线程池保有的最小线程数
    • maximumPoolSize：线程池建立的最大线程数
    • keepAliveTime：工做线程多久没收到任务，被认为是闲的
    • workQueue：工做队列
    • threadFactory：经过这个参数自定义如何建立线程
    • handler：任务拒绝策略
      • 默认为AbortPolicy，会抛出RejectedExecutionException，这是个运行时异常，要注意
  • 方法
    • void execute()
    • Future submit(Runnable task | Callable task)

9. 鸟瞰并行任务分类