J.U.C(1)

JUC（一）

一：AQS

• aqs全称是AbstractQueuedSynchronizer，是一种阻塞式锁和同步式工具的框架；

state：

• state表示资源的状态（包含独占模式和共享模式）。子类需要定义如何维护这些状态，包含获取锁和释放锁；
• getstate是获取state；
• setstate是设置state；
• compareAndsetState是使用cas机制设置state；
• 独占模式是指只能有一个线程去访问资源，共享模式是可以同时有多个线程去访问资源；
• 提供了基于fifo，先进先出的等待队列，类似于monitor的entryList；
• 条件变量用来实现等待和唤醒机制，支持多个条件变量，类似于monitor的waitset

需要子类实现的方法：

子类主要实现这样一些方法（默认抛出 UnsupportedOperationException）

• tryAcquire
• tryRelease
• tryAcquireShared
• tryReleaseShared
• isHeldExclusively

获取锁：

// 如果获取锁失败
if (!tryAcquire(arg)) {// 入队, 可以选择阻塞当前线程 park unpark
}

释放锁：

// 如果释放锁成功
if (tryRelease(arg)) {// 让阻塞线程恢复运行
}

• 自己实现一个不可重入锁，通过同步器；

class MyLock implements Lock{class Mysych extends AbstractQueuedSynchronizer {@Overrideprotected boolean tryAcquire(int arg) {if (compareAndSetState(0,1)){setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}protected Condition newCondition() {return new ConditionObject();}@Overrideprotected boolean tryRelease(int arg) {setExclusiveOwnerThread(null);setState(0);return true;}@Overrideprotected boolean isHeldExclusively() {return getState()==1;}}Mysych mysych=new Mysych();@Override//加锁，但是加锁失败就将线程加入等待队列public void lock() {mysych.acquire(1);}@Override//可中断public void lockInterruptibly() throws InterruptedException {mysych.acquireInterruptibly(1);}@Override//尝试加锁，和lock不同，只会加一次锁，失败了就会返回falsepublic boolean tryLock() {return mysych.tryAcquire(1);}@Override//带超时时间加锁public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {return mysych.tryAcquireNanos(1,unit.toNanos(time));}@Override//释放锁public void unlock() {mysych.tryRelease(0);}@Override//条件变量public Condition newCondition() {return mysych.newCondition();}
}

自己实现的锁需要实现lock接口，然后内部的一个同步器需要继承aqs类，并实现其中的一些方法，然后使用同步器的方法去实现lock接口的方法；

二：reentrantlock原理

1：加锁：

• reentrantlock实现了lock接口，然后还实现了一个内部的同步器，同步器是继承了aqs的，同步器的分为公平锁的同步器和非公平锁的同步器；

然后一般创建reentrantlock默认是创建的非公平锁：

public ReentrantLock() {sync = new NonfairSync();
}

• 没有竞争时，直接将state设为1，然后将owner设为当前线程

• 当出现竞争时：

就会进行aquire方法：

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

• 首先还是会再次尝试获取锁，如果获取锁失败就会执行acquireQueued，这里就会将当前线程加入等待队列，等待队列是一个双向链表。
• Node是懒惰创建的；
• Node有waitstatus，默认是0表示正常的；
• 刚开始第一个Node被称为哨兵， 只是用来占位的，不予其他线程相关联；

final boolean acquireQueued(final Node node, int arg) {boolean interrupted = false;try {for (;;) {final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCreturn interrupted;}if (shouldParkAfterFailedAcquire(p, node))interrupted |= parkAndCheckInterrupt();}} catch (Throwable t) {cancelAcquire(node);if (interrupted)selfInterrupt();throw t;}}

• 进入到acquireQueued逻辑，多次尝试获取锁，如果获取锁失败就会进入park阻塞；
• 循环中，获取当前节点的前驱节点，如果前驱节点是头节点，然后再获取锁，如果还是失败就会进入shouldParkAfterFailedAcquire
• shouldParkAfterFailedAcquire逻辑中，会将前驱节点的waitstatus设置为-1，然后这次会返回false；

• 进入下一次循环，再次尝试获取锁，还是失败；
• 然后进入shouldParkAfterFailedAcquire，这次返回true，进入if里面的逻辑，将线程阻塞；

之后如果多次竞争失败就会变成这样：

2：解锁

public void unlock() {sync.release(1);
}

• 解锁时调用unlock方法；

public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h);return true;}return false;
}

• release将state状态设为0，设置成功之后，他就会判断头节点的waitstatus是不是为-1，如果为-1就去唤醒头节点之后的节点；

private void unparkSuccessor(Node node) {/** If status is negative (i.e., possibly needing signal) try* to clear in anticipation of signalling.  It is OK if this* fails or if status is changed by waiting thread.*/int ws = node.waitStatus;if (ws < 0)node.compareAndSetWaitStatus(ws, 0);/** Thread to unpark is held in successor, which is normally* just the next node.  But if cancelled or apparently null,* traverse backwards from tail to find the actual* non-cancelled successor.*/Node s = node.next;if (s == null || s.waitStatus > 0) {s = null;for (Node p = tail; p != node && p != null; p = p.prev)if (p.waitStatus <= 0)s = p;}if (s != null)LockSupport.unpark(s.thread);
}

• unpark，唤醒park的节点；

然后唤醒之后还要去获取锁：

final boolean acquireQueued(final Node node, int arg) {boolean interrupted = false;try {for (;;) {final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCreturn interrupted;}if (shouldParkAfterFailedAcquire(p, node))interrupted |= parkAndCheckInterrupt();}} catch (Throwable t) {cancelAcquire(node);if (interrupted)selfInterrupt();throw t;}
}

• 还是在acquireQueued方法中，因为park之后阻塞在parkAndCheckInterrupt方法，unpark之后重新进入循环，这个时候再去tryAcquire就会成功，然后就会if中的逻辑，将当前节点设为头节点，当前节点2关联的线程也设置为空；
• 原本的头节点会被垃圾回收；

还有一种竞争失败情况，就是在阻塞队列中unpark之后的线程要将非公平锁的state改成1时，外部来了线程将state改为1了，就出现了竞争锁失败的情况；

final boolean acquireQueued(final Node node, int arg) {boolean interrupted = false;try {for (;;) {final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCreturn interrupted;}if (shouldParkAfterFailedAcquire(p, node))interrupted |= parkAndCheckInterrupt();}} catch (Throwable t) {cancelAcquire(node);if (interrupted)selfInterrupt();throw t;}}

就是在这段代码中，parkAndCheckInterrupt结束阻塞，进入下一次循环时，tryAcquire失败，就是获取锁失败会继续等待；

3：可重入锁原理

@ReservedStackAccess
final boolean nonfairTryAcquire(int acquires) {//获取当前线程final Thread current = Thread.currentThread();//获取锁的状态int c = getState();//判断是否的第一次加锁if (c == 0) {if (compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}//如果不是第一次加锁，就可能是重入锁，那么就要判断拥有锁的线程是否是当前线程else if (current == getExclusiveOwnerThread()) {//累加int nextc = c + acquires;if (nextc < 0) // overflowthrow new Error("Maximum lock count exceeded");//设置锁的状态setState(nextc);return true;}return false;
}

释放锁：

@ReservedStackAccess
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;
}

一直减，直到减为0返回true；

4：可打断原理

• 不可打断模式（默认是不可打断的）
• 进入队列之后，被打断不会立即响应，只有获取锁之后才能反应，但也是继续运行，即打断标记为true；

（一）final boolean acquireQueued(final Node node, int arg) {boolean interrupted = false;try {for (;;) {final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCreturn interrupted;}if (shouldParkAfterFailedAcquire(p, node))interrupted |= parkAndCheckInterrupt();//阻塞进入的方法}} catch (Throwable t) {cancelAcquire(node);if (interrupted)selfInterrupt();throw t;}
}
（二）private final boolean parkAndCheckInterrupt() {LockSupport.park(this);//此时阻塞//被打断后，会返回true，但是会被清除为true；return Thread.interrupted();}
（三）public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//返回的打断标记是true，就会进入if语句块中selfInterrupt();//会打断一下然后重新设置为true（表示之前被打断过）}
（四）static void selfInterrupt() {Thread.currentThread().interrupt();}

然后interrupted会被设置为true，直到某次循环获取锁之后会返回打断标记是true；

• 可打断模式，直接抛出异常停止等待；

public final void acquireInterruptibly(int arg)throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();if (!tryAcquire(arg))doAcquireInterruptibly(arg);//进入方法（二）
}
（二）private void doAcquireInterruptibly(int arg)throws InterruptedException {final Node node = addWaiter(Node.EXCLUSIVE);try {for (;;) {final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCreturn;}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())//如果被打断返回true，进入ifthrow new InterruptedException();//直接抛出异常停止等待}} catch (Throwable t) {cancelAcquire(node);throw t;}}

5：公平锁原理

• 非公平锁（不管等待队列，如果没人占用锁直接占用）

  @ReservedStackAccess
  final boolean nonfairTryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();if (c == 0) {if (compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}else if (current == getExclusiveOwnerThread()) {int nextc = c + acquires;if (nextc < 0) // overflowthrow new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;
  }
  

• 公平锁（调用hasQueuedPredecessors方法）

 @ReservedStackAccessprotected final boolean tryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();if (c == 0) {if (!hasQueuedPredecessors() &&compareAndSetState(0, acquires)) {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;}
}

public final boolean hasQueuedPredecessors() {Node h, s;if ((h = head) != null) {if ((s = h.next) == null || s.waitStatus > 0) {s = null; // traverse in case of concurrent cancellationfor (Node p = tail; p != h && p != null; p = p.prev) {if (p.waitStatus <= 0)s = p;}}if (s != null && s.thread != Thread.currentThread())return true;}return false;

• 如果队列中有多个节点，并且当前节点不是第二个节点就会返回true，这样的化！hasQueuedPredecessors就是false，不会进入后面的获取锁的方法；

6：条件变量

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 cancelledunlinkCancelledWaiters();if (interruptMode != 0)reportInterruptAfterWait(interruptMode);
}

await方法，首先将当前线程加入到ConditionObject中，这里的等待队列也是一个双向链表，有头节点和尾节点

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

这里将线程与节点关联，然后如果没有节点的话这个节点就是头尾节点，如果有节点的话加入到链表的末尾；

node的waitstatus是-2

然后就要释放锁，因为可能是可重入锁，所以要使用方法fullyRelease，将所有锁释放，将state置为0；

final int fullyRelease(Node node) {try {int savedState = getState();if (release(savedState))return savedState;throw new IllegalMonitorStateException();} catch (Throwable t) {node.waitStatus = Node.CANCELLED;throw t;}
}

调用release方法，因为进入了condition中了所已需要唤醒后面的线程；

进入release方法，执行unparkSuccessor唤醒后面的线程；

public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h);return true;}return false;
}

最后咱们要park在condition的线程；

• signal唤醒流程

thread-1要去唤醒thread-0，也是从condition中的第一个元素开始唤醒；

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

唤醒的操作主要有两个，一个从将该线程节点从condition中断开，然后将NOde的状态变成0，加入到AQS的队列中，然后将队列尾前的节点的状态设为-1；

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

final boolean transferForSignal(Node node) {/** If cannot change waitStatus, the node has been cancelled.*/if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))return false;/** Splice onto queue and try to set waitStatus of predecessor to* indicate that thread is (probably) waiting. If cancelled or* attempt to set waitStatus fails, wake up to resync (in which* case the waitStatus can be transiently and harmlessly wrong).*/Node p = enq(node);int ws = p.waitStatus;if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))LockSupport.unpark(node.thread);return true;
}

三：读写锁（reentrantreadwriteLock）

• 当读操作远远大于写操作时，我们使用读写锁让读- 读可以并发执行，提高效率；

class DataContainer {private Object data;private ReentrantReadWriteLock rw = new ReentrantReadWriteLock();private ReentrantReadWriteLock.ReadLock r = rw.readLock();private ReentrantReadWriteLock.WriteLock w = rw.writeLock();public Object read() {log.debug("获取读锁...");r.lock();try {log.debug("读取");sleep(1);return data;} finally {log.debug("释放读锁...");r.unlock();}}public void write() {log.debug("获取写锁...");w.lock();try {log.debug("写入");sleep(1);} finally {log.debug("释放写锁...");w.unlock();}}}

当两个线程进行读操作时没有互斥，一个读一个写，或者两个都是写操作的时候产生互斥；

注意事项：

1：读锁不支持条件变量

2：不支持锁升级，即以及索取了读锁不能再获取写锁，必须要释放了读锁之后才能获取写锁，否则就会永久等待；

3：可以锁降级，在获取写锁的情况下可以获取读锁；

class CachedData {Object data;// 是否有效，如果失效，需要重新计算 datavolatile boolean cacheValid;final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();void processCachedData() {rwl.readLock().lock();if (!cacheValid) {// 获取写锁前必须释放读锁rwl.readLock().unlock();rwl.writeLock().lock();try {// 判断是否有其它线程已经获取了写锁、更新了缓存, 避免重复更新if (!cacheValid) {data = ...cacheValid = true;}// 降级为读锁, 释放写锁, 这样能够让其它线程读取缓存rwl.readLock().lock();} finally {rwl.writeLock().unlock();}}// 自己用完数据, 释放读锁 try {use(data);} finally {rwl.readLock().unlock();}}
}

应用之缓存

缓存更新策略：

先清空缓存再更新缓存：如果有两个线程，一个线程清空缓存之后，另一个线程来读取，发现缓存没有，于是去数据库查找，这个时候前面的线程还没完成数据的更新，于是查找的信息还是没有更新之前的，并且并入缓存，之后查询虽然数据库已经更改了，但是缓存中是旧数据；

如果是先更新数据库，一个线程去更新数据库，另外一个线程查询旧数据，然后更新完数据库之后就清除缓存，旧数据也被清除，就可以重新建立缓存

还有一个概率非常小的情况：就是第一次建立缓存或者是缓存更新之后，建立缓存，会查询数据库，但是查询时间比较长，这个时候另一个线程修改了数据，并且清空了缓存之后，前面的线程查询到了缓存建立了缓存，这是旧的数据，就会造成缓存的不一致；

class GenericCachedDao<T> {// HashMap 作为缓存非线程安全, 需要保护HashMap<SqlPair, T> map = new HashMap<>();ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); GenericDao genericDao = new GenericDao();public int update(String sql, Object... params) {SqlPair key = new SqlPair(sql, params);// 加写锁, 防止其它线程对缓存读取和更改lock.writeLock().lock();try {int rows = genericDao.update(sql, params);map.clear();return rows;} finally {lock.writeLock().unlock();}}public T queryOne(Class<T> beanClass, String sql, Object... params) {SqlPair key = new SqlPair(sql, params);// 加读锁, 防止其它线程对缓存更改lock.readLock().lock();try {T value = map.get(key);if (value != null) {return value;}} finally {lock.readLock().unlock();}// 加写锁, 防止其它线程对缓存读取和更改lock.writeLock().lock();try {// get 方法上面部分是可能多个线程进来的, 可能已经向缓存填充了数据// 为防止重复查询数据库, 再次验证T value = map.get(key);if (value == null) {// 如果没有, 查询数据库value = genericDao.queryOne(beanClass, sql, params);map.put(key, value);}return value;} finally {lock.writeLock().unlock();}}// 作为 key 保证其是不可变的class SqlPair {private String sql;private Object[] params;public SqlPair(String sql, Object[] params) {this.sql = sql;this.params = params;}@Overridepublic boolean equals(Object o) {if (this == o) {return true;}if (o == null || getClass() != o.getClass()) {return false;}SqlPair sqlPair = (SqlPair) o;return sql.equals(sqlPair.sql) &&Arrays.equals(params, sqlPair.params);}@Overridepublic int hashCode() {int result = Objects.hash(sql);result = 31 * result + Arrays.hashCode(params);return result;}}}

• 修改数据库，更新缓存的位置加上写锁，查询数据库的位置加上读锁；
• 然后在建立缓存时，是一个写锁，可能有多个线程进入但是只有一个线程进来，进来之后查询数据库建立缓存，释放锁，后续的线程进来还是会再次查询数据库，所以我们进行二次检测，判断缓存有没有前面的线程建立好，就不用再去查询数据库了；

补充

• 适合读多写少的情况，当有大量的写操作时，性能会有影响；
• 没有缓存容量
• 没有缓存过期
• 只适合单机
• 只有一把锁，并发性还是不高，如果是有多张表应该配有多把锁；
• 更新过于简单，直接把所以key删除，应该将key重新划分

原理-加写锁

• 读写锁使用的是同一个sych同步器，因此等待队列和state用的都是同一个

• 读写锁加写锁的过程和reentrantlock加锁的原理一致，唯一不同的是，state分为两个部分，高16位是读锁，低16位是写锁；

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

• 首先进入acquire，然后尝试加锁；

@ReservedStackAccess
protected final boolean tryAcquire(int acquires) {/** Walkthrough:* 1. If read count nonzero or write count nonzero*    and owner is a different thread, fail.* 2. If count would saturate, fail. (This can only*    happen if count is already nonzero.)* 3. Otherwise, this thread is eligible for lock if*    it is either a reentrant acquire or*    queue policy allows it. If so, update state*    and set owner.*/Thread current = Thread.currentThread();int c = getState();//获取当前的状态值int w = exclusiveCount(c);//获取低16位if (c != 0) {//c不等于0分为两种情况一个是加了读锁，一个是加了写锁// (Note: if c != 0 and w == 0 then shared count != 0)if (w == 0 || current != getExclusiveOwnerThread())//如果w==0说明加了读锁，读写互斥falsereturn false;//如果当前线程不等于owner线程说明不是重入就返回falseif (w + exclusiveCount(acquires) > MAX_COUNT)//加写锁，判断写锁不会超出16位的限制throw new Error("Maximum lock count exceeded");// Reentrant acquiresetState(c + acquires);//更新state的状态return true;}//如果c等于0if (writerShouldBlock() ||//这个方法非公平锁返回false，公平锁判断是否的等待队列的第二个节点!compareAndSetState(c, c + acquires))//尝试加锁，加锁失败进入ifreturn false;//返回falsesetExclusiveOwnerThread(current);//获取锁成功，将owner设为当前线程return true;
}

• 然后进入tryacquire尝试加锁

原理-加读锁

public final void acquireShared(int arg) {if (tryAcquireShared(arg) < 0)doAcquireShared(arg);
}

• tryAcquireShared方法{-1-获取锁失败 。 0-获取锁成功。 正数- 获取锁成功，并且表示还有几个节点需要唤醒}

• 加锁失败进入doAcquireShared方法，在这里就是加入等待队列的逻辑，首先是加入两个节点第一个是占位的，第二个就是读锁的节点，这里的不同是NODE是shared模式，不是ex模式，此时t2还处于活跃状态

• 然后如果节点是第二个节点还会再去尝试获取锁，获取锁失败进入下面的代码将前驱节点的waitstatus改为-1，然后再进入一次循环，再次尝试获取锁，获取锁失败就能park；

private void doAcquireShared(int arg) {final Node node = addWaiter(Node.SHARED);boolean interrupted = false;try {for (;;) {final Node p = node.predecessor();if (p == head) {int r = tryAcquireShared(arg);if (r >= 0) {setHeadAndPropagate(node, r);p.next = null; // help GCreturn;}}if (shouldParkAfterFailedAcquire(p, node))interrupted |= parkAndCheckInterrupt();}} catch (Throwable t) {cancelAcquire(node);throw t;} finally {if (interrupted)selfInterrupt();}
}

• 后续t3加读锁，t4加写锁

原理-释放写锁

• 释放锁同样也是调用tryRelease，将owner置为空，同时也会唤醒等待队列中的第二个线程节点；

唤醒之后，再来一次循环，尝试给高16位加锁，加锁成功state的高位就会加一；

• 然后进入setHeadAndPropagate方法，这个方法会将加锁成功的线程节点作为头节点，再setHeadAndPropagate方法内还会判断下一个节点是否也是读锁，如果也是的话就会唤醒；

• 最后变成这样

原理-释放读锁

• 使用unlock释放读锁，然后state会减一，最后返回是否==0，这里因为state=2，。state减一之后不等于0，返回flase；
• 再次unlock释放t3；
• 释放t3，这时state==0了，返回true，进入doReleaseShared方法，在这个方法中会将头节点的status设为0，并且将阻塞队列中的老二唤醒，这时没有其他线程竞争，t4获取到锁；