Java中的锁原理–AQS
- 2019 年 12 月 4 日
- 筆記
大家或多或少会接触一些线程安全问题，什么是线程安全？
通俗的来讲，某个函数被多个线程调用多次，都能够处理各个线程中的局部变量，并且计算结果正确，我们一般称为线程安全。
如何解决线程安全问题？
一般有三种方式
- 使用 ThreadLocal 避免线程共享变量
- 使用 synchronized 和 Lock 进行同步控制。
- 使用原子变量声明变量。
Lock 的实现原理是什么？
AQS(AbstracctQueuedSynchronized) 队列同步器，是用来构建锁或者其他同步器组件的基础框架。
AQS 使用了一个 int 变量来表示同步状态，通过内置的 FIFO 队列来完成资源获取线程的排队工作。
经常使用的同步组件ReentrantLock、ReentrantReadWriteLock和 CountDownLatch 等都是基于同步器实现的。
AQS 主要包含两点，一个是同步状态，第二个是队列。
AQS 是怎么实现线程同步的？主要包括同步队列、独占是同步状态的释放和获取、共享式同步状态的释放和获取。
同步器依赖的是同步队列的来进行同步状态的管理。
同步队列的结构
队列中的节点 Node 是构成同步器的基础。
static final class Node { /** Marker to indicate a node is waiting in shared mode */ static final Node SHARED = new Node(); /** Marker to indicate a node is waiting in exclusive mode */ static final Node EXCLUSIVE = null; /** waitStatus value to indicate thread has cancelled */ static final int CANCELLED = 1; /** waitStatus value to indicate successor's thread needs unparking */ static final int SIGNAL = -1; /** waitStatus value to indicate thread is waiting on condition */ static final int CONDITION = -2; /** * waitStatus value to indicate the next acquireShared should * unconditionally propagate */ static final int PROPAGATE = -3; /** * Status field, taking on only the values: * SIGNAL: The successor of this node is (or will soon be) * blocked (via park), so the current node must * unpark its successor when it releases or * cancels. To avoid races, acquire methods must * first indicate they need a signal, * then retry the atomic acquire, and then, * on failure, block. * CANCELLED: This node is cancelled due to timeout or interrupt. * Nodes never leave this state. In particular, * a thread with cancelled node never again blocks. * CONDITION: This node is currently on a condition queue. * It will not be used as a sync queue node * until transferred, at which time the status * will be set to 0. (Use of this value here has * nothing to do with the other uses of the * field, but simplifies mechanics.) * PROPAGATE: A releaseShared should be propagated to other * nodes. This is set (for head node only) in * doReleaseShared to ensure propagation * continues, even if other operations have * since intervened. * 0: None of the above * * The values are arranged numerically to simplify use. * Non-negative values mean that a node doesn't need to * signal. So, most code doesn't need to check for particular * values, just for sign. * * The field is initialized to 0 for normal sync nodes, and * CONDITION for condition nodes. It is modified using CAS * (or when possible, unconditional volatile writes). */ volatile int waitStatus; /** * Link to predecessor node that current node/thread relies on * for checking waitStatus. Assigned during enqueuing, and nulled * out (for sake of GC) only upon dequeuing. Also, upon * cancellation of a predecessor, we short-circuit while * finding a non-cancelled one, which will always exist * because the head node is never cancelled: A node becomes * head only as a result of successful acquire. A * cancelled thread never succeeds in acquiring, and a thread only * cancels itself, not any other node. */ volatile Node prev; /** * Link to the successor node that the current node/thread * unparks upon release. Assigned during enqueuing, adjusted * when bypassing cancelled predecessors, and nulled out (for * sake of GC) when dequeued. The enq operation does not * assign next field of a predecessor until after attachment, * so seeing a null next field does not necessarily mean that * node is at end of queue. However, if a next field appears * to be null, we can scan prev's from the tail to * double-check. The next field of cancelled nodes is set to * point to the node itself instead of null, to make life * easier for isOnSyncQueue. */ volatile Node next; /** * The thread that enqueued this node. Initialized on * construction and nulled out after use. */ volatile Thread thread; /** * Link to next node waiting on condition, or the special * value SHARED. Because condition queues are accessed only * when holding in exclusive mode, we just need a simple * linked queue to hold nodes while they are waiting on * conditions. They are then transferred to the queue to * re-acquire. And because conditions can only be exclusive, * we save a field by using special value to indicate shared * mode. */ Node nextWaiter; /** * Returns true if node is waiting in shared mode. */ final boolean isShared() { return nextWaiter == SHARED; } /** * Returns previous node, or throws NullPointerException if null. * Use when predecessor cannot be null. The null check could * be elided, but is present to help the VM. * * @return the predecessor of this node */ 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; } }
Node 的构造方法可以看到，包含了线程 Thread 和 状态 waitStatus 或者 Thread 和 nextWaiter(Node) 。
/** 值为1，由于同步队列中等待的线程超时或者被中断，需要到同步队列中取消等待，节点进入该状态将不会变*/ static final int CANCELLED = 1; /**后继节点的线程处于阻塞状态，而如果当前节点的线程如果释放同步状态或者被取消，通知后继节点，使得后继节点可以运行*/ static final int SIGNAL = -1; /** 值为-2 节点在等待队列中，节点线程等待在Condition上，当其他线程对 Condition 调用了 signal() 方法后，该节点会从等待队列转移到同步队列中，进行同步状态的获取 */ static final int CONDITION = -2;
节点加入到同步队列
同步器拥有首节点 head 和 尾节点 tail 没有成功获取同步状态的线程将会组成Node 加入该队列的尾部。这个加入队尾的过程需要是线程安全的。同步器提供了一个基于 CAS 的设置尾节点的方法 compareAndSetTail(Node expt, Node update) 需要传递当前线程认为的尾节点 expt 和当前节点 update。
为什么 CAS 能够保证线程安全？
java 中的 CAS 是对 cmpxchg 的封装。
cmpxchg 中x86 中有 CAS 指令。 cmpxchg是汇编指令 作用：比较并交换操作数. 如：CMPXCHG r/m,r 将累加器AL/AX/EAX/RAX中的值与首操作数（目的操作数）比较，如果相等，第2操作数（源操作数）的值装载到首操作数，zf置1。如果不等， 首操作数的值装载到AL/AX/EAX/RAX并将zf清0 该指令只能用于486及其后继机型。第2操作数（源操作数）只能用8位、16位或32位寄存器。第1操作数（目地操作数）则可用寄存器或任一种存储器寻址方式
cmpxchg 功能就是保证一次只原子性的修改一个变量。
线程释放同步状态，节点出队
首节点的线程在释放同步状态时，将会唤醒后继节点。而后继节点将会在获取同步状态时，将自己设置成首节点。
设置首节点是通过获取同步状态成功的线程来完成的，由于只有一个线程能够成功的获取同步状态，因此，不需要使用 CAS 来保证只需要将首节点的后继节点设置成首节点即可。
