Lock接口概念
- 概念：
JDK1.5时，在Java.uitl.concurrent并发包中添加了Lock锁接口，其中定义了lock()、unLock()、tryLock()、lockInterruptibly()、newCondition()等一系列接口，它属于是JAVA语言层面的内置锁； - 特性：
- 显示锁：不同于
synchronized隐式锁，锁的获取和释放都是隐式的由JVM来管理，不需要开发人员手动实现；而显示锁的获取和释放都需要开发人员手动编码实现； - 非阻塞式获取锁：定义了
tryLock()方法，可以在获取锁时设置超时时长，达到超时时间还未获取到锁会返回一个false； - 支持中断获取锁：定义了
lockInterruptibly()方法，可以接受中断信号； - 多条件等待唤醒机制
Condition：Condition定义了一系列线程之间通信方法，能更细粒度的进行线程间的通信；
- 显示锁：不同于
Lock接口和synchronized的比较
-
锁的类型比较：
- 公平/非公平：
synchronized是非公平锁；ReentrantLock可以是公平锁也可以是非公平锁；具体可参考； - 都是可重入锁；
- 都为悲观锁；
- 公平/非公平：
-
锁的获取与释放比较：
Lock是显示锁必须手动获取，手动释放；而synchronized是隐式锁，锁的获取与释放由JVM来管理；Lock可以获取多个锁:ReentrantReadWriteLock中的读锁获取reentrantReadWriteLock.readLock().lock();Lock可以通过tryLock()方法非阻塞式获取锁；
-
异常情况：
synchronized在发生异常的时候，会自动释放线程占有的锁ReentrantLock在发生异常时，如果没有通过unlock去释放锁，很有可能造成死锁，因此需要在finally块中释放锁；
-
线程调度比较：
synchronized: 使用Object对象本身的wait 、notify、notifyAll调度机制；Lock: 使用Condition中的signal(); await();等， 进行线程之间的调度；
Lock接口的使用
Lock接口中的核心方法
public interface Lock {/*** 尝试获取锁，若资源空闲则直接获取锁，否则阻塞等待*/void lock();/*** 与lock()方法作用类似，只是若没有获取到锁阻塞时，可以接受中断命令，中断获取锁*/void lockInterruptibly() throws InterruptedException;/*** 尝试非阻塞获取锁并立刻返回，若资源空闲则直接获取锁并返回true，否则直接返回false不会阻塞等待；*/boolean tryLock();/*** tryLock()方法作用类似，只是设置了获取锁的等待时间，在等待时间内阻塞获取锁，获取到锁返回true，否则返回false;*/boolean tryLock(long time, TimeUnit unit) throws InterruptedException;/*** 释放锁*/void unlock();/*** 可以将一个condition实例绑定到lock实例上，然后可以通过signal(); await();等方法，对锁线程进行调度*/Condition newCondition();
}Lock接口的实现类：
ReentrantLock：
独占锁，可重入锁，公平/非公平锁，可中断锁；
简单使用案例：
package com.xrl.juc.lock_demo;import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;/*** @author xrl* @date 2024/7/16 22:47*/
public class ReentrantLockTest implements Runnable{private Lock lock = new ReentrantLock();@Overridepublic void run() {get();}private void get() {lock.lock();try {System.out.println(Thread.currentThread() + "---------get()方法");Thread.sleep(1000);} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();}}public static void main(String[] args) {ReentrantLockTest lockTest = new ReentrantLockTest();new Thread(lockTest).start();new Thread(lockTest).start();new Thread(lockTest).start();new Thread(lockTest).start();new Thread(lockTest).start();}
}
ReentrantReadWriteLock：
- 读写锁：一般在读多写少情况下使用。有两把锁一把读锁，一把写锁写锁是独占式的，读锁是共享式的
- 实现原理：底层使用
AQS实现，同步状态的实现是在一个整形变量上通过“按位切割使用”；将变量切割成两部分，高16位表示读状态，也就是获取到读锁的次数，低16位表示获取到写线程的可重入次数，并通过CAS对其进行操作实现读写分离； - 三个特性：
- 公平性：支持公平和非公平两种模式。
- 重入性：支持重入，读写锁都支持最多65535个。
- 锁降级：先获取写锁，再获取读锁，再释放写锁，写锁就能降级为读锁。
简单案例
package com.xrl.juc.lock_demo;import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantReadWriteLock;/*** 有两把锁,一把读锁一把写锁,同一个线程中可以同时拥有两把锁但必须先获取 writeLock 再获取 readLock* 读锁可以有多个线程共有,写锁只能一个线程* 在有读锁的情况下没有写锁,有写锁的小情况下只有拥有写锁的那个线程可以有读锁* 读多写少情况下使用** @author xrl* @date 2024/7/16 23:27*/
public class ReentrantReadWriteLockeTest {/*** 可重入的读写锁* ReadWriteLock接口需要实现两把锁**/private static ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();private static ExecutorService threadPool= Executors.newFixedThreadPool(5);/*** 让线程同步争抢锁*/private static CyclicBarrier cyclicBarrier = new CyclicBarrier(3);private static int i = 100;public static void main(String[] args) {//通过 cyclicBarrier 使三个线程同步运行threadPool.execute(() -> {write(Thread.currentThread());});threadPool.execute(() -> {read(Thread.currentThread());});threadPool.execute(() -> {read(Thread.currentThread());});threadPool.shutdown();}private static void write(Thread thread) {try {//线程阻塞, 让线程同步争抢锁cyclicBarrier.await();} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}//获取写锁reentrantReadWriteLock.writeLock().lock();//获取读锁
// reentrantReadWriteLock.readLock().lock();try {System.out.println("写线程开始运行" + thread.getName() + " : i=" + i);Thread.sleep(1000);i = 1;System.out.println(thread.getName() + "运行完" + " : i=" + i );} catch (InterruptedException e) {e.printStackTrace();} finally {//释放锁System.out.println("释放锁");reentrantReadWriteLock.writeLock().unlock();}}private static void read(Thread thread) {try {//线程阻塞, 让线程同步争抢锁cyclicBarrier.await();} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}// 线程请求后阻塞至写完成System.out.println("线程请求读锁");reentrantReadWriteLock.readLock().lock();System.out.println("得到读锁");//获取写锁
// reentrantReadWriteLock.writeLock().lock();try {System.out.println("读线程开始运行" + thread.getName() + " : i=" + i);Thread.sleep(1000);System.out.println(thread.getName() + "运行完");} catch (InterruptedException e) {e.printStackTrace();} finally {//释放锁reentrantReadWriteLock.readLock().unlock();}}
}
结果，说明获得写锁和获得读锁的线程不能同时运行，获得读锁的线程可以同时运行
写线程开始运行pool-1-thread-1 : i=100
pool-1-thread-1运行完 : i=1
读线程开始运行pool-1-thread-2 : i=1
读线程开始运行pool-1-thread-3 : i=1
pool-1-thread-2运行完
pool-1-thread-3运行完
AQS
-
概念：
AQS是AbstractQueuedSynchronizer抽象类的简称，它通过模板模式的设计模式构建，其内部除了提供并发操作核心方法以及同步队列操作外，还提供了一些模板方法让子类自己实现，如加锁操作及解锁操作，从而使AQS只关注内部公共方法实现并不关心外部不同模式的具体逻辑实现；模板方法如下；
ReentrantLock实现的是tryAcquire(int arg)、tryRelease(int arg)、isHeldExclusively()//独占模式下获取锁的方法 protected boolean tryAcquire(int arg) {throw new UnsupportedOperationException(); } //独占模式下释放锁的方法 protected boolean tryRelease(int arg) {throw new UnsupportedOperationException(); } //共享模式下获取锁的方法 protected int tryAcquireShared(int arg) {throw new UnsupportedOperationException(); } //共享模式下释放锁的方法 protected boolean tryReleaseShared(int arg) {throw new UnsupportedOperationException(); } //判断是否持有独占锁的方法 protected boolean isHeldExclusively() {throw new UnsupportedOperationException(); }
核心属性：
- 同步状态：
AQS内部通过一个用volatile修饰的int类型变量state作为同步状态private volatile int state;；当state为0时，代表着当前没有线程占用锁资源，反之不为0时，代表着锁资源已经被线程持有，通过CAS设置值； Node内部类（同步队列的节点）：当锁资源已经被占用时（state != 0），则当前线程则会被封装成一个Node节点加入同步队列进行等待，Node节点主要包括了当前执行的线程本身、线程的状态，是否被阻塞、是否处于等待唤醒、是否中断等；每个Node节点中都关联着前驱节点prev以及后继节点next，从而构成一个双端队列；- 同步队列：由上述未获取到锁资源的节点
Node构成的FIFO双端队列，被称之为CLH队列，并在AQS中维护了两个变量private transient volatile Node head;和private transient volatile Node tail;分别指向CLH队列的头和尾，方便数据新增和添加； - 阻塞队列：
AQS使用内部类ConditionObject用来构建等待队列，当Condition调用await()方法后，等待获取锁资源的线程将会加入等待队列中，而当Condition调用signal()方法后，线程将会从等待队列转移到同步队列；
原理分析
AQS主要内部维护着一个由Node节点构成的FIFO的同步队列；- 当一个线程执行
ReetrantLock.lock()加锁方法获取锁失败时，该线程会被封装成Node节点加入同步队列等待锁资源的释放，期间不断执行自旋等待； - 当该线程所在节点的前驱节点为队列头结点时，当前线程就会开始尝试对同步状态标识
state进行修改(+1)，如果可以修改成功则代表获取锁资源成功，然后将自己所在的节点设置为队头head节点，表示自己已经持有锁资源; - 当一个线程调用
ReetrantLock.unlock()释放锁时，最终会调用Sync内部类中的tryRelease(int releases)方法再次对同步状态标识state进行修改(-1)，成功之后唤醒当前线程所在节点的后继节点中的线程。
ReentrantLock源码分析
-
ReentrantLock是Lock接口的实现类，当我们在new ReentrantLock();时，实际是创建了一个内部类Sync（同步器）的子类sync = new NonfairSync();，它拥有两个子类一个为公平锁一个为非公平锁sync = fair ? new FairSync() : new NonfairSync();；当
ReentrantLock调用方法时，大部分方法实际都是调用sync类中的方法，如：public void lock() { sync.lock(); }由下面代码可知
sync继承AbstractQueuedSynchronizer抽象类简称AQS；public class ReentrantLock implements Lock, java.io.Serializable {private static final long serialVersionUID = 7373984872572414699L;private final Sync sync;abstract static class Sync extends AbstractQueuedSynchronizer {
ReentrantLock方法说明：
//构造器//默认构造器创建非公平锁public ReentrantLock();//传入一个Boolean类型的参数，true创建一个公平锁，false非公平public ReentrantLock(boolean fair);// 查询当前线程调用lock()的次数int getHoldCount() // 返回目前持有此锁的线程，如果此锁不被任何线程持有,返回null protected Thread getOwner(); // 返回一个集合，它包含可能正等待获取此锁的线程，其内部维持一个队列(后续分析)protected Collection<Thread> getQueuedThreads(); // 返回正等待获取此锁资源的线程估计数int getQueueLength();// 返回一个集合，它包含可能正在等待与此锁相关的Condition条件的线程(估计值)protected Collection<Thread> getWaitingThreads(Condition condition); // 返回调用当前锁资源Condition对象await方法后未执行signal()方法的线程估计数int getWaitQueueLength(Condition condition);// 查询指定的线程是否正在等待获取当前锁资源boolean hasQueuedThread(Thread thread); // 查询是否有线程正在等待获取当前锁资源boolean hasQueuedThreads();// 查询是否有线程正在等待与此锁相关的Condition条件boolean hasWaiters(Condition condition); // 返回当前锁类型，如果是公平锁返回true，反之则返回flaseboolean isFair() // 查询当前线程是持有当前锁资源boolean isHeldByCurrentThread() // 查询当前锁资源是否被线程持有boolean isLocked()
核心方法解析
构造方法：
-
public ReentrantLock(boolean fair);构造方法返回的是，实现了AQS抽象类中部分方法的抽象类Sync的子类。sync = fair ? new FairSync() : new NonfairSync();FairSync和NonfairSync两个对象的实现大致相同 ，只是一个是公平性的，一个是非公平性的；两者之前的区别在于
FairSync使用lock()方法获取锁之前，会先去判断同步队列中是否存在元素（hasQueuedPredecessors()方法），如果存在则将当前任务放入同步队列的队尾；NonfairSync只判断当前锁是否有人占用，如果没有就直接获取；
lock()方法
-
lock()方法，底层是调用的是AQS中的模板方法acquire(int arg)；public final void acquire(int arg) {// 调用子类中重写的`tryAcquire`判断能否获得资源；if (!tryAcquire(arg) &&// 将当前线程封装为一个`Node`节点，加入同步队列中；// 在同步队列中等待同步资源的释放，当线程争取到同步资源 的使用权后，线程中同步队列中出队；acquireQueued(addWaiter(Node.EXCLUSIVE), arg))// selfInterrupt();} -
tryAcquire(arg)：调用子类中重写的tryAcquire判断能否获得资源；protected final boolean tryAcquire(int acquires) {final Thread current = Thread.currentThread();// 得到当前锁的状态int c = getState();// 锁的状态为0时，表示现在锁没有被占用if (c == 0) {//hasQueuedPredecessors()判断当前的同步队列中是否存在元素，当前元素是不是队头元素，是公平锁和非公平锁的区别所在if (!hasQueuedPredecessors() &&// CAS修改state同步状态的值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;}// 到这里说明没有拿到锁，返回falsereturn false; } -
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)：在同步队列中等待同步资源的释放，当线程争取到同步资源 的使用权后，线程中同步队列中出队；final boolean acquireQueued(final Node node, int arg) {// 是否获取到资源标记boolean failed = true;try {// 中断标识boolean interrupted = false;// 自旋 for (;;) {// 获得传入节点的前驱节点final Node p = node.predecessor();// 如果当前节点的前驱节点节点等于头节点，且可以获得资源if (p == head && tryAcquire(arg)) {// 将当前结点设置为头节点setHead(node);// 设置原头结点的后继节点为null,帮助jvm回收垃圾；（能静方法说明头节点线程加锁操作已运行完成）p.next = null; // help GC// 标记已经成功获取同步资源failed = false;return interrupted;}// 当前节点不是头节或者获取资源失败// shouldParkAfterFailedAcquire(p, node)线程获取资源失败后，判断是否阻塞线程if (shouldParkAfterFailedAcquire(p, node) &&// 阻塞线程等待其他线程唤醒，并判断中断状态parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)// 如果最终都没能成功获取同步状态，结束该线程的请求cancelAcquire(node);} }-
shouldParkAfterFailedAcquire(p, node)：线程获取资源失败后，判断是否阻塞线程private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {// 得到前驱节点的同步状态int ws = pred.waitStatus;// 等于阻塞状态if (ws == Node.SIGNAL)// 返回ture说明要阻塞当前线程，等待前驱节点释放资源，唤醒后继节点 执行parkAndCheckInterrupt()return true;// 前驱状态大于0，说明state是CANCELLED，线程是已取消的if (ws > 0) {// 将同步队列中已经标记已取消的节点全部剔除do {// 从同步队列中去掉pred节点，（将后继节点的前驱节点改为这个节点的前驱节点node.prev = pred = pred.prev;} while (pred.waitStatus > 0);pred.next = node;} else {// 同步状态不大于0，状态是初始态或propagate态（在共享模式同步状态可以进行传播），则通过CAS将pred的状态改为SIGNAL阻塞状态compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false; } -
parkAndCheckInterrupt()：阻塞线程等待其他线程唤醒，并判断中断状态private final boolean parkAndCheckInterrupt() {// 阻塞当前线程，底层是使用了Unsafe类提供的方法来实现线程的park（阻塞）和unpark（唤醒）操作// 方法内部，调用 setBlocker(t, blocker);方法，将当前线程关联到指定的同步对象blocker，标记当前线程由该对象导致阻塞// 之后调用UNSAFE.park(false, 0L)方法阻塞线程，第一个参数false表示不相对时间进行阻塞，第二个参数0L表示无期限地阻塞当前线程，直到有其他线程唤醒// 当有线程唤醒当前线程时，再次调用setBlocker(t, null);解除线程和对象直接的关联LockSupport.park(this);// 判断当前线程是否已被中断，返回中断标识，并清除中断标识return Thread.interrupted(); } -
addWaiter(Node.EXCLUSIVE)：将当前线程封装为一个Node节点，加入同步队列中；private Node addWaiter(Node mode) {// 将当前线程封装为一个节点Node node = new Node(Thread.currentThread(), mode);Node pred = tail;// 尾结点不为空if (pred != null) {// 将当前接节点插入队尾node.prev = pred;// CAS设置尾结点if (compareAndSetTail(pred, node)) {// 构成双端队列，将前一个元素指向下一个元素pred.next = node;// 设置成功，返回当前接节点return node;}}// 尾节点为空也就是同步队列为空 或 CAS设置尾结点失败enq(node);return node; } -
enq(node)：将节点插入队列，队列为空时初始化private Node enq(final Node node) {// 自旋for (;;) {Node t = tail;// 尾节点为空if (t == null) { // cas 初始化头节点if (compareAndSetHead(new Node()))// 没有节点时，头尾相等tail = head;} else {// 将当前元素插入同步队列中，同addWaiter中的逻辑node.prev = t;if (compareAndSetTail(t, node)) {t.next = node;return t;}}} }
-
unlock()方法
-
底层调用的是
AQS的release(1);方法；public final boolean release(int arg) {// 尝试释放锁 ReentrantLock实现的模板方法if (tryRelease(arg)) {Node h = head;// 头节点不为空 且 节点状态不为初始状态if (h != null && h.waitStatus != 0)// 唤醒同步队列的后继节点unparkSuccessor(h);return true;}return false; }-
tryRelease(arg)：尝试释放锁 ReentrantLock实现的模板方法protected final boolean tryRelease(int releases) {// 计算同步状态int c = getState() - releases;// 如果当前释放锁的线程不为持有锁的线程则抛出异常if (Thread.currentThread() != getExclusiveOwnerThread())throw new IllegalMonitorStateException();boolean free = false;// 判断状态是否为0，如果是则说明已释放同步状态if (c == 0) {free = true;// 设置Owner为nullsetExclusiveOwnerThread(null);}// 修改同步状态setState(c);return free; } -
unparkSuccessor(h);： 唤醒同步队列的后继节点private void unparkSuccessor(Node node) {int ws = node.waitStatus;// 节点阻塞状态小于0，说明节点没有被取消if (ws < 0)// 将节点阻塞状态设置为0compareAndSetWaitStatus(node, ws, 0);Node s = node.next;// 后继节点为null或者为取消状态if (s == null || s.waitStatus > 0) {s = null;// 寻找到下一个非取消态的结点 （从同步队列队尾结点开始向前遍历，非空且不等于当前node ,则校验此结点t的状态）for (Node t = tail; t != null && t != node; t = t.prev)if (t.waitStatus <= 0)s = t;}// 唤醒结点t对应的线程if (s != null)LockSupport.unpark(s.thread); }
-
实现总结
-
基础属性：
- 同步状态标识：标识锁资源的占有状态；
- 同步队列：存放获取锁失败的线程；
- 等待队列：用于实现多条件唤醒；
- Node节点：队列的每个节点，线程封装体；
-
基础操作：
cas修改同步状态标识；- 获取锁失败加入同步队列阻塞；
- 释放锁时唤醒同步队列第一个节点线程；
-
加锁操作：
- 调用
tryAcquire()修改标识state，失败加入队列等待； - 加入队列后判断节点是否为
signal状态，是则调用LockSupport.park(this);阻塞挂起当前线程； - 判断是否为
cancel状态，是则往前遍历删除队列中所有cancel状态节点； - 如果节点为
0或者propagate状态则将其修改为signal状态； - 阻塞被唤醒后如果为
head则获取锁，成功返回true，失败则继续阻塞；
- 调用
-
解锁操作：
- 调用
tryRelease()释放锁修改标识state； - 释放锁成功后唤醒同步队列后继阻塞的线程节点；
- 被唤醒的节点会自动替换当前节点成为
head节点；
- 调用
Condition接口
使用Condition可以实现类似于Object和wait()、notify()等线程间通信操作；具体实现可参考；原理参考；
简单案例
package com.xrl.juc.conditiontest;import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;/*** 利用lock来替换synchronized实现 生产者,消费者模式** @author shkstart* @create 2024-07-24 17:56*/
public class producerConsumer {public static void main(String[] args) {AppleBox appleBox = new AppleBox();Producer producer = new Producer(appleBox);Consumer consumer = new Consumer(appleBox);Consumer consumer2 = new Consumer(appleBox);Thread p = new Thread(producer, "生产者");Thread c1 = new Thread(consumer, "消费者1");Thread c2 = new Thread(consumer2, "消费者2");p.start();c1.start();c2.start();}
}class Apple {int id;public Apple(int id) {this.id = id;}
}class AppleBox {AtomicInteger index = new AtomicInteger(0);AtomicReferenceArray<Apple> apples = new AtomicReferenceArray<>(new Apple[5]);
// AtomicReference<Apple[]> apples = new AtomicReference<>();//创建锁对象 ,替换synchronizedprivate final ReentrantLock lock = new ReentrantLock();//生产条件private final Condition produceCondition = lock.newCondition();//消费条件private final Condition consumeCondition = lock.newCondition();// public synchronized boolean deposite(Apple apple) {public boolean deposite(Apple apple) {lock.lock();try {while (index.get() == apples.length()) {//生产线程进入等待produceCondition.await();return false;}apples.set(index.getAndIncrement(), apple);//this.apples[index.incrementAndGet()] = apple;System.out.println(Thread.currentThread().getName() + "生产了: " + index.get());//唤醒消费线程consumeCondition.signal();return true;} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();}return false;}public Apple withdraw() throws Exception {lock.lock();try {while (index.get() == 0) {System.out.println(Thread.currentThread().getName() + "消费缺货" );consumeCondition.await();}Apple a = apples.get(index.decrementAndGet());System.out.println(Thread.currentThread().getName() + "消费了: " + a.id);return a;} catch (InterruptedException e) {e.printStackTrace();} finally {lock.unlock();}return null;}
}class Producer implements Runnable {AppleBox appleBox;public Producer(AppleBox appleBox) {this.appleBox = appleBox;}@Overridepublic void run() {while (true) {appleBox.deposite(new Apple(appleBox.index.get() + 1));try {Thread.sleep((int) (Math.random() * 1000));} catch (InterruptedException e) {e.printStackTrace();}}}
}class Consumer implements Runnable {AppleBox appleBox;public Consumer(AppleBox appleBox) {this.appleBox = appleBox;}@Overridepublic void run() {while (true) {try {appleBox.withdraw();} catch (Exception e) {e.printStackTrace();}try {Thread.sleep((int) (Math.random() * 3000));} catch (InterruptedException e) {e.printStackTrace();}}}
}
 算法】)
