 我们都知道`StringBuffer`是线程安全的，因为他的关键方法都加了`synchronized`，但是，从打印结果可以看出，锁被消除了。因为`buffer`这个引用只会在`main`方法中使用，不可能被其他线程引用(因为是局部变量，栈私有)，所以`buffer`是不可能共享的资源，`JVM`会自动消除`StringBuffer`对象内部的锁。 ## 锁粗化
/**
- @author yhd
- @createtime 2020/9/8 20:48
*/
public class Demo3 {
public static void main(String[] args) {
int i=0;
StringBuffer buffer = new StringBuffer();
while (i<100){
buffer.append(i);
i++;
}
System.out.println(buffer.toString());
System.out.println(ClassLayout.parseInstance(buffer).toPrintable());
}
}
`JVM`会检测到这样一连串的操作都对同一个对象加锁（`while` 循环内 100 次执行 `append`，没有锁粗化的就要进行 100 次加锁/解锁），此时 `JVM` 就会将加锁的范围粗化到这一连串的操作的外部（比如 `while` 虚幻体外），使得这一连串操作只需要加一次锁即可。 ## AQS 研究了`AQS`一天，终于找到了他的入口，接下来看我的想法： ## 多线程操做共享数据问题
/**
- @author yhd
- @createtime 2020/9/8 8:11
*/
public class Demo1 {
public static int m=0;
public static void main(String[] args)throws Exception {
Thread []threads=new Thread[100];
for (int i = 0; i < threads.length; i++) {
threads[i]=new Thread(()->{
for (int j = 0; j < 100; j++) {
m++;
}
});
}
for (Thread t :threads) t.start();
for (Thread t :threads) t.join();//线程顺序结束
System.out.println(m);
}
}
毫无疑问，这段代码是存在线程安全问题的，只要了解一点并发编程，都是可以看出来的。那么我们可以怎么来解决呢？ **使用synchronized来解决**
/**
- @author yhd
- @createtime 2020/9/8 8:32
*/
public class Demo2 {
public static int m=0;
public static void main(String[] args)throws Exception {
Thread []threads=new Thread[100];
for (int i = 0; i < threads.length; i++) {
threads[i]=new Thread(()->{
synchronized (Demo2.class) {
for (int j = 0; j < 100; j++) {
m++;
}
}
});
}
for (Thread t :threads) t.start();
for (Thread t :threads) t.join();//线程顺序结束
System.out.println(m);
}
这样解决了线程安全的问题，但是同时也存在一个问题，synchronized是操作系统层面的方法，也就是需要jvm和操做系统之间进行一个切换(用户态和内核态的切换)，这样实际上是影响效率的。另一种解决办法： **使用ReentrantLock来解决**
/**
- @author yhd
- @createtime 2020/9/8 8:41
*/
public class Demo3 {
public static int m=0;
public static Lock lock=new ReentrantLock();
public static void main(String[] args)throws Exception {
Thread []threads=new Thread[100];
for (int i = 0; i < threads.length; i++) {
threads[i]=new Thread(()->{
try { lock.lock(); for (int j = 0; j < 100; j++) { m++; } } finally { lock.unlock(); } }); } for (Thread t :threads) t.start(); for (Thread t :threads) t.join();//线程顺序结束 System.out.println(m);
- }
}
那么这种方式的底层是怎么做的呢？接下来追源码。
public ReentrantLock() { sync = new NonfairSync(); }
这个sync又是啥呢？接着追
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); } }
它实际上是`ReentrantLock`的一个内部类继承了`Sync`，而他里面的方法实际上也是调用了`Sync`的方法。这样目标就明确了，我们可以看一下`Sync`。 这个`Sync`的源码：
abstract static class Sync extends AbstractQueuedSynchronizer
由此可知，实际上`ReentrantLock`是利用`AbstractQueuedSynchronizer`也就是`AQS`来实现的。 ## AbstractQueuedSynchronizer概述 这个类的内部有一个内部类`Node`
static final class Node {
static final Node SHARED = new Node();
volatile Node prev;//前驱指针
volatile Node next;//后继指针
volatile Thread thread;//当前线程
private transient volatile Node head;//头节点
private transient volatile Node tail;//尾节点
private volatile int state;//锁状态，加锁成功则为1，重入+1 解锁则为0
…
}
看到这里就想到了`LinkedHashMap`，实际上也就是类似于维持了一个双向的链表，每个节点都是一个线程。  它维护了一个volatile int state（代表共享资源）和一个FIFO线程等待队列（多线程争用资源被阻塞时会进入此队列）。这里volatile是核心关键词，具体volatile的语义，在此不述。state的访问方式有三种:
getState() setState() compareAndSetState()
`AQS`定义两种资源共享方式：`Exclusive`（独占，只有一个线程能执行，如`ReentrantLock`）和`Share`（共享，多个线程可同时执行，如`Semaphore/CountDownLatch`）。 不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可，至于具体线程等待队列的维护（如获取资源失败入队/唤醒出队等），AQS已经在顶层实现好了。自定义同步器实现时主要实现以下几种方法：
isHeldExclusively()：该线程是否正在独占资源。只有用到condition才需要去实现它。
tryAcquire(int)：独占方式。尝试获取资源，成功则返回true，失败则返回false。
tryRelease(int)：独占方式。尝试释放资源，成功则返回true，失败则返回false。
tryAcquireShared(int)：共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
tryReleaseShared(int)：共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回true，否则返回false。
以`ReentrantLock`为例，state初始化为0，表示未锁定状态。A线程lock()时，会调用`tryAcquire()`独占该锁并将state+1。此后，其他线程再`tryAcquire()`时就会失败，直到A线程unlock()到state=0（即释放锁）为止，其它线程才有机会获取该锁。当然，释放锁之前，A线程自己是可以重复获取此锁的（state会累加），这就是可重入的概念。但要注意，获取多少次就要释放多么次，这样才能保证state是能回到零态的。 再以`CountDownLatch`以例，任务分为N个子线程去执行，state也初始化为N（注意N要与线程个数一致）。这N个子线程是并行执行的，每个子线程执行完后`countDown()`一次，state会`CAS`减1。等到所有子线程都执行完后(即state=0)，会`unpark()`主调用线程，然后主调用线程就会从await()函数返回，继续后余动作。 一般来说，自定义同步器要么是独占方法，要么是共享方式，他们也只需实现`tryAcquire-tryRelease`、`tryAcquireShared-tryReleaseShared`中的一种即可。但AQS也支持自定义同步器同时实现独占和共享两种方式，如`ReentrantReadWriteLock`。 ## 源码解读与执行流程分析 ## 独占锁 acquire(int)== 此方法是独占模式下线程获取共享资源的顶层入口。如果获取到资源，线程直接返回，否则进入等待队列，直到获取到资源为止，且整个过程忽略中断的影响。这也正是lock()的语义，当然不仅仅只限于lock()。获取到资源后，线程就可以去执行其临界区代码了。
public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
函数流程如下： `tryAcquire()`尝试直接去获取资源，如果成功则直接返回； `addWaiter()`将该线程加入等待队列的尾部，并标记为独占模式； `acquireQueued()`使线程在等待队列中获取资源，一直获取到资源后才返回。如果在整个等待过程中被中断过，则返回true，否则返回false。如果线程在等待过程中被中断过，它是不响应的。只是获取资源后才再进行自我中断selfInterrupt()，将中断补上。 **tryAcquire(int)** 此方法尝试去获取独占资源。如果获取成功，则直接返回true，否则直接返回false。这也正是tryLock()的语义，还是那句话，当然不仅仅只限于tryLock()。
protected boolean tryAcquire(int arg) { throw new UnsupportedOperationException(); }
一开始还傻乎乎的想为啥直接抛异常了，后来才反应过来，这不是给自定义的方法么？AQS这里只定义了一个接口，具体资源的获取交由自定义同步器去实现了。 **addWaiter(Node)** 此方法用于将当前线程加入到等待队列的队尾，并返回当前线程所在的结点。
private Node addWaiter(Node mode) { //以给定模式构造结点。mode有两种：EXCLUSIVE（独占）和SHARED（共享） 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.next = node; return node; } } enq(node); return node; }
Node结点是对每一个访问同步代码的线程的封装，其包含了需要同步的线程本身以及线程的状态，如是否被阻塞，是否等待唤醒，是否已经被取消等。变量waitStatus则表示当前被封装成Node结点的等待状态，共有4种取值CANCELLED、SIGNAL、CONDITION、PROPAGATE。 `CANCELLED`：值为1，在同步队列中等待的线程等待超时或被中断，需要从同步队列中取消该Node的结点，其结点的waitStatus为CANCELLED，即结束状态，进入该状态后的结点将不会再变化。 `SIGNAL`：值为-1，被标识为该等待唤醒状态的后继结点，当其前继结点的线程释放了同步锁或被取消，将会通知该后继结点的线程执行。说白了，就是处于唤醒状态，只要前继结点释放锁，就会通知标识为SIGNAL状态的后继结点的线程执行。 `CONDITION`：值为-2，与Condition相关，该标识的结点处于等待队列中，结点的线程等待在Condition上，当其他线程调用了Condition的signal()方法后，CONDITION状态的结点将从等待队列转移到同步队列中，等待获取同步锁。 `PROPAGATE`：值为-3，与共享模式相关，在共享模式中，该状态标识结点的线程处于可运行状态。 `0`状态：值为0，代表初始化状态。 AQS在判断状态时，通过用waitStatus>0表示取消状态，而waitStatus<0表示有效状态。 == enq(Node)== 此方法用于将node加入队尾。
private Node enq(final Node node) { for (;;) {//自旋 Node t = tail; if (t == null) { // Must initialize // 队列为空，创建一个空的标志结点作为head结点，并将tail也指向它 if (compareAndSetHead(new Node())) tail = head; } else {//正常放入队列尾部 node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }
`cas自旋volatile变量` **acquireQueued(Node, int)** 通过tryAcquire()和addWaiter()，该线程获取资源失败，已经被放入等待队列尾部了。聪明的你立刻应该能想到该线程下一步该干什么了吧：进入等待状态休息，直到其他线程彻底释放资源后唤醒自己，自己再拿到资源，然后就可以去干自己想干的事了。没错，就是这样！是不是跟医院排队拿号有点相似~~acquireQueued()就是干这件事：在等待队列中排队拿号（中间没其它事干可以休息），直到拿到号后再返回。这个函数非常关键，还是上源码吧：
final boolean acquireQueued(final Node node, int arg) { boolean failed = true;//标记是否已经拿到锁 try { boolean interrupted = false;//标记等待过程中是否被中断过 for (;;) { final Node p = node.predecessor();//拿到前驱节点 //如果前驱是head，即该结点已成老二，那么便有资格去尝试获取资源（可能是老大释放完资源唤醒自己的，当然也可能被interrupt了）。 if (p == head && tryAcquire(arg)) { setHead(node); //拿到资源后，将head指向该结点。所以head所指的标杆结点，就是当前获取到资源的那个结点或null。 p.next = null; // help GC failed = false; return interrupted; } //如果自己可以休息了，就进入waiting状态，直到被unpark() if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true;//如果等待过程中被中断过，哪怕只有那么一次，就将interrupted标记为true } } finally { if (failed) cancelAcquire(node); } }
**shouldParkAfterFailedAcquire(Node, Node)** 此方法主要用于检查状态，看看自己是否真的可以去休息了。
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus;//拿到前驱的状态 if (ws == Node.SIGNAL) //如果已经告诉前驱拿完号后通知自己一下，那就可以安心休息了 return true; if (ws > 0) { /* * 如果前驱放弃了，那就一直往前找，直到找到最近一个正常等待的状态，并排在它的后边。 * 注意：那些放弃的结点，由于被自己“加塞”到它们前边，它们相当于形成一个无引用链，稍后就会被保安大叔赶走了(GC回收)！ */ do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { //如果前驱正常，那就把前驱的状态设置成SIGNAL，告诉它拿完号后通知自己一下。有可能失败，人家说不定刚刚释放完呢！ compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; }
整个流程中，如果前驱结点的状态不是SIGNAL，那么自己就不能安心去休息，需要去找个安心的休息点，同时可以再尝试下看有没有机会轮到自己拿号。 **parkAndCheckInterrupt()** 如果线程找好安全休息点后，那就可以安心去休息了。此方法就是让线程去休息，真正进入等待状态。
private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); }
``
