并发与锁

先列个提纲逐步细化。

一、 JUC AQS

1) AbstractQueuedSynchronizer 的结构

static final class Node {
    static final Node SHARED = new Node(); // 共享模式的标记
    static final int CANCELLED = 1; // 标志着线程被取消
    static final int SIGNAL = -1; //标志着后在自己释放锁放弃等待时需要唤醒继节点（用于互斥锁）
    static final int CONDITION = -2; //标志着当前节点在条件等待队列中（由于条件是基于互斥锁的）
    static final int PROPAGATE = -3; // 标志着此链表存放都是共享锁等待节点，需要广播
    volatile int waitStatus; // 默认状态是0, 小于0的状态都是有特殊作用，大于0的状态表示已取消
    volatile Node prev;
    volatile Node next;
    volatile Thread thread; // 该节点对应的线程

    /**如果是SHARED，表示当前节点是共享模式，
    *  如果是null，当前节点是独占模式，
    *  如果是其他值，当前节点也是独占模式，不过这个值也是Condition队列的下一个节点。
    */
    Node nextWaiter;

    // 是不是共享模式
    final boolean isShared() {
        return nextWaiter == SHARED;
    }
}

2)互斥锁的获取过程

• CAS争夺所有权
  没有在等待队列中就是个CAS没什么要说的，如果前驱结点是头节点，那就需要再次尝试CAS争夺所有权。需要注意的是无论锁可中断与否，此时获取锁的成功都是不响应中断请求的。
• CAS 的精髓
  CAS的精髓有两个。首先是可以对当前值是预期相等才能CAS充分利用，结合不等式传递性，这样不仅可以做等值替换，大于小于都能做 stackoverflow上有例子就不列举了。再一个是增加的过程是逐步改变的从全局看是不能跳跃的；例如rotate 回转数的问题，netty线程老代码取余数的显然会被鄙视，后来2的指数用位运算回转勉强。

import java.util.concurrent.atomic.AtomicInteger;

public class RotateAtomicInteger {

  private final AtomicInteger val;
  private final int min;
  private final int max;

  public RotateAtomicInteger( int min, int max) {
    if(max-min<2){
      throw new IllegalArgumentException("可能有ABA问题");
    }
    this.val = new AtomicInteger(min);
    this.min = min;
    this.max = max;
  }

  public int getAndRotateIncr(){
    for(;;){
      int cur=val.get();//因为如果cas失败说明之前有人成功过所以这个值就不能再用了
      int next = cur==max ? min :cur+1;//全局来看是不可跳跃
      if(val.compareAndSet(cur, next)){
        return cur;
      }      
    }
  }

}

3) 获取失败挂起的过程

• 入队的注意点
  如果竞争失败就加入队列，加入队列如果为空就从头插，否则从尾部插入，插入后不能闲着还要再试一次，即便失败了还不能立即park，要绕过取消的等待者，并且CAS标记前驱结点出队时别忘了唤醒自己。CAS失败了入队后的各种尝试还得再来一遍。

• 如何防止丢失唤醒
  上面这样做就够了吗，显然还不行；满足自己去“睡”的条件的check和“睡”显然不是一个原子操作。在check后和“睡着”前如果条件变了就没人叫醒。 unpark和park早就预防相关问题，猜想内部是有变量记忆了上次的操作后状态，同时基于操作系统提供的锁保证了原子性。

4) 唤醒的过程

• 释放锁唤醒等待的过程
  先释放锁标记，再置通知标记为取消状态，然后从后往前找离自己最近的未取消结点唤醒，为什么0状态也生效刚好上面已经解释过了。另外没区分条件队列，所以虚假唤醒这也是原因之一。

• 如何防止惊群问题
  锁的唤醒通常是有唤醒一个等待线程和唤醒全部等待线程的；通常在编程条件下如果想减少全部唤醒的引发不必要的竞争时，还要注意虚假 唤醒问题，被唤醒的线程如果条件不满足条件释放锁后造成唤醒链脱节的问题。而这种情况又不能简单在自己释放锁前去唤醒其他等待者。 所以比如在leader-follower模式中，如果leader线程只在原leader线程离任时唤醒任意一个等待线程时，则当线程都在follower模式下因为没有唤醒者陷入停机状态，所以在每次完成工作时还要尝试自荐为leader。如果leader在离任时唤起其他所有线程显然是惊群的。

5) 条件等待队列

-为什么nextWaiter属性没用volatile修饰
首先如果要条件等待或者唤醒必须在持有锁（夹杂volatile中间）保证了可见性。

• 等待条件的流程 待续

• 唤醒条件的流程 待续

6) AQS小结

• 为什么要双向链表？用带头尾指针的单向链表行不行呢？
  我认为问题比较麻烦，不能说完全不可以。主要问题是中间节点取消等待获取锁时，摘除这个节点需要对前一节点的后继指针做CAS，例如：A->B->C->D 这种情况，摘除B时需要对A的后继指针做CAS，此时如果又要摘除C由于C没有稳定前驱指针做不了。如果CAS前先原子更新当前节点为取消状态，然后对前一节点的两个字段同时做CAS（状态和后继指针域１６字节的不知道除了X８６其他平台是否支持），如果前一节点的状态已经为取消状态了，需要从头再查找当前线程在链表中的位置进而取得其前一节点重试。再有一种办法就是标记删除了，这个方式累积过多的无效节点显然是不合适的。总之这两种办法要么时间复杂度不符合要么空间复杂度不符合。

• 流言：cas+clh队列语义上等价于锁
  这个说法的问题在于线程加入clh队列后，需要借助cpu指令中的屏蔽中断与禁止抢占保证调度切换的原子性。另外前面提到的获取锁的小过程不响应中断也是个启示。

• 双检锁的简单优化
  代码注释强调了关键的逻辑和注意点。重点对比findById与findByIdSync两个方法。

import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.locks.LockSupport;
import sun.misc.Unsafe;
//只是示意代码没测试慎用
public class Dcl { 
    private static final Unsafe unsafe = getUnsafeInstance();
    private static final long readableOffset;
    // 1 can read , 0 can not read
    private volatile int readable=1;
    private Map<String,Object> map=new HashMap<>();
    /**
     * 可以用数组 cas索引(类似ringbuffer)保存，不过扩容有点麻烦，不过通常线程数不会太大。。。
     * 这里用带头尾指针的单链表cas尾插法存储等待线程显然比较合适
     * 现在为突出双检锁，用了AQS显然违反自举原则
     */
    private LinkedTransferQueue<Thread> list = new LinkedTransferQueue<>();

    private boolean compareAndSwapReadable(int expect, int update) {
        return unsafe.compareAndSwapInt(this, readableOffset, expect, update);
    }
    
    private Object getCache(String id) {
        return map.get(id);//TODO
    }

    private Object loadFromDb(String id) {
        return null; //TODO 
    }

    private void setCache(String id, Object value) {
        //copy on write 这样可以是线程非安全的map  TODO
    }

    //volatile + unsafe cas + park/unpark 版本
    public Object findById(String id) throws InterruptedException {
        Object data;
        for (;;) {
            data = getCache(id);
            if (data != null) {
                return data;
            }
            if (readable==1 && compareAndSwapReadable(1, 0)) { //volatile读减少cas
                data = getCache(id); 
                if (data == null) {
                    data = loadFromDb(id);
                    setCache(id, data);
                }
                readable = 1;
                for (Thread th = list.peek(); th != null; th = list.peek()) {
                    LockSupport.unpark(th);//即便unpark早于park也不会有问题因为有记忆功能
                }
                return data;
            } else {//cas失败就不用像双检锁那样再次过独木桥
                list.add(Thread.currentThread());
                if (readable == 0) {
                    LockSupport.park();
                }
                list.remove(Thread.currentThread());
            }
        }
    }

    // synchronized 双检锁版本
    public Object findByIdSync(String id) {
        Object data;
        data = getCache(id);
        if (data != null) {
            return data;
        }
        synchronized (this) { // 防止并发，不过loadFromDb期间的线程都会阻塞在这里后面依次拿到锁通过
            data = getCache(id);
            if (data != null) {// 这个条件可以取反优化流程，只是为了和下面的双检测对比
                return data;
            }
            data = loadFromDb(id);
            setCache(id, data);
            return data;
        }
    }
    
    private static Unsafe getUnsafeInstance(){
        try {
            Field theUnsafeInstance;
            theUnsafeInstance = Unsafe.class.getDeclaredField("theUnsafe");
            theUnsafeInstance.setAccessible(true);
            return (Unsafe) theUnsafeInstance.get(Unsafe.class);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }       
    }
    
    static {
        try {
            Field f = Dcl.class.getDeclaredField("readable");
            f.setAccessible(true);
            readableOffset = unsafe.objectFieldOffset(f);
        } catch (Exception ex) {
            throw new Error(ex);
        }
    }

}

二、从缓存一致性协议和内存屏障看 volatile atomic

• 1）背景：cpu要提高性能执行，派发调度打乱了原有指令的顺序。同时为提高读写性能多了多级cache，在cache和执行引擎中间还有对外不可见的读写缓冲区。另一个是缓存类似硬件的hashmap，其中地址为key数据为value还有额外的有效性表示。为了有均衡的性能缓存采用通常是全相连与组相连结合的方式。而且缓存为了调高有效载荷和充分发挥程序的局部性采用是较大单元存储的，通常是64B。（另外本文整体都是以x86体系考虑）

  
  skylake_server架构图

注意几个关键（load buffer, store buffer, WC buffer） 注意外部传来的失效队列

硬件虽然在不断改进从QPI到UPI连接方式通信上有了不少变化，但是的思想最后这幅图依然不过时。

写缓存策略

1. 本地缓存的写策略有：写回WB，写穿透WT，在cache miss情况下还有 写合并WC；
2. 远程缓存的写策略有：写更新WU，写时效WI，
   写更新与写失效的对比：
   写更新，看起来不需要经过主存同步更快，但更新缓存行比起一条失效指令代价大许多；
   写更新可能是断续多次交互而失效只要一次交互，而且缓存行内容的大小比缓存地址大位宽大不少节约了总线带宽。

• 2） 缓存一致性协议
  MESI、MOSI、MESIF 一图胜过千言，段落后面的图
  多核在读写同一地址时有的对顺序和可见性甚至是大于字长的读写原子性，需要核间协调通信来支撑。所以在这样的情形下缓存一致性协议就出现，也就说MESI协议以及Intel扩展到NUMA模式下的MESIF。

  
  mesi.png

• 3）内存屏障
  在编译器层面限制预优化的指令重排，在硬件上内存屏障对指令乱序做了边界限定，同时对核心内对外不可见缓冲区刷新和核心间缓存同步过程的异步通信收敛这两种情况的做了约束。正是这样限定和约束保证了多核对同一内存地址读写时的正确性。

X86-64一致性很高，LoadLoad Memory Barrier是因为读屏障会强制等待Invalidate Queue清空才继续执行，这样可以消除其引发的读乱序问题。

缓存一致性协议显然在硬件上有锁的语义，其背后实现非常复杂。同时为提高性能过程又做了异步，牺牲了同步的实时性不能完全保证程序的正确性。总体来说缓存一致性协议提供了缓存一致性的必要的手段和机制，内存屏障是简单通过对乱序执行的抑制以及对异步过程的完成等待。在这两种机制配合下既可以保证程序的正确，又为编程提供灵活的接口，另外一个是这种编程接口使用也很简单。软硬实现和接口搭配的很巧妙，值得借鉴。

• 4)java volatile 语义实现
  如果a线程先写入新值，b线程紧接着去读（在a线程赋值时b线程由于受到MESI协议约束不能读，su）（赋值后隔一个时钟周期，这样可以使得无论是lock前缀的指令排空store buffer 并且失效其他处理器的对于缓存得以执行完，或者时钟中断引发store buffer排空进而触发缓存一致性同步）去读就能到 最新值，不过由于需要间隔一个时钟周期，看似违反happens-before；不过由于紧接着是lock的情况在紧接着的时钟周期里由于lock触发缓存一致性同步会失效该数据项的旧值所以不会有问题，时钟中断的间接触发缓存同步也类似。

• unsafe 源码 https://github.com/dmlloyd/openjdk/blob/jdk/jdk/src/hotspot/share/prims/unsafe.cpp

• 5. CAS 指令与总线锁与伪共享 CMPXCHG16B
     Java8的伪共享和缓存行填充--@Contended注释

三、Synchronized

• 同步锁升级流程知乎专栏讲很详细，直接上个概括的图https://zhuanlan.zhihu.com/p/29866981
  java_synchronized原理解析.png

有了这么详尽的图，过程就不赘述了，补充QA

• 对象头

  Q：为什么在偏向锁状态下线程id可以存下？
  A：我认为通常线程id两个字节表示就可以了所以存下妥妥的。
  Q：为什么在轻量级锁时，栈中锁记录地址仍然如用沾满整个对象头呢？
  A：我认为栈在低端地址空间，同时每个线程分配的栈空间大小又很小的所有几乎不可能突破这个上限。

另外jvm对象头和jvm对象锁以及Object类中有关线程同步的方法，显然是java的糟粕，不应有宗教式的谄媚甚至崇拜。

• 锁升级
  总体的原则就由近及远，尽量缩小范围尽量降低消耗。这个思想贯穿了所有锁实现中，不管是jvm、os还是分布式的。
• 安全点及延申
  通过R大的几篇博文了解到关于精确式GC和安全点的概念。安全点处可以调度锁同样可以其他的执行体，比如golang在gc安全点调度协程，并且有人以此说go的协程是抢占式调度，我觉得是不妥的。jvm的纤程还在孵化中估计到时候也会使用基于安全点调度。另外想必这些锁的实现都会改一轮，在使用os的锁之前做些协程的调度。
  另外说到GC和并就不得不说几天前R大在知乎发的关于ZGC利用内存屏障替代锁的实现方案，没有读懂如何做到，备忘下以后尝试做个分析。

四、Futex

Futex是一种用户态和内核态混合的同步机制。首先，同步的进程间通过mmap共享一段内存，futex变量就位于这段共享 的内存中且操作是原子的，当进程尝试进入互斥区或者退出互斥区的时候，先去查看共享内存中的futex变量，如果没有竞争发生，则只修改futex,而不 用再执行系统调用了。当通过访问futex变量告诉进程有竞争发生，则还是得执行系统调用去完成相应的处理(wait 或者 wake up)。简单的说，futex就是通过在用户态的检查，如果了解到没有竞争就不用陷入内核了，大大提高了低竞条件下的效率。

• 自旋锁

  内核的基本锁之一

• 自旋优化（jdk9与自旋提示）

自旋是在循环中不断重试检测条件变量的值，这样为了保证可见性，就需要不断的同步缓存，造成很多不必要的开销。intel有专门的pause指令优化，就是让流水线停顿而不是干傻事。

• RCU锁

  包含的数据通常都是指针类型的，每个核上都发生过调度就说明如果其他核在该rcu锁的读临界区，完成一轮调度后（读临界区不可抢占）

  如果再有新的线程可见的数据是新的老版本的可以释放了。如果是可抢占的RCU就需要在锁定开头和释放的结尾维护一个计数器。

• 屏蔽中断与禁止抢占

五、 锁的构成要素

cas ：锁的竞争手段
等待队列：锁的管理手段
屏蔽中断禁止抢占的原子调度上下文的切换：锁的收敛保障

六、理解与体会

• 不总结不准备就会拔剑四顾心茫然

• 读写时序

• 几种并发模型 actor、CSP、CPS/CallCC

• 控制流与数据流的等价性
  方法调用的派发方式：静态派发、查表派发、消息派发

• 锁的本质问题是时序问题，调度可以解决时序问题，调度问题从cpu角度看又是执行权的问题，锁是保护了数据，锁是通过限制代码的执行来保护数据的。

• 如何优化并发程序

• mesi在分布式环境的启示

分布式环境下缓存一致性也是个经典的话题，从mesi协议我们能有什么启示。首先总线的作用是巨大的，他保证了多核乃至多路cpu间的缓存一致性分布式环境下由于没有统一的总线，缺少收敛的协调控制。

引用

1. https://zh.wikipedia.org/wiki/%E5%86%92%E9%99%A9_(%E8%AE%A1%E7%AE%97%E6%9C%BA%E4%BD%93%E7%B3%BB%E7%BB%93%E6%9E%84)
2. https://zh.wikipedia.org/wiki/%E5%BE%AE%E6%9E%B6%E6%A7%8B
3. https://en.wikichip.org/wiki/intel
4. https://dzone.com/articles/memory-barriersfences
5. https://www.slideshare.net/frogd/cpu-cache-and-memory-ordering
6. https://mechanical-sympathy.blogspot.com/2013/02/cpu-cache-flushing-fallacy.html
7. http://www.wowotech.net/kernel_synchronization/Why-Memory-Barriers.html
8. https://github.com/WPCH/study/blob/master/CPU_Arch/CPU_Arch.md#cache-memory%E7%B3%BB%E7%BB%9F

---

总体思路是前面已经有很多人写了锁和并发相关what和how主题，本文尝试从why和why not这个角度去写，这显然超出我能力上限难免有错误，欢迎交流qq 506602332