并发编程

进程/线程

进程

是具有一定独立功能的程序、它是系统进行资源分配和调度的一个独立单位，重点在系统调度和单独的单位，也就是说进程是可以独立运行的一段程序。

线程

是进程的一个实体，是CPU调度和分派的基本单位，他是比进程更小的能独立运行的基本单位，线程自己基本上不拥有系统资源。在运行时，只是暂用一些计数器、寄存器和栈。
线程是进程的子集，一个进程可以有很多线程，每条线程并行执行不同的任务。不同的进程使用不同的内存空间，而所有的线程共享一片相同的内存空间。每个线程都拥有单独的栈内存用来存储本地数据。

1. 一个线程只能属于一个进程，而一个进程可以有多个线程，但至少有一个线程(通常说的主线程)。
2. 资源分配给进程，同一进程的所有线程共享该进程的所有资源。
3. 线程在执行过程中，需要协作同步。不同进程的线程间要利用消息通信的办法实现同步。
4. 处理机分给线程，即真正在处理机上运行的是线程。
5. 线程是指进程内的一个执行单元，也是进程内的可调度实体。

并行(Parallel)与并发(Concurrent)

Erlang之父Joe Armstrong用一张图解释：并发是两个队列交替使用一台咖啡机，并行是两个队列同时使用两台咖啡机。
并发和并行都可以是很多个线程，就看这些线程能不能同时被多个cpu执行，如果可以就说明是并行，而并发是多个线程被一个cpu轮流切换着执行。

多线程

    public
    class Thread implements Runnable {

        public enum State {
            /**
            * Thread state for a thread which has not yet started.
            */
            NEW,

            /**
            * Thread state for a runnable thread.  A thread in the runnable
            * state is executing in the Java virtual machine but it may
            * be waiting for other resources from the operating system
            * such as processor.
            */
            RUNNABLE,

            /**
            * Thread state for a thread blocked waiting for a monitor lock.
            * A thread in the blocked state is waiting for a monitor lock
            * to enter a synchronized block/method or
            * reenter a synchronized block/method after calling
            * {@link Object#wait() Object.wait}.
            */
            BLOCKED,

            /**
            * Thread state for a waiting thread.
            * A thread is in the waiting state due to calling one of the
            * following methods:
            * <ul>
            *   <li>{@link Object#wait() Object.wait} with no timeout</li>
            *   <li>{@link #join() Thread.join} with no timeout</li>
            *   <li>{@link LockSupport#park() LockSupport.park}</li>
            * </ul>
            *
            * <p>A thread in the waiting state is waiting for another thread to
            * perform a particular action.
            *
            * For example, a thread that has called <tt>Object.wait()</tt>
            * on an object is waiting for another thread to call
            * <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on
            * that object. A thread that has called <tt>Thread.join()</tt>
            * is waiting for a specified thread to terminate.
            */
            WAITING,

            /**
            * Thread state for a waiting thread with a specified waiting time.
            * A thread is in the timed waiting state due to calling one of
            * the following methods with a specified positive waiting time:
            * <ul>
            *   <li>{@link #sleep Thread.sleep}</li>
            *   <li>{@link Object#wait(long) Object.wait} with timeout</li>
            *   <li>{@link #join(long) Thread.join} with timeout</li>
            *   <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li>
            *   <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li>
            * </ul>
            */
            TIMED_WAITING,

            /**
            * Thread state for a terminated thread.
            * The thread has completed execution.
            */
            TERMINATED;

        }

    }

在Thread中共定义了6种状态：

• NEW(初始)
  新创建了一个线程对象，但还没有调用start()方法。
• RUNNABLE(就绪)
  Java线程中将就绪(ready)和运行中(running)两种状态笼统的称为“运行”。
  线程对象创建后，其他线程(比如main线程)调用了该对象的start()方法。该状态的线程位于可运行线程池中，等待被线程调度选中，获取CPU的使用权，此时处于就绪状态(ready)。就绪状态的线程在获得CPU时间片后变为运行中状态(running)。
• BLOCKED(运行中)
  表示线程阻塞于锁。
• WAITING(等待)
  进入该状态的线程需要等待其他线程做出一些特定动作(通知或中断)。等待的时间是永久的，即必须等到某个条件符合才能继续往下走，否则线程不会被唤醒。当执行如下代码的时候，对应的线程会进入到WAITING状态：
  • Object.wait()
    当一个线程执行了Object.wait()的时候，它一定在等待另一个线程执行Object.notify()或者Object.notifyAll()。
  • Thread.join()
  • LockSupport.park()
    当一个线程执行了LockSupport.park()的时候，其在等待执行LockSupport.unpark(thread)。

    public class ThreadState {

        public static void main(String[] args) throws InterruptedException {
            WaitThread t1 = new WaitThread();
            t1.start();
            System.out.println(t1.getState());
            Thread.sleep(100);
            System.out.println(t1.getState());
            LockSupport.unpark(t1);
            System.out.println(t1.getState());
        }
    }

    class WaitThread extends Thread {

        @Override
        public void run() {
            LockSupport.park();
        }
    }

输出为：

RUNNABLE
WAITING
TERMINATED

• TIMED_WAITING(超时等待)
  与WAITING状态的区别就是，这个等待是有一定时效的。等待一段时间之后，会唤醒线程去重新获取锁。当执行如下代码的时候，对应的线程会进入到TIMED_WAITING状态：
  • Thread.sleep(long)
  • Object.wait(long)
  • Thread.join(long)
  • LockSupport.parkNanos()
  • LockSupport.parkUntil()

    public class ThreadState {

        public static void main(String[] args) throws InterruptedException {
            TimeWaitThread thread = new TimeWaitThread("Thread");
            System.out.println(thread.getState());
            thread.start();
            Thread.sleep(100);
            System.out.println(thread.getState());
            Thread.sleep(2000);
            System.out.println(thread.getState());
        }
    }


    class TimeWaitThread extends Thread {
        private Object lock;

        TimeWaitThread(Object lock) {
            this.lock = lock;
        }

        @Override
        public void run() {
            synchronized (lock) {
                try {
                    lock.wait(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("lock end");
            }
        }
    }

输出为：

NEW
TIMED_WAITING
lock end
TERMINATED

• TERMINATED(终止)
  表示该线程已经执行完毕。

悲观锁

总是假设最坏的情况，每次去拿数据的时候都认为别人会修改，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会阻塞直到它拿到锁(共享资源每次只给一个线程使用，其它线程阻塞，用完后再把资源转让给其它线程)。Synchronized和ReentrantLock等独占锁就是悲观锁思想的实现。

乐观锁

总是假设最好的情况，每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号机制和CAS算法实现。
乐观锁适用于多读的应用类型，这样可以提高吞吐量java.util.concurrent.atomic包下面的原子变量类就是使用了乐观锁的一种实现方式CAS实现的。

独占锁

锁在一个时间点只能被一个线程锁占有。根据锁的获取机制，它又划分为：

• 公平锁
  是按照通过CLH等待线程按照先来先得的规则，公平的获取锁。
• 非公平锁
  当线程要获取锁时，它会无视CLH等待队列而直接获取锁。

独占锁的典型实例子是ReentrantLock、ReentrantReadWriteLock.WriteLock。

共享锁

能被多个线程同时拥有，能被共享的锁。
JUC包中的ReentrantReadWriteLock.ReadLock、CyclicBarrier、CountDownLatch、Semaphore都是共享锁。

可重入锁

是同一个线程再次进入同步代码的时候，可以使用自己已经获取到的锁。
重进入是指任意线程在获取到锁之后，再次获取该锁而不会被该锁所阻塞。
每个锁都关联了一个线程持有者和计数器。
线程再次获取锁：锁需要识别获取锁的现场是否为当前占据锁的线程，如果是，则再次成功获取；释放锁时计数器自减，当计数器为0时，锁释放成功。
其它线程请求该锁，则必须等待；而该持有锁的线程如果再次请求这个锁，就可以再次拿到这个锁。

Synchronized

Synchronized是重量级的锁，由C++在JVM实现，无法在Java层面进行扩展和优化，灵活性不高。托管给JVM执行，不会因为异常、或者未释放而发生死锁。 适用于写比较多的情况下(多写场景，冲突一般较多)。对于资源竞争较少(线程冲突较轻)的情况，使用Synchronized同步锁进行线程阻塞和唤醒切换以及用户态内核态间的切换操作额外浪费消耗cpu资源。
独占锁属于悲观锁一类，synchronized就是一种独占锁，假设处于最坏的情况，只有一个线程执行，阻塞其他线程，如果并发高，处理耗时长，会导致多个线程挂起，等待持有锁的线程释放锁。
同步控制实现是基于Object的监视器：
Synchronized通过获取自增，释放自减的方式实现重入。
Synchronized致使线程阻塞，线程会进入到BLOCKED状态。

• synchronized需要放在在返回值前，如上在void前。可以在public之前也可以之后。
• 继承中子类覆盖父类方法，synchronized关键字特性不能继承传递，必须显式声明。
• 构造方法不能使用synchronized关键字，构造方法中支持同步代码块。
• 接口中方法，抽象方法也不支持synchronized关键字。

使用方式：

• 普通方法：锁是当前实例对象 ，进入同步代码前要获得当前实例的锁。
• 静态方法：锁是当前类的class对象，也就是锁的类的所有对象，进入同步代码前要获得当前类对象的锁。
• 代码块：锁是括号里面的对象，对给定对象加锁，进入同步代码库前要获得给定对象的锁。

class SynchronizedTest {

 public static synchronized void show() {//锁当前类
 }

 public synchronized void m1() {//锁实例对象
 }

 public void m3() {
  synchronized (this) {//锁实例对象
  }
 }

 public void m4() {
  synchronized (SynchronizedTest.class) {//锁当前类
  }
 }

 public void m5() {
  Object obj = new Object();
  synchronized (obj) {//锁特定对象
  }
 }
}

Semaphore

Semaphore类是一个计数信号量，必须由获取它的线程释放，通常用于限制可以访问某些资源（物理或逻辑的）线程数目。通过使用内部类Syn继承AQS实现。

    public class Semaphore implements java.io.Serializable {
        private static final long serialVersionUID = -3222578661600680210L;
        /** All mechanics via AbstractQueuedSynchronizer subclass */
        private final Sync sync;

        public Semaphore(int permits) {
            sync = new NonfairSync(permits);
        }

        public Semaphore(int permits, boolean fair) { //公平模式OR非公平模式
            sync = fair ? new FairSync(permits) : new NonfairSync(permits);
        }

        abstract static class Sync extends AbstractQueuedSynchronizer {
            private static final long serialVersionUID = 1192457210091910933L;
        }
    }

一个信号量有且仅有3种操作，且它们全部是原子的。

• 初始化、增加和减少。
• 增加可以为一个进程解除阻塞。
• 减少可以让一个进程进入阻塞。

acquire()方法阻塞，直到有一个许可证可以获得然后拿走一个许可证。
release()方法增加一个许可证，这可能会释放一个阻塞的acquire()方法。
Semaphore不使用实际的许可对象，只对可用许可的号码进行计数，并采取相应的行动。

• 在计数器不为0的时候对线程就放行，一旦达到0，那么所有请求资源的新线程都会被阻塞，包括增加请求到许可的线程，Semaphore是不可重入的。
• 每一次请求一个许可都会导致计数器减少1，同样每次释放一个许可都会导致计数器增加1，一旦达到0，新的许可请求线程将被挂起。

Semaphore有两种模式：

• 公平模式
  acquire 的顺序就是获取许可证的顺序，遵循FIFO。
• 非公平模式
  抢占式的，也就是有可能一个新的获取线程恰好在一个许可证释放时得到了这个许可证，而前面还有等待的线程。

    public class SemaphoreTest {
        private static final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
        private static Semaphore semaphore = new Semaphore(1);

        public static void main(String[] args) {
            for (int i = 1; i <= 5; i++) {
                Runnable t1 = new Task(i);
                threadPool.execute(t1);
            }
        }

        static class Task implements Runnable {
            private int id;

            Task(int id) {
                this.id = id;
            }

            @Override
            public void run() {
                try {
                    semaphore.acquire();
                    System.out.println("线程 " + this.id + "开始了");
                    Thread.sleep(20);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    semaphore.release();
                    System.out.println("线程 " + this.id + "释放了");
                }
            }
        }
    }

输出为：

线程 1开始了
线程 2开始了
线程 1释放了
线程 2释放了
线程 3开始了
线程 3释放了
线程 4开始了
线程 4释放了
线程 5开始了
线程 5释放了

LockSupport

是一个线程阻塞工具类，所有的方法都是静态方法，让线程在阻塞/唤醒。LockSupprt方法阻塞线程会致使线程进入到WAITING状态。
LockSupport使用方式和wait/notify很类似。LockSupport是以线程为单位进行阻塞和唤醒，wait/notify是以对象为单位进行阻塞和唤醒，也可以一次唤醒被这个对象阻塞的线程(notifyAll)。

public static void park(Object blocker); // 暂停当前线程
public static void parkNanos(Object blocker, long nanos); // 暂停当前线程，不过有超时时间的限制
public static void parkUntil(Object blocker, long deadline); // 暂停当前线程，直到某个时间
public static void park(); // 无期限暂停当前线程
public static void parkNanos(long nanos); // 暂停当前线程，不过有超时时间的限制
public static void parkUntil(long deadline); // 暂停当前线程，直到某个时间
public static void unpark(Thread thread); // 恢复当前线程
public static Object getBlocker(Thread t);

ThreadLocal

使用场合主要解决多线程中数据数据因并发产生不一致问题。
为每个线程的中并发访问的数据提供一个副本，通过访问副本来运行业务，这样的结果是耗费了内存，但大大减少了线程同步所带来性能消耗，也减少了线程并发控制的复杂度。
不能使用原子类型，只能使用Object类型。ThreadLocal的使用比Synchronized要简单得多。
ThreadLocal和Synchonized都用于解决多线程并发访问。但是ThreadLocal与Synchronized有本质的区别。处理不同的问题域。Synchronized用于线程间的数据共享，而ThreadLocal则用于线程间的数据隔离。

• Synchronized是利用锁的机制，使变量或代码块在某一时该只能被一个线程访问。
• ThreadLocal为每一个线程都提供了变量的副本，使得每个线程在某一时间访问到的并不是同一个对象，这样就隔离了多个线程对数据的数据共享。

• 每个Thread线程内部都有一个Map。
• Map里面存储线程本地对象(key)和线程的变量副本(value)
• Thread内部的Map是由ThreadLocal维护的，由ThreadLocal负责向map获取和设置线程的变量值。

    public T get() {
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null) {
            ThreadLocalMap.Entry e = map.getEntry(this);
            if (e != null) {
                @SuppressWarnings("unchecked")
                T result = (T)e.value;
                return result;
            }
        }
        return setInitialValue();
    }

    public void set(T value) {
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null)
            map.set(this, value);
        else
            createMap(t, value);
    }

    public void remove() {
        ThreadLocalMap m = getMap(Thread.currentThread());
        if (m != null)
            m.remove(this);
    }

Hibernate中getCurrentSession()使用ThreadLocal：

private static final ThreadLocal<Session> threadLocal = new ThreadLocal<Session>();

//获取Session
public static Session getCurrentSession(){
    Session session =  threadLocal.get();
    //判断Session是否为空，如果为空，将创建一个session，并设置到本地线程变量中
    try {
        if(session ==null && !session.isOpen()){
            if(sessionFactory == null){
                rbuildSessionFactory();// 创建Hibernate的SessionFactory
            }else{
                session = sessionFactory.openSession();
            }
        }
        threadLocal.set(session);
    } catch (Exception e) {
        // TODO: handle exception
    }

    return session;
}

Volatile

具有可见性、有序性，不具备原子性。不会让线程阻塞，响应速度比Synchronized高。

• 可见性：当多个线程访问同一个变量x时，线程1修改了变量x的值，线程1、线程2...线程n能够立即读取到线程1修改后的值。
• 有序性：即程序执行时按照代码书写的先后顺序执行。在Java内存模型中，允许编译器和处理器对指令进行重排序，但是重排序过程不会影响到单线程程序的执行，却会影响到多线程并发执行的正确性。(本文不对指令重排作介绍，但不代表它不重要，它是理解JAVA并发原理时非常重要的一个概念)。

当一个变量被定义为volatile之后，看做“程度较轻的 Synchronized”，具备两个特性：

• 保证此变量对所有线程的可见性(当一条线程修改这个变量值时，新值其他线程立即得知)。
• 禁止指令重新排序。

Volatile是不能保证原子性的，不能保证在并发条件下是线程安全的，因为java里面的运算并非原子操作。
适用于对变量的写操作不依赖于当前值，对变量的读取操作不依赖于非volatile变量。适用于读多写少的场景。可用作状态标志。
JDK中volatie应用：
ConcurrentHashMap的Entry的value和next被声明为volatile。

public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
    implements ConcurrentMap<K,V>, Serializable {

    static class Node<K,V> implements Map.Entry<K,V> {
        volatile V val;
        volatile Node<K,V> next;
        ...
    }
    ...
}

AtomicLong中的value被声明为volatile。AtomicLong通过CAS原理(也可以理解为乐观锁)保证了原子性。

public class AtomicLong extends Number implements java.io.Serializable {

    private volatile long value;

    public AtomicLong(long initialValue) {
        value = initialValue;
    }
 
 public final long get() {
        return value;
    }
 
 public final void set(long newValue) {
        value = newValue;
    }
}

自旋锁(spinlock)

是指当一个线程在获取锁的时候，如果锁已经被其它线程获取，那么该线程将循环等待，然后不断的判断锁是否能够被成功获取，直到获取到锁才会退出循环。
实现保护共享资源而提出一种锁机制。自旋锁与互斥锁比较类似，它们都是为了解决对某项资源的互斥使用。无论是互斥锁，还是自旋锁，在任何时刻，最多只能有一个保持者，也就说，在任何时刻最多只能有一个执行单元获得锁。但是两者在调度机制上略有不同：

• 互斥锁，如果资源已经被占用，资源申请者只能进入睡眠状态。
• 自旋锁不会引起调用者睡眠，如果自旋锁已经被别的执行单元保持，调用者就一直循环在那里看是否该自旋锁的保持者已经释放了锁，”自旋”一词就是因此而得名。

public class SpinLock {
    private AtomicReference<Thread> sign =new AtomicReference<>();
        public void lock(){
        Thread current = Thread.currentThread();
        while(!sign.compareAndSet(null, current)){ //
        }
    }

    public void unlock (){
        Thread current = Thread.currentThread();
        sign.compareAndSet(current, null);
    }
}

获取锁的线程一直处于活跃状态，但是并没有执行任何有效的任务，使用这种锁会造成busy-waiting。
自旋锁不会使线程状态发生切换，一直处于用户态，即线程一直都是active的；不会使线程进入阻塞状态，减少了不必要的上下文切换，执行速度快。
若某个线程持有锁的时间过长，就会导致其它等待获取锁的线程进入循环等待，消耗CPU。使用不当会造成CPU使用率极高。非自旋锁在获取不到锁的时候会进入阻塞状态，从而进入内核态，当获取到锁的时候需要从内核态恢复，需要线程上下文切换。 (线程被阻塞后便进入内核(Linux)调度状态，这个会导致系统在用户态与内核态之间来回切换，严重影响锁的性能)

• 在JDK1.6中，Java虚拟机提供-XX:+UseSpinning参数来开启自旋锁，使用-XX:PreBlockSpin参数来设置自旋锁等待的次数。
• 在JDK1.7开始，自旋锁的参数被取消，虚拟机不再支持由用户配置自旋锁，自旋锁总是会执行，自旋锁次数也由虚拟机自动调整。

CLH锁

CLH(Craig, Landin, and Hagersten locks)， 是一个自旋锁，能确保无饥饿性，提供先来先服务的公平性。
CLH锁也是一种基于链表的可扩展、高性能、公平的自旋锁，申请线程只在本地变量上自旋，它不断轮询前驱的状态，如果发现前驱释放了锁就结束自旋。

public class CLHLock {
    //指向最后加入的线程
    AtomicReference<Node> tail = new AtomicReference<Node>();
    //当前线程持有的节点,使用ThreadLocal实现了变量的线程隔离
    ThreadLocal<Node> node;
    //前驱节点，使用ThreadLocal实现了变量的线程隔离
    ThreadLocal<Node> preNode = new ThreadLocal<Node>();

    public CLHLock() {
        //初始化node
        node = new ThreadLocal<Node>() {
            //线程默认变量的值,如果不Override这个函数，默认值为null
            @Override
            protected Node initialValue() {
                return new Node();
            }
        };
        //初始化tail，指向一个node，类似一个head节点，并且该节点locked属性为false
        tail.set(new Node());
    }

    public void lock() {
        //因为上面提到的构造函数中initialValue()方法，所以每个线程会有一个默认的值
        //并且node的locked属性为false.
        Node myNode = node.get();
        //修改为true，表示需要获取锁
        myNode.locked = true;
        //获取这之前最后加入的线程，并把当前加入的线程设置为tail，
        // AtomicReference的getAndSet操作是原子性的
        Node preNode = tail.getAndSet(myNode);
        //设置当前节点的前驱节点
        this.preNode.set(preNode);
        //轮询前驱节点的locked属性，尝试获取锁
        while (preNode.locked) {

        }
    }

    public void unlock() {
        //解锁很简单，将节点locked属性设置为false，
        //这样轮询该节点的另一个线程可以获取到释放的锁
        node.get().locked = false;
        //当前节点设置为前驱节点，也就是上面初始化提到的head节点
        node.set(preNode.get());
    }

    private class Node{
        //默认不需要锁
        private boolean locked = false;
    }
}

MCS锁

MCS来自于其发明人名字的首字母：John Mellor-Crummey和Michael Scott。
MCS锁可以解决上面的CLH锁的缺点，是一种基于链表的可扩展、高性能、公平的自旋锁，申请线程只在本地变量上自旋，直接前驱负责通知其结束自旋（与CLH自旋锁不同的地方，不在轮询前驱的状态，而是由前驱主动通知），从而极大地减少了不必要的处理器缓存同步的次数，降低了总线和内存的开销。

public class MCSLock {
    public static class MCSNode {
        //持有后继者的引用
        MCSNode next;
        // 默认是在等待锁
        boolean locked = true;
    }

    volatile MCSNode tail;// 指向最后一个申请锁的MCSNode
    private static final AtomicReferenceFieldUpdater<MCSLock, MCSNode> UPDATER = AtomicReferenceFieldUpdater
            .newUpdater(MCSLock.class, MCSNode.class, "tail");

    public void lock(MCSNode currentThreadMcsNode) {
        //更新tial为最新加入的线程节点，并取出之前的节点（也就是前驱）
        MCSNode predecessor = UPDATER.getAndSet(this, currentThreadMcsNode);//step4
        //前驱为空表示没有线程占用锁
        if (predecessor != null) {
            //将当前节点设置为前驱节点的后继者
            predecessor.next = currentThreadMcsNode;//step5
            //轮询自己的isLocked属性
            while (currentThreadMcsNode.locked) {

            }
        }
    }

    public void unlock(MCSNode currentThreadMcsNode) {
        //UPDATER.get(this) 获取最后加入的线程的node
        //如果获取到的最后加入的node和当前node（currentThreadMcsNode）不相同，表示还有其他线程等待锁，直接修改后继者的isLocked属性。
        //相同代表当前没其他有线程等待锁，进入下面的处理
        if (UPDATER.get(this) == currentThreadMcsNode) {//step1
            //这个时候可能会有其他线程又加入了进来，检查时候有人排在自己后面，currentThreadMcsNode.next 表示依然没有染排在自己后面
            if (currentThreadMcsNode.next == null) { //step2
                //将tail设置为空，如果返回true设置成功，如果返回false，表示设置失败（其他线程加入了进来，使得当前tail持有的节点不等于currentThreadMcsNode）
                if (UPDATER.compareAndSet(this, currentThreadMcsNode, null)) {// //step3
                    // 设置成功返回，没有其他线程等待锁
                    return;
                } else {
                    // 突然有其他线程加入，需要检测后继者是否有值，因为：step4执行完后，step5可能还没执行完
                    while (currentThreadMcsNode.next == null) {
                    }
                }
            }
            //修改后继者的isLocked,通知后继者结束自旋
            currentThreadMcsNode.next.locked = false;
            currentThreadMcsNode.next = null;// for GC
        }
    }
}

CAS(Compare and Swap)

比较并替换。CAS属于乐观锁，乐观地认为程序中的并发情况不那么严重，所以让线程不断去重试更新。用于写比较少的情况下(多读场景，冲突一般较少)。
CAS机制中使用了3个基本操作数：

• 需要读写的内存值V
• 旧的预期值A
• 要修改的新值B

更新一个变量的时候，只有当变量旧的预期值A和内存地址V当中的实际值相同时，才会将内存地址V对应的值修改为B。
CAS的缺点：

• CPU开销过大，在并发量比较高的情况下，如果许多线程反复尝试更新某一个变量，却又一直更新不成功，循环往复，会给CPU带来很到的压力。
• 不能保证代码块的原子性
  CAS机制所保证的只是一个变量的原子性操作，而不能保证整个代码块的原子性。比如需要保证3个变量共同进行原子性的更新，就不得不使用Synchronized了。
• ABA问题
  这是CAS机制最大的问题所在。CAS算法实现一个重要前提需要取出内存中某时刻的数据，而在下时刻比较并替换，那么在这个时间差类会导致数据的变化。
  通过增加版本号，可以解决ABA问题，AtomicStampedReference类就实现了用版本号作比较CAS机制。

ABA

如果另一个线程修改V值假设原来是A，先修改成B，再修改回成A。当前线程的CAS操作无法分辨当前V值是否发生过变化。

public class AtomicStampedReference<V> {
  /**
     * Atomically sets the value of both the reference and stamp
     * to the given update values if the
     * current reference is {@code ==} to the expected reference
     * and the current stamp is equal to the expected stamp.
     *
     * @param expectedReference the expected value of the reference
     * @param newReference the new value for the reference
     * @param expectedStamp the expected value of the stamp
     * @param newStamp the new value for the stamp
     * @return {@code true} if successful
     */
    public boolean compareAndSet(V   expectedReference,
                                 V   newReference,
                                 int expectedStamp,
                                 int newStamp) {
        Pair<V> current = pair;
        return
            expectedReference == current.reference &&
            expectedStamp == current.stamp &&
            ((newReference == current.reference &&
              newStamp == current.stamp) ||
             casPair(current, Pair.of(newReference, newStamp)));
    }
}

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {

        /**
        * Atomically sets synchronization state to the given updated
        * value if the current state value equals the expected value.
        * This operation has memory semantics of a {@code volatile} read
        * and write.
        *
        * @param expect the expected value
        * @param update the new value
        * @return {@code true} if successful. False return indicates that the actual
        *         value was not equal to the expected value.
        */
        protected final boolean compareAndSetState(int expect, int update) {
            // See below for intrinsics setup to support this
            return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
        }
}

Tips

Java语言不像C，C++那样可以直接访问底层操作系统，但是JVM为我们提供了一个后门，这个后门就是unsafe。unsafe为我们提供了硬件级别的原子操作。
valueOffset对象，是通过unsafe.objectFiledOffset方法得到，所代表的是AtomicLong对象value成员变量在内存中的偏移量。我们可以简单的把valueOffset理解为value变量的内存地址。
unsafe的compareAndSwapLong方法的参数包括了这三个基本元素：valueOffset参数代表了V，expect参数代表了A，update参数代表了B。
正是unsafe的compareAndSwapLong方法保证了Compare和Swap操作之间的原子性操作。

public class AtomicLong extends Number implements java.io.Serializable {

    private static final Unsafe unsafe = Unsafe.getUnsafe();
    private static final long valueOffset;

    static {
        try {
            valueOffset = unsafe.objectFieldOffset
                (AtomicLong.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }

    public final long getAndSet(long newValue) { 
        while (true) { 
            long current = get(); 
            if (compareAndSet(current, newValue)) 
                return current; 
        } 
    } 
    public final boolean compareAndSet(long expect, long update) { 
        return unsafe.compareAndSwapLong(this, valueOffset, expect, update); 
    }
}

版本号机制

一般是在数据表中加上一个数据版本号version字段，表示数据被修改的次数，当数据被修改时，version值会加一。当线程A要更新数据值时，在读取数据的同时也会读取version值，在提交更新时，若刚才读取到的version值为当前数据库中的version值相等时才更新，否则重试更新操作，直到更新成功。

AbstractQueuedSynchronizer(AQS)

public abstract class AbstractQueuedSynchronizer
        extends AbstractOwnableSynchronizer
        implements java.io.Serializable {
}

AQS是java中管理“锁”的抽象类，锁的许多公共方法都是在这个类中实现。AQS是独占锁(例如ReentrantLock)和共享锁(例如Semaphore)的公共父类。
AQS提供了一个基于FIFO队列，可以用于构建锁或者其他相关同步装置的基础框架。底层的数据结构是使用双向链表，是队列的一种实现，故也可看成是队列，其中Sync queue，即同步队列，是双向链表，包括head结点和tail结点，head结点主要用作后续的调度。而Condition queue不是必须的，其是一个单向链表，只有当使用Condition时，才会存在此单向链表。并且可能会有多个Condition queue。
AQS里面的CLH队列是CLH同步锁的一种变形。其主要从两方面进行了改造：节点的结构与节点等待机制。在结构上引入了头结点和尾节点，他们分别指向队列的头和尾，尝试获取锁、入队列、释放锁等实现都与头尾节点相关，并且每个节点都引入前驱节点和后后续节点的引用；在等待机制上由原来的自旋改成阻塞唤醒。
核心是通过一个共享变量来同步状态，变量的状态由子类去维护，而AQS框架做的是：

• 线程阻塞队列的维护
• 线程阻塞和唤醒

有两个内部类，分别为Node类与ConditionObject类。 每个线程被阻塞的线程都会被封装成一个Node结点，放入队列。每个节点包含了一个Thread类型的引用，并且每个节点都存在一个状态。

Node

static final class Node {
        static final Node SHARED = new Node();  //共享模式
        static final Node EXCLUSIVE = null; //独占模式
        // 等于0表示当前节点在sync queue中，等待着获取锁
        static final int CANCELLED =  1; //表示当前的线程被取消。
        static final int SIGNAL    = -1; //L表示当前节点的后继节点包含的线程需要运行，需要进行unpark操作
        static final int CONDITION = -2; //表示当前节点在等待condition，也就是在condition queue中
        static final int PROPAGATE = -3; //PROPAGATE表示当前场景下后续的acquireShared能够得以执行
        volatile int waitStatus; //结点状态
        volatile Node prev; //前驱结点
        volatile Node next; //后继结点
        volatile Thread thread; //结点所对应的线程
        Node nextWaiter; //下一个等待者

        final boolean isShared() {
            return nextWaiter == SHARED;
        }

        final Node predecessor() throws NullPointerException {//获取前驱结点，若前驱结点为空，抛出异常
            Node p = prev;
            if (p == null)
                throw new NullPointerException();
            else
                return p;
        }

        Node() {    // Used to establish initial head or SHARED marker
        }

        Node(Thread thread, Node mode) {     // Used by addWaiter
            this.nextWaiter = mode;
            this.thread = thread;
        }

        Node(Thread thread, int waitStatus) { // Used by Condition
            this.waitStatus = waitStatus;
            this.thread = thread;
        }
    }

ConditionObject

public class ConditionObject implements Condition, java.io.Serializable {
        private static final long serialVersionUID = 1173984872572414699L;
        /** First node of condition queue. */
        private transient Node firstWaiter;
        /** Last node of condition queue. */
        private transient Node lastWaiter;

        public ConditionObject() { }

        private Node addConditionWaiter() {
            Node t = lastWaiter;
            // If lastWaiter is cancelled, clean out.
            if (t != null && t.waitStatus != Node.CONDITION) {
                unlinkCancelledWaiters();
                t = lastWaiter;
            }
            Node node = new Node(Thread.currentThread(), Node.CONDITION);
            if (t == null)
                firstWaiter = node;
            else
                t.nextWaiter = node;
            lastWaiter = node;
            return node;
        }

        private void doSignal(Node first) {
            do {
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                first.nextWaiter = null;
            } while (!transferForSignal(first) &&
                     (first = firstWaiter) != null);
        }

        private void doSignalAll(Node first) {
            lastWaiter = firstWaiter = null;
            do {
                Node next = first.nextWaiter;
                first.nextWaiter = null;
                transferForSignal(first);
                first = next;
            } while (first != null);
        }

        private void unlinkCancelledWaiters() {
            Node t = firstWaiter;
            Node trail = null;
            while (t != null) {
                Node next = t.nextWaiter;
                if (t.waitStatus != Node.CONDITION) {
                    t.nextWaiter = null;
                    if (trail == null)
                        firstWaiter = next;
                    else
                        trail.nextWaiter = next;
                    if (next == null)
                        lastWaiter = trail;
                }
                else
                    trail = t;
                t = next;
            }
        }

        public final void signal() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignal(first);
        }

        public final void signalAll() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignalAll(first);
        }

        public final void awaitUninterruptibly() {
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            boolean interrupted = false;
            while (!isOnSyncQueue(node)) {
                LockSupport.park(this);
                if (Thread.interrupted())
                    interrupted = true;
            }
            if (acquireQueued(node, savedState) || interrupted)
                selfInterrupt();
        }

        /** Mode meaning to reinterrupt on exit from wait */
        private static final int REINTERRUPT =  1;
        /** Mode meaning to throw InterruptedException on exit from wait */
        private static final int THROW_IE    = -1;

        private int checkInterruptWhileWaiting(Node node) {
            return Thread.interrupted() ?
                (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
                0;
        }

        private void reportInterruptAfterWait(int interruptMode)
            throws InterruptedException {
            if (interruptMode == THROW_IE)
                throw new InterruptedException();
            else if (interruptMode == REINTERRUPT)
                selfInterrupt();
        }

        public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                LockSupport.park(this);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

        public final long awaitNanos(long nanosTimeout)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            final long deadline = System.nanoTime() + nanosTimeout;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (nanosTimeout <= 0L) {
                    transferAfterCancelledWait(node);
                    break;
                }
                if (nanosTimeout >= spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
                nanosTimeout = deadline - System.nanoTime();
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return deadline - System.nanoTime();
        }

        //等待，当前线程在接到信号、被中断或到达指定最后期限之前一直处于等待状态
        public final boolean awaitUntil(Date deadline)
                throws InterruptedException {
            long abstime = deadline.getTime();
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            boolean timedout = false;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (System.currentTimeMillis() > abstime) {
                    timedout = transferAfterCancelledWait(node);
                    break;
                }
                LockSupport.parkUntil(this, abstime);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return !timedout;
        }

        //等待，当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。
        //此方法在行为上等效于：awaitNanos(unit.toNanos(time)) > 0
        public final boolean await(long time, TimeUnit unit)
                throws InterruptedException {
            long nanosTimeout = unit.toNanos(time);
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            final long deadline = System.nanoTime() + nanosTimeout;
            boolean timedout = false;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (nanosTimeout <= 0L) {
                    timedout = transferAfterCancelledWait(node);
                    break;
                }
                if (nanosTimeout >= spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
                nanosTimeout = deadline - System.nanoTime();
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return !timedout;
        }

        final boolean isOwnedBy(AbstractQueuedSynchronizer sync) {
            return sync == AbstractQueuedSynchronizer.this;
        }

        //查询是否有正在等待此条件的任何线程
        protected final boolean hasWaiters() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
                if (w.waitStatus == Node.CONDITION)
                    return true;
            }
            return false;
        }

        //返回正在等待此条件的线程数估计值
        protected final int getWaitQueueLength() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            int n = 0;
            for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
                if (w.waitStatus == Node.CONDITION)
                    ++n;
            }
            return n;
        }

        //返回包含那些可能正在等待此条件的线程集合
        protected final Collection<Thread> getWaitingThreads() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            ArrayList<Thread> list = new ArrayList<Thread>();
            for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
                if (w.waitStatus == Node.CONDITION) {
                    Thread t = w.thread;
                    if (t != null)
                        list.add(t);
                }
            }
            return list;
        }
    }

Lock

    /**
     * The synchronization state.
     */
    private volatile int state;

    protected final int getState() {
        return state;
    }

    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

    private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode); //新生成一个结点，默认为独占模式
        // Try the fast path of enq; backup to full enq on failure
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {//比较pred是否为尾结点，是则将尾结点设置为node 
                pred.next = node; //设置尾结点的next域为node
                return node; //返回新生成的结点
            }
        }
        enq(node); //尾结点为空(即还没有被初始化过)，或者是compareAndSetTail操作失败，则入队列
        return node;
    }

    // 入队列
    private Node enq(final Node node) {
        for (;;) { // 无限循环，确保结点能够成功入队列
            Node t = tail;  //保存尾结点
            if (t == null) { // 尾结点为空，即还没被初始化
                if (compareAndSetHead(new Node())) // 头结点为空，并设置头结点为新生成的结点
                    tail = head; // 头结点与尾结点都指向同一个新生结点
            } else { // 尾结点不为空，即已经被初始化过
                // 将node结点的prev域连接到尾结点
                node.prev = t; 
                if (compareAndSetTail(t, node)) { // 比较结点t是否为尾结点，若是则将尾结点设置为node
                    // 设置尾结点的next域为node
                    t.next = node; 
                    return t; //返回尾结点
                }
            }
        }
    }

• 调用tryAcquire函数，调用此方法的线程会试图在独占模式下获取对象状态。此方法应该查询是否允许它在独占模式下获取对象状态，如果允许，则获取它。在AbstractQueuedSynchronizer源码中默认会抛出一个异常，即需要子类去重写此函数完成自己的逻辑。
• 若tryAcquire失败，则调用addWaiter函数，addWaiter函数完成的功能是将调用此方法的线程封装成为一个结点并放入Sync queue。
• 调用acquireQueued函数，此函数完成的功能是Sync queue中的结点不断尝试获取资源，若成功，则返回true，否则，返回false。

ReentrantLock

ReentrantLock重入锁，是实现Lock接口的一个类，支持重入性，表示能够对共享资源能够重复加锁，即当前线程获取该锁再次获取不会被阻塞。

要想支持重入性，就要解决两个问题：

1. 在线程获取锁的时候，如果已经获取锁的线程是当前线程的话则直接再次获取成功。
2. 由于锁会被获取n次，那么只有锁在被释放同样的n次之后，该锁才算是完全释放成功。

Sync是AQS的子类。Sync有两个子类FairSync、NonFairSync。

Sync

    abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = -5179523762034025860L;
        abstract void lock();

        final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState(); //获取父类AQS中的标志位 
            if (c == 0) {//值为0,那么当前独占性变量还未被线程占有
                if (compareAndSetState(0, acquires)) {//如果通过CAS操作将状态为更新成功则代表当前线程获取锁
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                // 如果不为0意味着，锁已经被拿走了，但是，因为 ReentrantLock 是重入锁，是可以重复lock,unlock的，只要成对出现
                int nextc = c + acquires; //累加state
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }

        protected final boolean tryRelease(int releases) {
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            if (c == 0) {
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }

        protected final boolean isHeldExclusively() {
            // While we must in general read state before owner,
            // we don't need to do so to check if current thread is owner
            return getExclusiveOwnerThread() == Thread.currentThread();
        }

        final ConditionObject newCondition() {
            return new ConditionObject();
        }

        // Methods relayed from outer class

        final Thread getOwner() {
            return getState() == 0 ? null : getExclusiveOwnerThread();
        }

        final int getHoldCount() {
            return isHeldExclusively() ? getState() : 0;
        }

        final boolean isLocked() {
            return getState() != 0;
        }

        /**
         * Reconstitutes the instance from a stream (that is, deserializes it).
         */
        private void readObject(java.io.ObjectInputStream s)
            throws java.io.IOException, ClassNotFoundException {
            s.defaultReadObject();
            setState(0); // reset to unlocked state
        }
    }

NonFairSync(非公平锁)

    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;

        /**
         * Performs lock.  Try immediate barge, backing up to normal
         * acquire on failure.
         */
        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }

        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }

FairSync(公平锁)

    static final class FairSync extends Sync {
        private static final long serialVersionUID = -3000897897090466540L;

        final void lock() {
            acquire(1);
        }

        /**
         * Fair version of tryAcquire.  Don't grant access unless
         * recursive call or no waiters or is first.
         */
        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();  //获取父类AQS中的标志位 
            if (c == 0) {//值为0,那么当前独占性变量还未被线程占有
                if (!hasQueuedPredecessors() && //如果队列中没有其他线程，说明没有线程正在占有锁
                    compareAndSetState(0, acquires)) { //如果通过CAS操作将状态为更新成功则代表当前线程获取锁
                    setExclusiveOwnerThread(current);                   
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    }

公平锁和非公平锁的处理逻辑基本上一致，唯一的不同在于增加了hasQueuedPredecessors()的逻辑判断，用来判断当前节点在同步队列中是否有前驱节点的判断，如果有前驱节点说明有线程比当前线程更早的请求资源，根据公平性，当前线程请求资源失败。如果当前节点没有前驱节点的话，再才有做后面的逻辑判断的必要性。
公平锁每次都是从同步队列中的第一个节点获取到锁，而非公平性锁则不一定，有可能刚释放锁的线程能再次获取到锁。

• 公平锁每次获取到锁为同步队列中的第一个节点，保证请求资源时间上的绝对顺序，而非公平锁有可能刚释放锁的线程下次继续获取该锁，则有可能导致其他线程永远无法获取到锁，造成“饥饿”现象。
• 公平锁为了保证时间上的绝对顺序，需要频繁的上下文切换，而非公平锁会降低一定的上下文切换，降低性能开销。因此，ReentrantLock默认选择的是非公平锁，则是为了减少一部分上下文切换，保证了系统更大的吞吐量。

public class ReentrantLock implements Lock, java.io.Serializable {

    private final Sync sync;

    public ReentrantLock() {
        sync = new NonfairSync();
    }
    
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

    // 尝试获取锁,立即返回获取结果,轮询锁 
    public boolean tryLock() {
        return sync.nonfairTryAcquire(1);
    }
    //尝试获取锁,最多等待timeout时长,超时锁 
    public boolean tryLock(long timeout, TimeUnit unit)
            throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }
    //可中断锁,调用线程interrupt方法,则锁方法抛出InterruptedException,中断锁
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }
}

锁优化

• 减少锁持有时间

• 减小锁粒度
• 锁分离(ReadWriteLock)
• 锁粗化
• 锁消除