Java并发学习笔记

日常学习笔记
会的越多，不会的越多
戒浮戒躁，脚踏实地
记录和东哥、小海海、小灿灿一起奋斗的日子

java并发编程实践

01 | 可见性、原子性和有序性问题：并发编程Bug的源头

笔记

• 并发编程的三个问题
  • 原子性 -> 一个操作是不可中断的，要么全部执行成功要么全部执行失败。
    • 指令级别语义：CPU单个指令一定是原子性的。
    • java语言语义：java中一个指令不代表是具备原子性的。java指令是对CPU指令的封装。（1 - n）
  • 有序性 -> 程序按照代码顺序有序执行
    • 编译期优化：在java编译期，JVM认为改变指令顺序不会影响结果的场景中（不违反happens-before），会进行编译期的指令重排
    • CPU指令重排：为了解决MESI协议导致的CPU空闲，引入了指令重排机制。大大提高了CPU的利用率
  • 可见性 -> 当前线程对共享变量的修改，对其它线程立即可见
    • JMM语义：在JMM中线程对共享变量的修改对其它线程立即可见。
    • CPU语义：一个内核对L1/l2缓存的M(modify)操作对其它S(share)该变量的内核可见。
      • 可见性问题的根本来源：指令重排导致的CPU指令乱序执行。最终根源
        • Store Buffere
        • Invalidte Queue

金句

在采用一项技术的同时，一定要清楚它带来的问题是什么，以及如何规避

举个例子，我们为了对系统实施监控，会引入例如pinpoint之类的AMP组件，解决了监控问题
的同时也带来了性能问题，比如对带宽的占用，增加了接口响应的延时等；再比如，微服务
架构是为了解决单体应用灵活性差等问题而出现，同时也带了了架构的复杂度，增加了服务
之间通信，数据隔离等问题。所以，一个技术在解决某个问题的同时可能带来新的问题，这样
我们可能又会为新的问题引入别的技术来处理，这是个不断循环的过程。因此，我们在评估
一项技术的时候，需要充分考虑其负面影响，怎么权衡利弊，实现利益最大化。
CPU --> 三级缓存 --> MESI协议 --> 指令重排

02 | Java内存模型：看Java如何解决可见性和有序性问题

笔记

• java内存模型
  • JMM是JVM兼容不同的CPU架构的基础。为了屏蔽底层硬件的差异，向开发者提供统一的接口，故诞生了JMM
  • JMM只是规范，JMM只是规范，JMM只是规范
  • JVM对JMM的实现，才是常见的堆、栈、方法区等一些耳熟能详的名词
• 可见性、有序性的根本解决方案
  • 程序员：对CPU缓存、编译器等按需禁用缓存以及编译优化
    • 方法：volatile/synchronized/final
      • 以上三种方法是java提供给程序员“按需”禁止缓存及编译优化的手段。
  • JVM：happens-before原则
    在JVM可预见的场景中禁止CPU缓存、编译器优化
    • 程序次序规则
      • 在一个线程中，前面的操作总是对后面操作可见
    • 锁定规则
      • 一个unlock操作先行发生于后面对这个锁的lock操作（先释放，才能加锁）
    • 传递性规则
      • A happens-before B B happens-before C 则 A happens-before C （以前理解不到位）

      class VolatileExample {
        int x = 0;
        volatile boolean v = false;
        public void writer() {
          x = 42;
          v = true;
        }
        public void reader() {
          if (v == true) {
            // 这里x会是多少呢？
          }
        }
      }
      

    • 线程start规则
      • start前的操作，总是对被start的操作可见。它是指主线程 A 启动子线程 B 后，子线程 B 能够看到主线程在启动子线程 B 前的操作。

      Thread B = new Thread(()->{
        // 主线程调用B.start()之前
        // 所有对共享变量的修改，此处皆可见
        // 此例中，var==77
      });
      // 此处对共享变量var修改
      var = 77;
      // 主线程启动子线程
      B.start();
      

    • 线程 join() 规则
      • 这条是关于线程等待的。它是指主线程 A 等待子线程 B 完成（主线程 A 通过调用子线程 B 的 join() 方法实现），
        当子线程 B 完成后（主线程 A 中 join() 方法返回），主线程能够看到子线程的操作。当然所谓的“看到”，指的是对共享变量的操作。

      Thread B = new Thread(()->{
        // 此处对共享变量var修改
        var = 66;
      });
      // 例如此处对共享变量修改，
      // 则这个修改结果对线程B可见
      // 主线程启动子线程
      B.start();
      B.join()
      // 子线程所有对共享变量的修改
      // 在主线程调用B.join()之后皆可见
      // 此例中，var==66
      

理解

• 文中的禁用CPU缓存，更深层次的理解见
  • store buffer
  • Invalidate Queue
• volatile/synchronized/final等是java提供给程序员禁止指令重排和禁用缓存的工具
  • CPU为了提高利用率需要指令重排，但是在一些场景中指令重排会导致一些意想不到的错误。这时候需要程序员来发现问题并给出了解决问题的手段
• 重点理解JMM是一种规范、不能混淆JVM对JMM的实现

03 | 互斥锁（上）：解决原子性问题

笔记

• 原子性问题到底该如何解决呢
  • 原子性问题的源头是线程切换，如果能够禁用线程切换那不就能解决这个问题了吗？而操作系统做线程切换是依赖 CPU 中断的，所以禁止 CPU 发生中断就能够禁止线程切换。
  • 同一时刻只有一个线程执行”这个条件非常重要，我们称之为互斥。如果我们能够保证对共享变量的修改是互斥的，那么，无论是单核 CPU 还是多核 CPU，就都能保证原子性了。
• synchronized
  • synchronized属于重量级锁，性能不高，在锁竞争激烈的场所不建议使用
  • synchronized好处在于简单易用，绝对不会unlock
• 锁和受保护资源的关系
  • 受保护资源和锁之间的关联关系是 N:1 的关系

重点

• synchronized锁膨胀过程
• synchronized 对象头 monitor
• long类型的并发读写问题（long64位 -- 32位操作系统）

金句

单核时代通过控制线程的切花就可以保证原子性，但是在多核时代，单纯的控制线程切换是无法保证原子性的，需要通过锁的互斥来
保证高并发场景下的原子性。

04 | 互斥锁（下）：如何用一把锁保护多个资源

笔记

• 保护没有关联关系的多个资源
  • 用不同的锁对受保护资源进行精细化管理，能够提升性能。这种锁还有个名字，叫细粒度锁
• 保护有关联关系的多个资源
  • 锁能覆盖所有受保护资源
  • 对象锁无法解决这个问题，因为会产生，我家的锁锁住别人家的资源的情况
  • 正确姿势是采用类锁（性能有待优化）

理解

• 以前没考虑过也没遇到过 同一把锁管理多个资源的情况，以后在用锁的场景需要注意。

05 | 一不小心就死锁了，怎么办

笔记

• 死锁的专业定义
  一组线程因为竞争共享资源而陷入互相等待，导致“永久”阻塞的现象。

  class Account {
    private int balance;
    // 转账
    void transfer(Account target, int amt){
      // 锁定转出账户
      synchronized(this){     //①
        // 锁定转入账户
        synchronized(target){ //②
          if (this.balance > amt) {
            this.balance -= amt;
            target.balance += amt;
          }
        }
      }
    } 
  }
  

• 在现实中寻找答案
  我们试想在古代，没有信息化，账户的存在形式真的就是一个账本，而且每个账户都有一个账本，这些账本都统一存放在文件架上。银行柜员在给我们做转账时，要去文件架上把转出账本和转入账本都拿到手，然后做转账
  • 文件架上恰好有转出账本和转入账本，那就同时拿走；
  • 如果文件架上只有转出账本和转入账本之一，那这个柜员就先把文件架上有的账本拿到手，同时等着其他柜员把另外一个账本送回来；
  • 转出账本和转入账本都没有，那这个柜员就等着两个账本都被送回来。
    死锁产生
    同一时刻A柜员拿到了入账账本，B柜员拿到了出账账本，A等待出账账本，B等待入账账本。AB柜员就会陷入“永久”等待。这就是死锁。
• 粗粒度锁
  解决上述问题，可以采用粗粒度锁，也就是类锁，但是类锁带来的是性能问题。
• 细粒度锁
  • 优点：使用细粒度锁可以提高并行度，是性能优化的有效手段。
  • 风险：机会和风险是并存的。细粒度锁可能导致死锁。
• 如何预防死锁
  解决死锁最好的办法是预防死锁，将其扼杀在摇篮里。
  • 产生死锁的四个条件
    • 互斥，共享资源X和Y只能被一个线程占用
    • 占有且等待，线程T1占有X资源，在等待资源Y的同时，不释放资源X。
    • 不可抢占，其他线程不能强行抢占线程T1占有的资源
    • 循环等待，线程T1等待线程T2占有的资源，线程T2等待线程T1占有的资源，就是循环等待
  • 解决死锁的思路就很简单了，就是破坏以上一个条件就不会造成死锁
    • 破坏占用且等待条件：一次性申请所有的资源
      • example：不允许柜员直接在文件架上拿账本，而是增加管理员，柜员拿账本需要通过管理员来审核。比如，A柜员需要拿进账
        管理员发现文件架上没有出账账本，所以不允许柜员只拿进账账本。这样就解决了占用且等待问题。
    • 破坏不可抢占条件：核心是要能够主动释放它占有的资源
      • 这一点 synchronized 是做不到的。因为synchronized一旦申请不到资源就会进入阻塞状态
        进入阻塞态就以为什么也干不了，也释放不了线程占用的资源。
      • ReetrentLock 可以解决这个问题
    • 破坏循环等待条件：对资源排序，然后按序申请资源
      • 我们假设每个账户都有不同的属性 id，这个 id 可以作为排序字段，申请的时候，我们可以按照从小到大的顺序来申请

      class Account {
        private int id;
        private int balance;
        // 转账
        void transfer(Account target, int amt){
          Account left = this;        //①
          Account right = target;    //②
          if (this.id > target.id) { //③
            left = target;           //④
            right = this;            //⑤
          }                          //⑥
          // 锁定序号小的账户
          synchronized(left){
            // 锁定序号大的账户
            synchronized(right){ 
              if (this.balance > amt){
                this.balance -= amt;
                target.balance += amt;
              }
            }
          }
        } 
      }
      

金句

当我们在编程世界里遇到问题时，应不局限于当下，可以换个思路，向现实世界要答案，利用现实世界的模型来构思解决方案，
这样往往能够让我们的方案更容易理解，也更能够看清楚问题的本质。

用细粒度锁来锁定多个资源时，要注意死锁的问题。这个就需要你能把它强化为一个思维定势，遇到这种场景，
马上想到可能存在死锁问题。当你知道风险之后，才有机会谈如何预防和避免，因此，识别出风险很重要

我们在选择具体方案的时候，还需要评估一下操作成本，从中选择一个成本最低的方案。

收获

while(true)循环是不是应该有个timeout，避免一直阻塞下去？
加超时在项目中非常实用。

06 | 用“等待-通知”机制优化循环等待

笔记

• 问题：05中破坏占用且等待条件，while循环会浪费CPU资源

  // 一次性申请转出账户和转入账户，直到成功
  while(!actr.apply(this, target)){...};
  

  当并发冲突增加，可能上述while循环会循环上万次，浪费CPU资源
• 方案：等待-通知机制
  • 05中解决占用且等待条件，其实根本原因在于所有线程都在盲目申请，而不是等到“机会”合适的时候再申请。所谓来得早不如来得巧
  • 使用 synchronized, wait(), notify(), notifyAll()实现等待-通知机制
    • 这个等待队列和互斥锁是一对一的关系，每个互斥锁都有自己独立的等待队列
    • notify() 只能保证在通知时间点，条件是满足的。而被通知线程的执行时间点和通知的时间点基本上不会重合，所以当线程执行的时候，很可能条件已经不满足了（保不齐有其他线程插队）

//单例
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();
  }
}
//测试方法
public class Test{
    public void test(){
        //加锁
        allocator.apply(from, to);
        //TODO ...
        //释放锁
        allocator.free();
    }
}

收获

• 尽量使用notifyAll
  • notify() 是会随机地通知等待队列中的一个线程，而 notifyAll() 会通知等待队列中的所有线程。从感觉上来讲，应该是 notify() 更好一些，因为即便通知所有线程，也只有一个线程能够进入临界区。但那所谓的感觉往往都蕴藏着风险，实际上使用 notify() 也很有风险，它的风险在于可能导致某些线程永远不会被通知到。
    假设我们有资源 A、B、C、D，线程 1 申请到了 AB，线程 2 申请到了 CD，此时线程 3 申请 AB，会进入等待队列（AB 分配给线程 1，线程 3 要求的条件不满足），线程 4 申请 CD 也会进入等待队列。我们再假设之后线程 1 归还了资源 AB，如果使用 notify() 来通知等待队列中的线程，有可能被通知的是线程 4，但线程 4 申请的是 CD，所以此时线程 4 还是会继续等待，而真正该唤醒的线程 3 就再也没有机会被唤醒了。所以除非经过深思熟虑，否则尽量使用 notifyAll()。
    
    notify工作原理图
• 每个互斥锁都有各自独立的等待池
  
• wait和sleep的区别
  wait()方法与sleep()方法的不同之处在于，wait()方法会释放对象的“锁标志”。当调用某一对象的wait()方法后，会使当前线程暂停执行，并将当前线程放入对象等待池中，直到调用了notify()方法后，将从对象等待池中移出任意一个线程并放入锁标志等待池中，只有锁标志等待池中的线程可以获取锁标志，它们随时准备争夺锁的拥有权。当调用了某个对象的notifyAll()方法，会将对象等待池中的所有线程都移动到该对象的锁标志等待池。
  sleep()方法需要指定等待的时间，它可以让当前正在执行的线程在指定的时间内暂停执行，进入阻塞状态，该方法既可以让其他同优先级或者高优先级的线程得到执行的机会，也可以让低优先级的线程得到执行机会。但是sleep()方法不会释放“锁标志”，也就是说如果有synchronized同步块，其他线程仍然不能访问共享数据。
  • wait释放锁
  • sleep不释放锁

07 | 安全性、活跃性以及性能问题

笔记

• 并发编程的问题
  • 微观上：原子性、可见性、有序性
  • 宏观上：安全性、活跃性、性能
• 安全性问题
  • 理论知识
    • 什么是线程安全？本质上就是正确性：程序按照我们的期望执行
    • 理论上线程安全的程序就是要避免原子性、可见性、有序性问题
    • 需要着重的关注线程安全性的场景：存在共享数据并且该数据会发生变化，通俗的讲就是多个线程同时读写同一个数据
  • 数据竞争：当多个线程同时修改一个共享数据时，导致的并发bug（其实就是线程安全性）

  public class Test {
    private long count = 0;
    void add10K() {
      int idx = 0;
      while(idx++ < 10000) {
        count += 1;
      }
    }
  }
  

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

  public class Test {
    private long count = 0;
    synchronized long get(){
      return count；
    }
    synchronized void set(long v){
      count = v;
    } 
    void add10K() {
      int idx = 0;
      while(idx++ < 10000) {
        set(get()+1) //当两个线程同时运行到get()方法时，get()方法先后（有sync锁）0，count结果为1。当先后执行时，count结果为2     
      }
    }
  }
  

  • 解决方案 -- 互斥（锁）
    CPU 提供了相关的互斥指令，操作系统、编程语言也会提供相关的 API。从逻辑上来看，我们可以统一归为：锁
• 活跃性问题
  所谓活跃性问题，指的是某个操作无法执行下去。我们常见的“死锁”就是一种典型的活跃性问题，当然除了死锁外，还有两种情况，分别是“活锁”和“饥饿”
  • 活锁：互相“谦让”的例子。线程因为总是同时的进行竞争而导致的互相等待的现象。
    • 解决方案：让线程等待一个随机的时间。避免“同时”即可
  • 死锁：一组线程线程因为竞争共享数据而陷入永久性等待，导致线程“永久”的阻塞。
    • 解决方案：破坏四个条件即可：互斥、占用且等待、不可抢占、循环等待
  • 饥饿：所谓“饥饿”指的是线程因为无法访问所需资源而无法执行下去的情况
    • 解决方案
      • 公平的分配资源
      • 保证资源充足
      • 避免线程长时间持有锁
        • 公平锁，先来先得
• 性能问题
  • 尽量采用无锁方式 乐观锁(CAS) 本地化存储（ThreadLocal） copy-on-write
  • 尽量减少线程持有锁的时间
  • 优化锁粒度 联想1.8前后ConcurrentHashMap的锁设计

总结

• 安全性方面注意数据竞争 竞态条件问题
• 活跃性方面注意死锁 活锁 饥饿等问题
• 性能方面尽量采用无锁CAS，优化锁粒度，减少锁持有时间

08 | 管程：并发编程的万能钥匙

笔记

• 管程和信号量
  • 管程和信号量是同步的，所谓同步也就是管程能实现信号量，信号量也能实现管程
• 什么是管程
  • 管程是一个概念
  • 在java中每一个对象都绑定着一个管程（信号量）
  • 线程访问加锁对象其实就是去拥有一个监视器的过程。
  • 线程访问共享变量的过程其实就是申请拥有监视器的过程。
  • 监视器至少有两个等待队列。
    总结起来就是，管程就是一个对象监视器。任何线程想要访问该资源，就要排队进入监控范围。进入之后，接受检查，不符合条件，则要继续等待，直到被通知，然后继续进入监视器。
• java中的管程
  • 1.5之前：①synchronized + wait、notify、notifyAll
  • 1.5之后：②lock + condition
  • 区别
    • ①只支持一种一个条件变量，即wait，调用wait时将会将其加入到等待队列。被notify时会随机通知一个线程加入到锁的等待池
    • ②相对①condition支持中断和增加了等待时间
• 三种实现管程的模型
  • HASEN：执行完，再去唤醒另外一个线程。能够保证线程的执行
  • HOARE：是中断当前线程，唤醒另外一个线程，执行玩再去唤醒，也能够保证完成。
  • MESA：是进入等待队列，不一定有机会能够执行（公平竞争公平 == 容易饥饿）

09 | Java线程（上）：Java线程的生命周期

笔记

• 通用的五种线程状态
  • 初始状态
  • 可运行状态
  • 运行状态
  • 休眠状态
  • 终止状态
• java线程的五种状态
  • new(初始化)
  • runnable(可运行/运行状态)
  • blocked(阻塞状态)
  • waiting(无限时等待)
  • timed_waiting(有时限等待)
  • terminated(终止)
• RUNNABLE 与 BLOCKED 的状态转换
  • 只有一种场景会触发这种转换，就是线程等待 synchronized 的隐式锁。synchronized 修饰的方法、代码块同一时刻只允许一个线程执行，其他线程只能等待，这种情况下，等待的线程就会从 RUNNABLE 转换到 BLOCKED 状态。而当等待的线程获得 synchronized 隐式锁时，就又会从 BLOCKED 转换到 RUNNABLE 状态
• RUNNABLE 与 WAITING 的状态转换
  • 第一种场景，获得 synchronized 隐式锁的线程，调用无参数的 Object.wait() 方法
  • 第二种场景，调用无参数的 Thread.join() 方法
  • 第三种场景，调用 LockSupport.park() 方法。其中的 LockSupport 对象，也许你有点陌生，其实 Java 并发包中的锁，都是基于它实现的
• RUNNABLE 与 TIMED_WAITING 的状态转换
  • 调用带超时参数的 Thread.sleep(long millis) 方法
  • 获得 synchronized 隐式锁的线程，调用带超时参数的 Object.wait(long timeout) 方法
  • 调用带超时参数的 Thread.join(long millis) 方法
  • 调用带超时参数的 LockSupport.parkNanos(Object blocker, long deadline) 方法
  • 调用带超时参数的 LockSupport.parkUntil(long deadline) 方法
• 从 NEW 到 RUNNABLE 状态
  • Java 刚创建出来的 Thread 对象就是 NEW 状态，而创建 Thread 对象主要有两种方法
• RUNNABLE 到 TERMINATED 状态
  • 正常结束
  • interrupt()

10 | Java线程（中）：创建多少线程才是合适的？

笔记

• 为什么要使用多线程？

  • 使用多线程，本质上就是提升程序性能。
  • 两个指标
    • 延迟指的是发出请求到收到响应这个过程的时间；延迟越短，意味着程序执行得越快，性能也就越好。
    • 吞吐量指的是在单位时间内能处理请求的数量；吞吐量越大，意味着程序能处理的请求越多，性能也就越好
  • 本质上就是将硬件的性能发挥到极致

• 多线程的应用场景

  • I/O密集型
  • CPU密集型

• 评论区

  • 个人觉得公式话性能问题有些不妥，定性的io密集或者cpu密集很难在定量的维度上反应出性能瓶颈，而且公式上忽略了线程数增加带来的cpu消耗，性能优化还是要定量比较好，这样不会盲目，比如io已经成为了瓶颈，增加线程或许带来不了性能提升，这个时候是不是可以考虑用cpu换取带宽，压缩数据，或者逻辑上少发送一些。最后一个问题，我的答案是大部分应用环境是合理的，老师也说了是积累了一些调优经验后给出的方案，没有特殊需求，初始值我会选大家都在用伪标准

11 | Java线程（下）：为什么局部变量是线程安全的？

笔记

• 为什么局部变量不存在线程安全问题
  • 从线程栈解释：局部变量在线程的独享栈中
  • 没有共享就没有伤害
• 线程安全问题的解决方案之一
  • 线程封闭
  • ThreadLocal
    • 注意ThreadLocal的内存泄漏问题

总结

• new出来的对象都在堆中的合理解释
  • 对象在堆中，但是对象的句柄（引用或者指针）在栈中
• 没有共享就没有伤害
• 递归注意深度，容易导致栈内存溢出
• ThreadLocal的内存泄漏问题
• Spring对数据源连接池的抽象 ThreadLocal实现的

12 | 如何用面向对象思想写好并发程序？

笔记

• 三个思路
  • 封装共享变量
    • 将共享变量作为对象属性封装在内部，对所有公共方法制定并发访问策略
  • 识别共享变量间的约束条件

  public class SafeWM {
    // 库存上限
    private final AtomicLong upper =
          new AtomicLong(0);
    // 库存下限
    private final AtomicLong lower =
          new AtomicLong(0);
    // 设置库存上限
    void setUpper(long v){
      upper.set(v);
    }
    // 设置库存下限
    void setLower(long v){
      lower.set(v);
    }
    // 省略其他业务代码
  }
  

    * 约束条件，决定了并发访问策略
    * 忽略了一个约束条件：下限 < 上限
    * 不安全的一个例子
  

  public class SafeWM {
    // 库存上限
    private final AtomicLong upper =
          new AtomicLong(0);
    // 库存下限
    private final AtomicLong lower =
          new AtomicLong(0);
    // 设置库存上限
    void setUpper(long v){
      // 检查参数合法性
      if (v < lower.get()) {
        throw new IllegalArgumentException();
      }
      upper.set(v);
    }
    // 设置库存下限
    void setLower(long v){
      // 检查参数合法性
      if (v > upper.get()) {
        throw new IllegalArgumentException();
      }
      lower.set(v);
    }
    // 省略其他业务代码
  }
  

    当setUpper(5) 和 setLower(7)同时发生时，会发生upper = 5 lower = 7
    * 着重注意if else语句造成的竞态条件
  

• 制定并发访问策略
  • 避免共享
  • 不变模式（不可变对象/变量）
  • 管程/并发工具(JUC)
    • 优先考虑java的并发包，一般能解决绝大多数并发问题
    • 迫不得已时再考虑“低级”原语：synchronized、Lock、Semaphore，使用时千万小心
    • 避免过早优化，首先保证安全，等到确实遇到性能瓶颈的时候，再考虑优化

14 | Lock和Condition（上）：隐藏在并发包中的管程

笔记

• 并发编程的两个核心问题
  • 互斥-同一个时刻只能有一个线程可以访问共享资源
  • 同步-线程之间如何通信、协作
• Java中管程的两个实现--java线程的两种协作方式
  • synchronized + wait + notify
    • synchronized实现互斥
    • wait + notify实现同步
  • lock + Condition
    • lock实现互斥
    • condition实现同步
• 再造管程的理由
  • synchronized存在的问题
    • 1.5之前性能不够好 + 容易膨胀为重量级锁
    • 死锁问题无法破坏不可抢占条件
• 设计一个锁能解决不可抢占条件
  • 能够响应中断
  • 支持超时
  • 非阻塞的获取锁

      // 支持中断的API
      void lockInterruptibly() 
        throws InterruptedException;
      // 支持超时的API
      boolean tryLock(long time, TimeUnit unit) 
        throws InterruptedException;
      // 支持非阻塞获取锁的API
      boolean tryLock();
  

• 如何保证可见性
  • synchronized
    • happens-before中有一个锁规则，保证了synchronized的可见性
  • Lock
    • 利用了 volatile 相关的 Happens-Before 规则
      ReentrantLock 内部持有一个volatile的变量

    class SampleLock {
      volatile int state;
      // 加锁
      lock() {
        // 省略代码无数
        state = 1;
      }
      // 解锁
      unlock() {
        // 省略代码无数
        state = 0;
      }
    }
    

• 可重入锁
  • 线程可以重复获取同一把锁
  • ReentrantLock汉语意思就是可重入锁的含义
• 公平锁与非公平锁
  • 公平锁：按照先来先得的原则，完全公平。其实就是排队等待
  • 非公平锁：公平竞争锁，每次竞争所有线程获取锁的机会是均等待。（为什么叫非公平锁呢？因为运气不好的可能造成线程饥饿）

15 | Lock和Condition（下）：Dubbo如何用管程实现异步转同步？

笔记

• 相对synchronized + wait + notify/ReentrantLock + Condition的优势
  • Lock&Condition 实现的管程是支持多个条件变量的，这是二者的一个重要区别。
  • sync + wait只能支持一个条件，因为条件都是绑定到monitor上的，每一个锁只有一个monitor
• 如何实现一个阻塞队列
  • 阻塞队列需要两个条件：满阻塞/空阻塞
    • sync管程只能实现一个阻塞，因为其只能支持一个条件变量
    • lock + Condition 可以支持多个条件变量
  • 复习
    • sync + wait + notify + notifyAll
    • lock + Condition + await + signal + signalAll

    public class BlockedQueue<T>{
      final Lock lock =
        new ReentrantLock();
      // 条件变量：队列不满  
      final Condition notFull =
        lock.newCondition();
      // 条件变量：队列不空  
      final Condition notEmpty =
        lock.newCondition();
    
      // 入队
      void enq(T x) {
        lock.lock();
        try {
          while (队列已满){
            // 等待队列不满
            notFull.await();
          }  
          // 省略入队操作...
          //入队后,通知可出队
          notEmpty.signal();
        }finally {
          lock.unlock();
        }
      }
      // 出队
      void deq(){
        lock.lock();
        try {
          while (队列已空){
            // 等待队列不空
            notEmpty.await();
          }  
          // 省略出队操作...
          //出队后，通知可入队
          notFull.signal();
        }finally {
          lock.unlock();
        }  
      }
    }
    

• Dubbo如何实现异步的RPC实现同步的等待结果

  // 创建锁与条件变量
  private final Lock lock 
      = new ReentrantLock();
  private final Condition done 
      = lock.newCondition();
  
  // 调用方通过该方法等待结果
  Object get(int timeout){
    long start = System.nanoTime();
    lock.lock();
    try {
    while (!isDone()) {
      done.await(timeout);
        long cur=System.nanoTime();
      if (isDone() || 
            cur-start > timeout){
        break;
      }
    }
    } finally {
    lock.unlock();
    }
    if (!isDone()) {
    throw new TimeoutException();
    }
    return returnFromResponse();
  }
  // RPC结果是否已经返回
  boolean isDone() {
    return response != null;
  }
  // RPC结果返回时调用该方法   
  private void doReceived(Response res) {
    lock.lock();
    try {
      response = res;
      if (done != null) {
        done.signal();
      }
    } finally {
      lock.unlock();
    }
  }
  

猜想

• Future应该也是通过 Lock + Condition实现的
• 明天看源码ArrayListBlockQueue/LinkedListBlockQueue，看二者如何实现的阻塞

16 | Semaphore：如何快速实现一个限流器？

笔记

• 信号量
  • 互斥性
    • Semaphore如何实现互斥。指定Semaphore的计数器为1，也就意味着同一个时刻只能有一个线程可以访问临界区资源
  • 线程并行控制
    • Semaphore如何控制并发。Semaphore通过计数器，控制访问临界区的线程不能超过计数器值。
• 操作系统中也存在信号量--作用和java中的信号量也是相同的
  • 操作系统利用信号量控制进程的并行
• Semaphore 的公平性
  • 默认Semaphore是非公平的，同 ReentrantLock
  • Semaphore提供了两个构造方法，如下所示，两个参数的构造方法，第二个参数可以指定公平性
    • false：非公平，也就是公平竞争，容易饥饿
    • true：公平，先来后到

      public Semaphore(int permits) {
          sync = new NonfairSync(permits);
      }
      public Semaphore(int permits, boolean fair) {
          sync = fair ? new FairSync(permits) : new NonfairSync(permits);
      }
  

总结

• 信号量在 Java 语言里面名气并不算大，但是在其他语言里却是很有知名度的。Java 在并发编程领域走的很快，
  重点支持的还是管程模型。 管程模型理论上解决了信号量模型的一些不足，主要体现在易用性和工程化方面，
  例如用信号量解决我们曾经提到过的阻塞队列问题，就比管程模型麻烦很多。

17 | ReadWriteLock：如何快速实现一个完备的缓存？

笔记

• 管程和信号量都能解决所有并发问题了，JUC中还存在那么多并发工具？
  • 分场景优化，提升易用性
• 什么是读写锁
  • 读写锁普遍存在于各种语言，三条基本原则
    • 允许多个线程同时读共享变量
    • 统一时刻只允许一个线程写共享变量
    • 如果一个写线程正在执行写操作，此时禁止读线程读共享变量（读锁和写锁是互斥的）
      • 对比mysql（以下两条结论都建立在两个独立的事务中）
        • 在read-uncommitted、read-committed、repeatable-read级别下，对同一行数据的写不会阻塞读。原因在于在以上三个隔离级别中，是通过MVCC控制的。当然如果采用当前读（lock in share mode读读不互斥，读写互斥。ps：for update读读互斥、读写互斥）则可以产生阻塞
        • 在serializable隔离级别下，同读写锁的第三条读写互斥规则。原因是在serializable级别下所有的读都是当前读（互斥读）。
        • mysql 演示

        -- session1
        -- 查询事务级别
        select @@tx_isolation;
        -- 设置事务级别
        set session transaction isolation level read committed;
        -- 开启事务
        start transaction;
        select * from sys_test where id = 2 ; -- 加上lock in share mode同读写锁的读写互斥;
        -- 提交事务
        commit;
        
        -- session2
        set session transaction isolation level read committed;
        start transaction;
        update sys_test set name = '55' where id = 2;
        select * from sys_test where id = 2;
        commit;
        

• ReadWriteLock读写锁
  • 读多写少的场景
  • 缓存
• 读写锁的升级与降级
  • ReadWriteLock不支持锁升级（会饥饿），但是支持锁降级。

理解

• ReadWriteLock读写锁如果不互斥，也就没必要存在读锁了。
  • 类似mysql，如果读写不互斥，则没必要加读锁。
  • 读锁存在的意义在于第三条规则。写同时阻塞读，可以保证读到的一定是最新的。
  • mysql在前三个隔离级别下默认读是快照读（无锁读），所以才存在了脏读、不可重复度、幻读。所以解决以上三个问题的终极方法就是所有读都采用当前读（锁读）、当然这会影响性能，不建议使用。

18 | StampedLock：有没有比读写锁更快的锁？

笔记

• StampedLock 和 ReadWriteLock的区别
  • ReadWriteLock 支持两种模式（读读不互斥，写写互斥，读写互斥）
    • 读锁
    • 写锁
  • StampedLock 支持三种模式
    • 写锁 语义同ReadWriteLock的写锁
    • 悲观读锁 语义同ReadWriteLock的读锁
    • 乐观读锁 乐观锁-无锁，性能更好
• StampedLock 锁升级 （不是内部实现）
  • 当 StampedLock 乐观读期间遇到写操作（validate(stamp)方法可判断）
• 注意事项
  • StampedLock 是不可重入的
  • 使用 StampedLock 一定不要调用中断操作，如果需要支持中断功能，一定使用可中断的悲观读锁 readLockInterruptibly() 和写锁 writeLockInterruptibly()。这个规则一定要记清楚。

范式

• 读范式

  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);
  }
  

19 | CountDownLatch和CyclicBarrier：如何让多线程步调一致？

笔记

• CountDownLatch
  • 主要用来解决一个线程等待多个线程的场景。可以类比旅游团团长要等待所有的游客到齐才能去下一个景点；
  • 一旦计数器到0，再有线程调用await()，会直接通过。
• CyclicBarrier
  • 一组线程之间互相等待。
  • 具备自动重置功能。CyclicBarrier 的计数器是可以循环利用的。一旦计数器减到 0 会自动重置到你设置的初始值。
  • CyclicBarrier 还可以设置回调函数

TODO

• CyclicBarrier实操

20 | 并发容器：都有哪些“坑”需要我们填？

笔记

• List
  • LinkedList
  • ArrayList
  • 同步容器
    • Vector
  • 并发容器
    • CopyOnWriteArrayList
• Set
  • HashSet
  • TreeSet
  • LinkedSet
  • 并发容器
    • CopyOnWriteArraySet
    • CopyOnWriteSkipListSet
• Map
  • LinkedHashMap
  • HashMap
  • TreeMap
  • 同步容器
    • HashTable
  • 并发容器
    • ConcurrentHashMap
    • ConcurrentSkipListMap
• Queue
  • 非阻塞
    • 线程不安全
      • PriorityQueue
      • LinkedList
    • 线程安全
      • 单端
        • ConcurrentLinkedQueue
      • 双端
        • ConcurrentLinkedDeque
  • 阻塞
    • ArrayBlockingQueue
      • 出队入队同一把锁
      • 底层数据结构：数组
      • 有界
      • 默认不保证线程安全性
    • LinkedBlockingQueue
      • 底层链表
      • “有界”阻塞队列（长度为int长度）
    • SynchronousQueue
      • 无空间（不存储元素）
    • LinkedTransferQueue
      • 无界：由链表组成的无界TransferQueue
    • PriorityBlockingQueue
      • 支持优先级
      • 无界
    • DelayQueue
      • 延时阻塞队列
      • 无界
• 对于Collections.synchronizedXXX()的方法要着重注意竞态条件问题
• 使用无界队列时要着重注意oom问题。例如：线程池的阻塞队列

21 | 原子类：无锁工具类的典范

笔记

• CAS全程

  • Compare And Swap

• 无锁方案的优点

  • 无锁方案相对于互斥锁方案，最大的好处就是性能
    • 互斥锁方案为了保证互斥性，需要执行加锁、解锁操作，而加锁、解锁操作本身就消耗性能；
      同时拿不到锁的线程还会进入阻塞状态，进而触发线程切换，线程切换对性能的消耗也很大。
      相比之下，无锁方案则完全没有加锁、解锁的性能消耗，同时还能保证互斥性，既解决了问题，又没有带来新的问题，可谓绝佳方案。
  • 不会出现死锁
  • 注意活锁和饥饿问题

• 无锁方案的实现原理

  • 硬件支持-CPU为了解决并发问题，提供了CAS指令。作为一条 CPU 指令，CAS 指令本身是能够保证原子性的。
  • CAS的使用一般都伴随着自旋

• ABA问题

  • 解决方案
    • 理论上增加时间戳或者版本号都可以实现，JDK采用的是使用版本号。

• 单纯的累加操作，使用累加器相对原子化的基本类型，性能更高

• TODO实操

22 | Executor与线程池：如何创建正确的线程池？

• 线程池是一种生产者 - 消费者模式
  • 线程池的使用方式 是 生产者
  • 线程池本身是消费者
• ThreadPoolExecutor

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

  • corePoolSize 核心池大小
    表示线程池保有的最小线程数。有些项目很闲，但是也不能把人都撤了，至少要留 corePoolSize 个人坚守阵地。
    • 当线程数小于核心线程数时，即使有线程空闲，线程池也会优先创建新线程处理
    • 核心线程会一直存活，即使没有任务需要执行
    • 设置allowCoreThreadTimeout=true（默认false）时，核心线程会超时关闭
  • maximumPoolSize 最大池
    表示线程池创建的最大线程数。当项目很忙时，就需要加人，但是也不能无限制地加，最多就加到 maximumPoolSize 个人。当项目闲下来时，就要撤人了，最多能撤到 corePoolSize 个人。
  • keepAliveTime & unit 空闲线程存活的时间
    上面提到项目根据忙闲来增减人员，那在编程世界里，如何定义忙和闲呢？很简单，一个线程如果在一段时间内，都没有执行任务，说明很闲，keepAliveTime 和 unit 就是用来定义这个“一段时间”的参数。也就是说，如果一个线程空闲了keepAliveTime & unit这么久，而且线程池的线程数大于 corePoolSize ，那么这个空闲的线程就要被回收了。
  • workQueue 工作队列
    工作队列，当线程核心池已满，接下来来的任务就会被放到工作队列（阻塞队列）
    • 最好不要用无界队列，一旦业务量增加很容易OOM
  • threadFactory
    通过这个参数你可以自定义如何创建线程，例如你可以给线程指定一个有意义的名字。（PS：一般给线程池名字加前缀用这个方法）
  • handler 拒绝策略
    当工作队列和线程池都满了，那么此时提交任务，线程池就会拒绝接收
    • CallerRunsPolicy：提交任务的线程自己去执行该任务。
    • AbortPolicy：默认的拒绝策略，会 throws RejectedExecutionException
    • DiscardPolicy：直接丢弃任务，没有任何异常抛出
    • DiscardOldestPolicy：丢弃最老的任务，其实就是把最早进入工作队列的任务丢弃，然后把新任务加入到工作队列。
    • 实现 RejectedExecutionHandler 接口 自定义拒绝策略，如果处理的任务不允许丢失，则可以和降级策略配合使用。
      可以放数据库，放mq,redis，本地文件都可以，具体要看实际需求。

重点

• 注意事项
  • 不要使用无界工作队列 最好不要使用 无界队列 因为业务量的突然增加很容易导致OOM
  • 默认拒绝策略要慎重使用 实际开发任务中可以配合MQ和服务降级处理
  • 线程池的异常处理 execute() 方法提交任务时，如果任务在执行的过程中出现运行时异常，会导致执行任务的线程终止；不过，最致命的是任务虽然异常了，但是你却获取不到任何通知，这会让你误以为任务都执行得很正常。
    最好的办法还是捕获所有异常，按需处理
  • 线程池的使用建议
    • 业务隔离
    • 压测确定队列长度或线程数
  • 核心池大小
    • 经验值
      • IO密集型
        • 2 * CPU + 1
      • CPU密集型
        • CPU + 1
    • 实践值
      • 最佳线程数目 = （线程等待时间与线程CPU时间之比 + 1）* CPU数目
    • 最佳值
      • 压测
• 线程池流程
  

23 | Future：如何用多线程实现最优的“烧水泡茶”程序？

笔记

• 如何获取异步执行结果
  • Future
  • CountdownLatch
  • join方法
• Future接口是一个获取异步结果的通用接口
• FutureTask是如何获取结果的
  • FutureTask源码中状态标识 state
    • NEW = 0;初始状态
    • COMPLETING = 1;正在执行
    • NORMAL = 2;
    • EXCEPTIONAL = 3;
    • CANCELLED = 4;
    • INTERRUPTING = 5;
    • INTERRUPTED = 6;
  • 状态转换 TODO详细阅读源码理解
    • NEW -> COMPLETING -> NORMAL
    • NEW -> COMPLETING -> EXCEPTIONAL
    • NEW -> CANCELLED
    • NEW -> INTERRUPTING -> INTERRUPTED
• FutureTask的阻塞方法不像有些博客说的 Object#wait而是 LockSupport.park(this);对应的唤醒线程的方法 LockSupport.unpark(this);

TODO

• 总结java中实现获取异步结果的方式和工具
• 总结 异步转同步的方式

24 | CompletableFuture：异步编程没那么难

笔记

• CompletableFuture 的核心优势
  • 无需手工维护线程，没有繁琐的手工维护线程的工作，给任务分配线程的工作也不需要我们关注；(对比FutureTask的实现)
  • 语义更清晰，例如 f3 = f1.thenCombine(f2, ()->{}) 能够清晰地表述“任务 3 要等待任务 1 和任务 2 都完成后才能开始”；
  • 代码更简练并且专注于业务逻辑，几乎所有代码都是业务逻辑相关的。
• 创建CompletableFuture对象
  • runAsync(Runnable runnable) 不获取返回值的静态方法
  • supplyAsync(Supplier supplier) 可获取返回值的方法 （ps：作用同Future），Supplier相对于Runnable，get方法可以获取返回值
  • 以上两个方法可以指定线程池 （PS：CompletableFuture默认使用ForkJoinPool线程池）

  //使用默认线程池
  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)  
  

• CompletableFuture实现的CompletionStage接口的作用
  任务的时序关系管理
  • 串行关系
    

    CompletionStage<R> thenApply(fn);
    CompletionStage<R> thenApplyAsync(fn);
    CompletionStage<Void> thenAccept(consumer);
    CompletionStage<Void> thenAcceptAsync(consumer);
    CompletionStage<Void> thenRun(action);
    CompletionStage<Void> thenRunAsync(action);
    CompletionStage<R> thenCompose(fn);
    CompletionStage<R> thenComposeAsync(fn);
    

  • 并行关系
    
  • 汇聚关系
    
    • AND汇聚关系

      CompletionStage<R> thenCombine(other, fn);
      CompletionStage<R> thenCombineAsync(other, fn);
      CompletionStage<Void> thenAcceptBoth(other, consumer);
      CompletionStage<Void> thenAcceptBothAsync(other, consumer);
      CompletionStage<Void> runAfterBoth(other, action);
      CompletionStage<Void> runAfterBothAsync(other, action);
      

    • OR 汇聚关系

    CompletionStage applyToEither(other, fn);
    CompletionStage applyToEitherAsync(other, fn);
    CompletionStage acceptEither(other, consumer);
    CompletionStage acceptEitherAsync(other, consumer);
    CompletionStage runAfterEither(other, action);
    CompletionStage runAfterEitherAsync(other, action);
    

• 异常处理

  CompletionStage exceptionally(fn);
  CompletionStage<R> whenComplete(consumer);
  CompletionStage<R> whenCompleteAsync(consumer);
  CompletionStage<R> handle(fn);
  CompletionStage<R> handleAsync(fn);
  

课后思考

• 创建采购订单的时候，需要校验一些规则，例如最大金额是和采购员级别相关的。有同学利用 CompletableFuture 实现了这个校验的功能，逻辑很简单，首先是从数据库中把相关规则查出来，然后执行规则校验。你觉得他的实现是否有问题呢？

//采购订单
PurchersOrder po;
CompletableFuture<Boolean> cf = 
  CompletableFuture.supplyAsync(()->{
    //在数据库中查询规则
    return findRuleByJdbc();
  }).thenApply(r -> {
    //规则校验
    return check(po, r);
});
Boolean isOk = cf.join();

• 解答
  • 没有进行异常处理，
  • 要指定专门的线程池做数据库查询（读数据库属于io操作，应该放在单独线程池，避免线程饥饿）
  • 如果检查和查询都比较耗时，那么应该像之前的对账系统一样，采用生产者和消费者模式，让上一次的检查和下一次的查询并行起来。

25 | CompletionService：如何批量执行异步任务？

• CompletionService 批量提交异步任务

// 创建线程池
ExecutorService executor =
  Executors.newFixedThreadPool(3);
// 异步向电商S1询价
Future<Integer> f1 = 
  executor.submit(
    ()->getPriceByS1());
// 异步向电商S2询价
Future<Integer> f2 = 
  executor.submit(
    ()->getPriceByS2());
// 异步向电商S3询价
Future<Integer> f3 = 
  executor.submit(
    ()->getPriceByS3());
    
// 获取电商S1报价并保存
r=f1.get();
executor.execute(()->save(r));
  
// 获取电商S2报价并保存
r=f2.get();
executor.execute(()->save(r));
  
// 获取电商S3报价并保存  
r=f3.get();
executor.execute(()->save(r));

• Future 实现“询价”程序
  • 如上所示，需要一个个的get然后执行下一步操作，f1, f2, f3需要一次等待

// 创建阻塞队列
BlockingQueue<Integer> bq =
  new LinkedBlockingQueue<>();
//电商S1报价异步进入阻塞队列  
executor.execute(()->
  bq.put(f1.get()));
//电商S2报价异步进入阻塞队列  
executor.execute(()->
  bq.put(f2.get()));
//电商S3报价异步进入阻塞队列  
executor.execute(()->
  bq.put(f3.get()));
//异步保存所有报价  
for (int i=0; i<3; i++) {
  Integer r = bq.take();
  executor.execute(()->save(r));
}  

• 阻塞队列的优化方案
  • 如上所示 阻塞队列也能解决这种互相等待操作造成的资源浪费
• 利用 CompletionService 实现询价系统
  • CompletionService 内部实现了一个阻塞队列，默认 LinkedListBlockingQueue（建议覆盖，因为默认的是无界的）
  • CompletionService 会把Future对象放到阻塞队列中。
    代码实现如下所示

// 创建线程池
ExecutorService executor = 
  Executors.newFixedThreadPool(3);
// 创建CompletionService
CompletionService<Integer> cs = new 
  ExecutorCompletionService<>(executor);
// 异步向电商S1询价
cs.submit(()->getPriceByS1());
// 异步向电商S2询价
cs.submit(()->getPriceByS2());
// 异步向电商S3询价
cs.submit(()->getPriceByS3());
// 将询价结果异步保存到数据库
for (int i=0; i<3; i++) {
  Integer r = cs.take().get();// **TODO 验证是否是先执行完的，先入队**
  executor.execute(()->save(r));
}

TODO 验证是否是先执行完的，先入队

• 利用 CompletionService 实现 Dubbo 中的 Forking Cluster
  • Dubbo 中有一种叫做 Forking 的集群模式，这种集群模式下，支持并行地调用多个查询服务，只要有一个成功返回结果，整个服务就可以返回了。

// 创建线程池
ExecutorService executor =
  Executors.newFixedThreadPool(3);
// 创建CompletionService
CompletionService<Integer> cs =
  new ExecutorCompletionService<>(executor);
// 用于保存Future对象
List<Future<Integer>> futures =
  new ArrayList<>(3);
//提交异步任务，并保存future到futures 
futures.add(
  cs.submit(()->geocoderByS1()));
futures.add(
  cs.submit(()->geocoderByS2()));
futures.add(
  cs.submit(()->geocoderByS3()));
// 获取最快返回的任务执行结果
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 能够让异步任务的执行结果有序化，先执行完的先进入阻塞队列，利用这个特性，你可以轻松实现后续处理的有序性，避免无谓的等待，同时还可以快速实现诸如 Forking Cluster 这样的需求。
• CompletionService 的实现类 ExecutorCompletionService，需要你自己创建线程池，虽看上去有些啰嗦，但好处是你可以让多个 ExecutorCompletionService 的线程池隔离，这种隔离性能避免几个特别耗时的任务拖垮整个应用的风险。

课后思考

• 本章使用 CompletionService 实现了一个询价应用的核心功能，后来又有了新的需求，需要计算出最低报价并返回，下面的示例代码尝试实现这个需求，你看看是否存在问题呢？

// 创建线程池
ExecutorService executor = 
  Executors.newFixedThreadPool(3);
// 创建CompletionService
CompletionService<Integer> cs = new 
  ExecutorCompletionService<>(executor);
// 异步向电商S1询价
cs.submit(()->getPriceByS1());
// 异步向电商S2询价
cs.submit(()->getPriceByS2());
// 异步向电商S3询价
cs.submit(()->getPriceByS3());
// 将询价结果异步保存到数据库
// 并计算最低报价
AtomicReference<Integer> m =
  new AtomicReference<>(Integer.MAX_VALUE);
for (int i=0; i<3; i++) {
  executor.execute(()->{
    Integer r = null;
    try {
      r = cs.take().get();
    } catch (Exception e) {}
    save(r);
    m.set(Integer.min(m.get(), r));
  });
}
return m;

Copy

解答

• 以上代码无法保证三个线程 和 主线程 return m的顺序。可以加个CountDownLatch 来保证线程执行完成 再让主线程return。

26 | Fork/Join：单机版的MapReduce

笔记

• 从任务的角度看待并发编程
  • 不难发现线程池、Future、CompletableFuture、CompletionService都是站在任务的角度看待并发编程，将视野扩大，不再将精力都浪费在线程的协作上
  • 对于简单的并行任务，可以通过线程池+Future
  • 如果任务之间有耦合关系，不管是AND或者OR，都可以CompletableFuture来解决
  • 如果是批量任务，可以通过CompletionService来解决
• 并发编程关注的三个问题
  • 互斥、
    • 实现互斥的方案：锁：Synchronized、ReentrantLock、ReadWriteLock、StampedLock、、、
  • 分工、
    • 实现分工的方案：线程池、CompletionService、CompletableFuture
  • 协作
    • 线程协作的方案：管程：synchronized+wait+notify、Lock+Condition、CountDownLatch、CyclicBarrier
• 从更高的视野看待并发编程
  • 任务
    • 线程任务调度方式：线程池+Future、CompletionService、CompletableFuture
• 任务类型
  • 并行任务：线程池 + Future + 阻塞队列、CompletableFuture
  • 批量任务：CompletionService
  • 聚合任务：CompletableFuture
  • 分治任务：Fork/Join
• Fork/Join
  • 理解
    • Fork对应任务分解
    • join对应任务聚合
  • 工具
    • ForkJoinPool
      • 生产者-消费者模型
      • 任务窃取当pool中的呃线程空闲了，会去窃取其他工作队列中的任务。TODO 确认是一个等待池还是每个线程有一个等待池
    • ForkJoinTask
      • 抽象实现类：RecursiveAction（compute()无返回值）
      • 抽象实现类：RecursiveTask（compute()有返回值）
        以上两者的关系类似 ThreadPoolExecutor 和 Runnable的关系

总结

• Fork/Join 并行计算框架的核心组件是 ForkJoinPool。ForkJoinPool 支持任务窃取机制，能够让所有线程的工作量基本均衡，
  不会出现有的线程很忙，而有的线程很闲的状况，所以性能很好。Java 1.8 提供的 Stream API 里面并行流也是以 ForkJoinPool
  为基础的。不过需要你注意的是，默认情况下所有的并行流计算都共享一个 ForkJoinPool，
  这个共享的 ForkJoinPool 默认的线程数是 CPU 的核数；如果所有的并行流计算都是 CPU 密集型计算的话，完全没有问题，
  但是如果存在 I/O 密集型的并行流计算，那么很可能会因为一个很慢的 I/O 计算而拖慢整个系统的性能。
  所以建议用不同的 ForkJoinPool 执行不同类型的计算任务。

27 | 并发工具类模块热点问题答疑

笔记

• 注意while(true)的问题
  • 一般while(true)需要break条件，一般是超时时间，不然容易导致死循环
  • while(true) & Lock和Condition里面的活锁问题
• notifyAll和signalAll
  • 一般使用All比notify和signal更安全
• Semaphore 需要锁中锁
  • Semaphore 允许多个线程访问一个临界区，这也是一把双刃剑，当多个线程进入临界区时，如果需要访问共享变量就会存在并发问题，所以必须加锁，也就是说 Semaphore 需要锁中锁。
  • 个人理解，Semaphore不是一个锁，只是一个并发工具，所以遇到共享变量问题，依然是需要锁的
• 锁的申请和释放要成对出现
• 回调总要关心执行线程是谁
  • 当看到回调函数的时候，一定问一问执行回调函数的线程是谁
• 共享线程池：有福同享就要有难同当
  • 对于I/O密集型和CPU密集型的线程池要做业务隔离
• 线上问题定位的利器：线程栈 dump
  • 善于利用JSP和jstack命令

28 | Immutability模式：如何利用不变性解决并发问题？

• 快速实现具备不可变性的类
  • 将一个类所有的属性都设置成 final 的，并且只允许存在只读方法，那么这个类基本上就具备不可变性了
  • 将类也设置成 final 保证不能通过继承修改该类的final属性
• String
  • String 类是final的，就是为了保证线程安全的
  • String 类的字符串替换操作 String#replace方法如何保证的线程安全？
    • replace 是通过将替换后的值返回的形式"修改"的字符串，严格来说不能说是修改，而是生成了一个新的字符串。也就是生成了一个新的不可变String对象
  • TODO，思考 String 类型为什么是存在常量池的
• 享元模式
  享元模式本质上其实就是一个对象池，利用享元模式创建对象的逻辑也很简单：创建之前，首先去对象池里看看是不是存在；如果已经存在，就利用对象池里的对象；
  如果不存在，就会新创建一个对象，并且把这个新创建出来的对象放进对象池里。
  • Long & Integer
    • Long 这个类并没有照搬享元模式，Long 内部维护了一个静态的对象池，仅缓存了[-128,127]之间的数字，这个对象池在 JVM 启动的时候就创建好了，
      而且这个对象池一直都不会变化，也就是说它是静态的。
• 使用 Immutability 模式的注意事项
  • 对象的所有属性都是 final 的，并不能保证不可变性；
  • 不可变对象也需要正确发布。
  • 注意不可变的边界，如下所示，虽然Bar对象中Foo属性是final的，但是不能保证foo的属性不能被修改。

class Foo{
  int age=0;
  int name="abc";
}
final class Bar {
  final Foo foo;
  void setAge(int a){
    foo.age=a;
  }
}

总结

• 利用 Immutability 模式解决并发问题，也许你觉得有点陌生，其实你天天都在享受它的战果。Java 语言里面的 String 和 Long、Integer、Double
  等基础类型的包装类都具备不可变性，这些对象的线程安全性都是靠不可变性来保证的。
  Immutability 模式是最简单的解决并发问题的方法，建议当你试图解决一个并发问题时，可以首先尝试一下 Immutability 模式，看是否能够快速解决。
• 无状态
  具备不变性的对象，只有一种状态，这个状态由对象内部所有的不变属性共同决定。其实还有一种更简单的不变性对象，那就是无状态。
  无状态对象内部没有属性，只有方法。除了无状态的对象，你可能还听说过无状态的服务、无状态的协议等等。无状态有很多好处，最核心的一点就是性能。
  在多线程领域，无状态对象没有线程安全问题，无需同步处理，自然性能很好；在分布式领域，无状态意味着可以无限地水平扩展，所以分布式领域里面性能的瓶颈一定不是出在无状态的服务节点上。
• 其实spring的 单例Bean 也是一种无状态的Bean ，也就是在单例Bean中最好不要定义共享变量，因为如果不加入线程安全的控制的话，读写一定能引起并发问题。

29 | Copy-on-Write模式：不是延时策略的COW

笔记

• Copy-on-Write 写时复制
  • 基本思想-懒惰策略
    • Copy-On-Write简称COW，是一种用于程序设计中的优化策略。其基本思路是，从一开始大家都在共享同一个内容，当某个人想要修改这个内容的时候，
      才会真正把内容Copy出去形成一个新的内容然后再改，这是一种延时懒惰策略。
  • 基本思想-读写分离
    • 当修改容器时，不首先修改原容器，而是将原容器copy一份，然后修改copy的容器，之后再将修改后的容器替换原容器。这样读和写操作的永远都是两个容器。
      也就实现了读写分离。
• Copy-on-Write 在操作系统中的应用
  • 类 Unix 的操作系统中创建进程的 API 是 fork()，传统的 fork() 函数会创建父进程的一个完整副本，例如父进程的地址空间现在用到了 1G 的内存，那么 fork() 子进程的时候要复制父进程整个进程的地址空间（占有 1G 内存）给子进程，这个过程是很耗时的。而 Linux 中的 fork() 函数就聪明得多了，fork() 子进程的时候，并不复制整个进程的地址空间，而是让父子进程共享同一个地址空间；只用在父进程或者子进程需要写入的时候才会复制地址空间，从而使父子进程拥有各自的地址空间。
• java中的 CopyOnWriteArrayList 和 CopyOnWriteArraySet是一种典型的空间换时间的做法
  • java中的CopyOnWriteXXX容器采用的都是当元素发生变化，就进行copy操作。
    • 存在的问题：当容器非常大，这样就会浪费很多的空间。

总结

• 目前 Copy-on-Write 在 Java 并发编程领域知名度不是很高，很多人都在无意中把它忽视了，但其实 Copy-on-Write 才是最简单的并发解决方案。
  它是如此简单，以至于 Java 中的基本数据类型 String、Integer、Long 等都是基于 Copy-on-Write 方案实现的。Copy-on-Write 是一项非常通用的技术方案，
  在很多领域都有着广泛的应用。不过，它也有缺点的，那就是消耗内存，每次修改都需要复制一个新的对象出来，好在随着自动垃圾回收（GC）
  算法的成熟以及硬件的发展，这种内存消耗已经渐渐可以接受了。所以在实际工作中，如果写操作非常少，那你就可以尝试用一下 Copy-on-Write，效果还是不错的。

思考题

• 为什么java中没有 CopyOnWriteLinkedList
  • 知识点：链表是分散的存储空间，通过 指针串联起来的
  • 知识点：数组是连续的存储空间
  • 当需要复制数组，只需要复制数组所在的连续空间即可，是一个时间复杂度为O(1)的操作。
    而链表是不连续的存储空间，所以复制链表意味着需要 按照链表的指针，遍历整个链表。这将是一个时间复杂度为O(n)的操作。

30 | 线程本地存储模式：没有共享，就没有伤害

笔记

• 线程封闭
  • 局部变量
  • ThreadLocal
• ThreadLocal源码
  • 切记 ThreadLocalMap是Thread持有的
    

class Thread {
  //内部持有ThreadLocalMap
  ThreadLocal.ThreadLocalMap 
    threadLocals;
}
class ThreadLocal<T>{
  public T get() {
    //首先获取线程持有的
    //ThreadLocalMap
    ThreadLocalMap map = Thread.currentThread().threadLocals;
    //在ThreadLocalMap中
    //查找变量
    Entry e = map.getEntry(this);
    return e.value;  
  }
  static class ThreadLocalMap{
    //内部是数组而不是Map
    Entry[] table;
    //根据ThreadLocal查找Entry
    Entry getEntry(ThreadLocal key){
      //省略查找逻辑
    }
    //Entry定义
    static class Entry extends
    WeakReference<ThreadLocal>{
      Object value;
    }
  }
}

• ThreadLocal 的内存泄漏问题
  • 一个误区、不是说ThreadLocal一定存在内存泄漏，在一般场景中，Thread持有ThreadLocalMap，ThreadLocalMap以弱引用的方式持有ThreadLocal，当线程Thread被回收，意味着ThreadLocal一定会被回收。
  • ThreadLocal的内存泄漏发生在配合线程池使用的场景中
    • 在线程池中 线程存活时间很长，往往同应用程序是同生共死的，这就意味着 Thread 持有的 ThreadLocalMap 一直都不会被回收，再加上 ThreadLocalMap 的 Entry对 ThreadLocal的引用是弱引用（WeakReference）
    • 所以 只要 ThreadLocal 结束了自己的生命周期 是可以被回收掉的。但是 Entry 中的value 却是被Entry强引用。所以即便 Value的生命周期结束了，value 也是无法被回收的（可达性分析算法），从而导致内存泄漏
  • 解决方案-一般ThreadLocal配合线程池使用，需要使用try{}finally{}手动释放资源

ExecutorService es;
ThreadLocal tl;
es.execute(()->{
  //ThreadLocal增加变量
  tl.set(obj);
  try {
    // 省略业务逻辑代码
  }finally {
    //手动清理ThreadLocal 
    tl.remove();
  }
});

总结

• Spring的 线程池管理 采用的就是 ThreadLocal ，每个线程可以针对性的修改自己的线程池。通过这个，可以写一个切面来切换数据源。
• 线程本地存储模式本质上是一种避免共享的方案，由于没有共享，所以自然也就没有并发问题。

31 | Guarded Suspension模式：等待唤醒机制的规范实现

笔记

• 等待唤醒机制
  
  • 场景：
    • 服务调用方（生产者） 将消息 发动到 MQ。
    • 服务提供方（消费者） 将消息 消费。
      现在需要消费者消费完消息后，通知 生产者，然后生产者再响应其客户端。
  • 分析
    • 生产者需要等待
    • 消费者消费完需要通知生产者
  • 方案
    • 分析场景，其实就是一个异步转同步的过程
    • 生产者等待 消费者通知 可以使用 管程实现
    • 管程采用Lock + Condition 实现等待 和 通知的关键 需要找到一个将 生产者发送消息 和 消费者响应消息，将两个消息对应到一个Condition
      • 方案就是将消息对应的Condition缓存起来 这样的话就可以采用 **Key(msg.id) --> value(Condition)
    • 当 上下游 生产者和消费者都是集群化部署，如何解决？
      • 设计 MQ的topic = msg + ip 这样就能保证 同一个节点消费的通知消息一定是 本节点生产的消息

32 | Balking模式：再谈线程安全的单例模式

笔记

• 多线程版本的if
  • 场景：
    • 类似语雀，编辑完了需要保存。
      • 首先需要维护一个变量“changed”标识当前文档有没有被修改
      • 然后一个定时任务 定时执行自动保存任务。
        伪代码如下所示

//自动存盘操作
void autoSave(){
  synchronized(this){//同步代码块保证 对changed的读和下面的写互斥
    if (!changed) {
      return;
    }
    changed = false;
  }
  //执行存盘操作
  //省略且实现
  this.execSave();
}
//编辑操作
void edit(){
  //省略编辑逻辑
  ......
  synchronized(this){
    changed = true;
  }
}  

• 针对上面的场景，总结成为一个并发设计模式 - Balking模式
  • Balking 本质上是一种规范化的解决“多线程版本if”的方案
• Balking 模式范式 -- 使用管程实现

boolean changed=false;
//自动存盘操作
void autoSave(){
  synchronized(this){
    if (!changed) {
      return;
    }
    changed = false;
  }
  //执行存盘操作
  //省略且实现
  this.execSave();
}
//编辑操作
void edit(){
  //省略编辑逻辑
  ......
  change();
}
//改变状态
void change(){
  synchronized(this){
    changed = true;
  }
}

• volatile 实现Balking模式
  • 对于上面使用 管程 实现的balking模式，是针对多线程版本的互斥方案。如果，在场景中不需要保证变量 changed 的互斥，可以采用volatile来实现
  • 切记 使用 volatile 的前提是对原子性没有要求。
    仍然是上面的场景，如果对互斥要求不严格，表现在业务上，也就是如果多保存几次也无所谓，那么就可以去掉 管程 使用volatile来修饰_changed_变量。
  • 重点 volatile实现Balking 的场景十分有限。只需要记住使用 volatile 的前提是对原子性没有要求。 即可

课后思考

以下代码是为了保证 count只被计算一次。

class Test{
  volatile boolean inited = false;
  int count = 0;
  void init(){
    if(inited){//1
      return;
    }
    inited = true;//2
    //计算count的值
    count = calc();//3
  }
}  

• 以上代码是否存在线程安全性问题？
• 当多个线程同时进入1处代码，也就意味着，这几个想成都能完整执行2,3处的代码。所以count就有可能被计算多次。

总结

• 只执行一次的场景
  • 场景：
    SpringBoot 启动流程的源码中，refresh 方法只能被执行一次。
  • 方案：保证代码绝对只执行一次最好的解决方案就是使用原子类AtomicBoolean。
    AtomicBoolean.compareAndSet(false, true)。
• Balking 模式的启发
  • 之前实现的离线 license 可以通过 Balking 模式优化一般