Java - "JUC" ReentrantLock获取锁

时间:2022-05-14 20:51:24

【Java并发编程实战】-----“J.U.C”:ReentrantLock之一简介

ReentrantLock介绍

ReentrantLock是一个可重入的互斥锁,又被称为“独占锁”。

顾名思义,ReentrantLock锁在同一个时间点只能被一个线程锁持有;而可重入的意思是,ReentrantLock锁,可以被单个线程多次获取。
ReentrantLock分为“公平锁”和“非公平锁”。它们的区别体现在获取锁的机制上是否公平。“锁”是为了保护竞争资源,防止多个线程同时操作线程而出错,ReentrantLock在同一个时间点只能被一个线程获取(当某线程获取到“锁”时,其它线程就必须等待);ReentraantLock是通过一个FIFO的等待队列来管理获取该锁所有线程的。在“公平锁”的机制下,线程依次排队获取锁;而“非公平锁”在锁是可获取状态时,不管自己是不是在队列的开头都会获取锁。

 

注:由于要介绍ReentrantLock的东西太多了,免得各位客官看累,所以分三篇博客来阐述。本篇博客介绍ReentrantLock基本内容,后两篇博客从源码级别分别阐述ReentrantLock的lock、unlock实现机制。

ReentrantLock,可重入的互斥锁,是一种递归无阻塞的同步机制。它可以等同于synchronized的使用,但是ReentrantLock提供了比synchronized更强大、灵活的锁机制,可以减少死锁发生的概率。

对于ReentrantLock,官方有详细的说明:一个可重入的互斥锁定 Lock,它具有与使用 synchronized 方法和语句所访问的隐式监视器锁定相同的一些基本行为和语义,但功能更强大。ReentrantLock 将由最近成功获得锁定,并且还没有释放该锁定的线程所拥有。当锁定没有被另一个线程所拥有时,调用 lock 的线程将成功获取该锁定并返回。如果当前线程已经拥有该锁定,此方法将立即返回。可以使用 isHeldByCurrentThread() 和 getHoldCount() 方法来检查此情况是否发生。

ReentrantLock提供公平锁机制,构造方法接收一个可选的公平参数。当设置为true时,它是公平锁,这些所将访问权授予等待时间最长的线程。否则该锁将无法保证线程获取锁的访问顺序。但是公平锁与非公平锁相比,公平锁的程序在许多线程访问时表现为很低的总体吞吐量。

 

ReentrantLock示例

通过对比“示例1”和“示例2”,我们能够清晰的认识lock和unlock的作用

示例1

Java - "JUC" ReentrantLock获取锁
 1 import java.util.concurrent.locks.Lock;
 2 import java.util.concurrent.locks.ReentrantLock;
 3 
 4 // LockTest1.java
 5 // 仓库
 6 class Depot { 
 7     private int size;        // 仓库的实际数量
 8     private Lock lock;        // 独占锁
 9 
10     public Depot() {
11         this.size = 0;
12         this.lock = new ReentrantLock();
13     }
14 
15     public void produce(int val) {
16         lock.lock();
17         try {
18             size += val;
19             System.out.printf("%s produce(%d) --> size=%d\n", 
20                     Thread.currentThread().getName(), val, size);
21         } finally {
22             lock.unlock();
23         }
24     }
25 
26     public void consume(int val) {
27         lock.lock();
28         try {
29             size -= val;
30             System.out.printf("%s consume(%d) <-- size=%d\n", 
31                     Thread.currentThread().getName(), val, size);
32         } finally {
33             lock.unlock();
34         }
35     }
36 }; 
37 
38 // 生产者
39 class Producer {
40     private Depot depot;
41     
42     public Producer(Depot depot) {
43         this.depot = depot;
44     }
45 
46     // 消费产品:新建一个线程向仓库中生产产品。
47     public void produce(final int val) {
48         new Thread() {
49             public void run() {
50                 depot.produce(val);
51             }
52         }.start();
53     }
54 }
55 
56 // 消费者
57 class Customer {
58     private Depot depot;
59     
60     public Customer(Depot depot) {
61         this.depot = depot;
62     }
63 
64     // 消费产品:新建一个线程从仓库中消费产品。
65     public void consume(final int val) {
66         new Thread() {
67             public void run() {
68                 depot.consume(val);
69             }
70         }.start();
71     }
72 }
73 
74 public class LockTest1 {  
75     public static void main(String[] args) {  
76         Depot mDepot = new Depot();
77         Producer mPro = new Producer(mDepot);
78         Customer mCus = new Customer(mDepot);
79 
80         mPro.produce(60);
81         mPro.produce(120);
82         mCus.consume(90);
83         mCus.consume(150);
84         mPro.produce(110);
85     }
86 }
Java - "JUC" ReentrantLock获取锁

运行结果

Thread-0 produce(60) --> size=60
Thread-1 produce(120) --> size=180
Thread-3 consume(150) <-- size=30
Thread-2 consume(90) <-- size=-60
Thread-4 produce(110) --> size=50

结果分析
(01) Depot 是个仓库。通过produce()能往仓库中生产货物,通过consume()能消费仓库中的货物。通过独占锁lock实现对仓库的互斥访问:在操作(生产/消费)仓库中货品前,会先通过lock()锁住仓库,操作完之后再通过unlock()解锁。
(02) Producer是生产者类。调用Producer中的produce()函数可以新建一个线程往仓库中生产产品。
(03) Customer是消费者类。调用Customer中的consume()函数可以新建一个线程消费仓库中的产品。
(04) 在主线程main中,我们会新建1个生产者mPro,同时新建1个消费者mCus。它们分别向仓库中生产/消费产品。
根据main中的生产/消费数量,仓库最终剩余的产品应该是50。运行结果是符合我们预期的!

这个模型存在两个问题:
(01) 现实中,仓库的容量不可能为负数。但是,此模型中的仓库容量可以为负数,这与现实相矛盾!
(02) 现实中,仓库的容量是有限制的。但是,此模型中的容量确实没有限制的!
这两个问题,我们稍微会讲到如何解决。现在,先看个简单的示例2;通过对比“示例1”和“示例2”,我们能更清晰的认识lock(),unlock()的用途。

 

示例2

Java - "JUC" ReentrantLock获取锁
 1 import java.util.concurrent.locks.Lock;
 2 import java.util.concurrent.locks.ReentrantLock;
 3 
 4 // LockTest2.java
 5 // 仓库
 6 class Depot { 
 7     private int size;        // 仓库的实际数量
 8     private Lock lock;        // 独占锁
 9 
10     public Depot() {
11         this.size = 0;
12         this.lock = new ReentrantLock();
13     }
14 
15     public void produce(int val) {
16 //        lock.lock();
17 //        try {
18             size += val;
19             System.out.printf("%s produce(%d) --> size=%d\n", 
20                     Thread.currentThread().getName(), val, size);
21 //        } catch (InterruptedException e) {
22 //        } finally {
23 //            lock.unlock();
24 //        }
25     }
26 
27     public void consume(int val) {
28 //        lock.lock();
29 //        try {
30             size -= val;
31             System.out.printf("%s consume(%d) <-- size=%d\n", 
32                     Thread.currentThread().getName(), val, size);
33 //        } finally {
34 //            lock.unlock();
35 //        }
36     }
37 }; 
38 
39 // 生产者
40 class Producer {
41     private Depot depot;
42     
43     public Producer(Depot depot) {
44         this.depot = depot;
45     }
46 
47     // 消费产品:新建一个线程向仓库中生产产品。
48     public void produce(final int val) {
49         new Thread() {
50             public void run() {
51                 depot.produce(val);
52             }
53         }.start();
54     }
55 }
56 
57 // 消费者
58 class Customer {
59     private Depot depot;
60     
61     public Customer(Depot depot) {
62         this.depot = depot;
63     }
64 
65     // 消费产品:新建一个线程从仓库中消费产品。
66     public void consume(final int val) {
67         new Thread() {
68             public void run() {
69                 depot.consume(val);
70             }
71         }.start();
72     }
73 }
74 
75 public class LockTest2 {  
76     public static void main(String[] args) {  
77         Depot mDepot = new Depot();
78         Producer mPro = new Producer(mDepot);
79         Customer mCus = new Customer(mDepot);
80 
81         mPro.produce(60);
82         mPro.produce(120);
83         mCus.consume(90);
84         mCus.consume(150);
85         mPro.produce(110);
86     }
87 }
Java - "JUC" ReentrantLock获取锁

(某一次)运行结果

Thread-0 produce(60) --> size=-60
Thread-4 produce(110) --> size=50
Thread-2 consume(90) <-- size=-60
Thread-1 produce(120) --> size=-60
Thread-3 consume(150) <-- size=-60

结果说明
“示例2”在“示例1”的基础上去掉了lock锁。在“示例2”中,仓库中最终剩余的产品是-60,而不是我们期望的50。原因是我们没有实现对仓库的互斥访问。

 

示例3

在“示例3”中,我们通过Condition去解决“示例1”中的两个问题:“仓库的容量不可能为负数”以及“仓库的容量是有限制的”。
解决该问题是通过Condition。Condition是需要和Lock联合使用的:通过Condition中的await()方法,能让线程阻塞[类似于wait()];通过Condition的signal()方法,能让唤醒线程[类似于notify()]。

Java - "JUC" ReentrantLock获取锁
  1 import java.util.concurrent.locks.Lock;
  2 import java.util.concurrent.locks.ReentrantLock;
  3 import java.util.concurrent.locks.Condition;
  4 
  5 // LockTest3.java
  6 // 仓库
  7 class Depot {
  8     private int capacity;    // 仓库的容量
  9     private int size;        // 仓库的实际数量
 10     private Lock lock;        // 独占锁
 11     private Condition fullCondtion;            // 生产条件
 12     private Condition emptyCondtion;        // 消费条件
 13 
 14     public Depot(int capacity) {
 15         this.capacity = capacity;
 16         this.size = 0;
 17         this.lock = new ReentrantLock();
 18         this.fullCondtion = lock.newCondition();
 19         this.emptyCondtion = lock.newCondition();
 20     }
 21 
 22     public void produce(int val) {
 23         lock.lock();
 24         try {
 25              // left 表示“想要生产的数量”(有可能生产量太多,需多此生产)
 26             int left = val;
 27             while (left > 0) {
 28                 // 库存已满时,等待“消费者”消费产品。
 29                 while (size >= capacity)
 30                     fullCondtion.await();
 31                 // 获取“实际生产的数量”(即库存中新增的数量)
 32                 // 如果“库存”+“想要生产的数量”>“总的容量”,则“实际增量”=“总的容量”-“当前容量”。(此时填满仓库)
 33                 // 否则“实际增量”=“想要生产的数量”
 34                 int inc = (size+left)>capacity ? (capacity-size) : left;
 35                 size += inc;
 36                 left -= inc;
 37                 System.out.printf("%s produce(%3d) --> left=%3d, inc=%3d, size=%3d\n", 
 38                         Thread.currentThread().getName(), val, left, inc, size);
 39                 // 通知“消费者”可以消费了。
 40                 emptyCondtion.signal();
 41             }
 42         } catch (InterruptedException e) {
 43         } finally {
 44             lock.unlock();
 45         }
 46     }
 47 
 48     public void consume(int val) {
 49         lock.lock();
 50         try {
 51             // left 表示“客户要消费数量”(有可能消费量太大,库存不够,需多此消费)
 52             int left = val;
 53             while (left > 0) {
 54                 // 库存为0时,等待“生产者”生产产品。
 55                 while (size <= 0)
 56                     emptyCondtion.await();
 57                 // 获取“实际消费的数量”(即库存中实际减少的数量)
 58                 // 如果“库存”<“客户要消费的数量”,则“实际消费量”=“库存”;
 59                 // 否则,“实际消费量”=“客户要消费的数量”。
 60                 int dec = (size<left) ? size : left;
 61                 size -= dec;
 62                 left -= dec;
 63                 System.out.printf("%s consume(%3d) <-- left=%3d, dec=%3d, size=%3d\n", 
 64                         Thread.currentThread().getName(), val, left, dec, size);
 65                 fullCondtion.signal();
 66             }
 67         } catch (InterruptedException e) {
 68         } finally {
 69             lock.unlock();
 70         }
 71     }
 72 
 73     public String toString() {
 74         return "capacity:"+capacity+", actual size:"+size;
 75     }
 76 }; 
 77 
 78 // 生产者
 79 class Producer {
 80     private Depot depot;
 81     
 82     public Producer(Depot depot) {
 83         this.depot = depot;
 84     }
 85 
 86     // 消费产品:新建一个线程向仓库中生产产品。
 87     public void produce(final int val) {
 88         new Thread() {
 89             public void run() {
 90                 depot.produce(val);
 91             }
 92         }.start();
 93     }
 94 }
 95 
 96 // 消费者
 97 class Customer {
 98     private Depot depot;
 99     
100     public Customer(Depot depot) {
101         this.depot = depot;
102     }
103 
104     // 消费产品:新建一个线程从仓库中消费产品。
105     public void consume(final int val) {
106         new Thread() {
107             public void run() {
108                 depot.consume(val);
109             }
110         }.start();
111     }
112 }
113 
114 public class LockTest3 {  
115     public static void main(String[] args) {  
116         Depot mDepot = new Depot(100);
117         Producer mPro = new Producer(mDepot);
118         Customer mCus = new Customer(mDepot);
119 
120         mPro.produce(60);
121         mPro.produce(120);
122         mCus.consume(90);
123         mCus.consume(150);
124         mPro.produce(110);
125     }
126 }
Java - "JUC" ReentrantLock获取锁

(某一次)运行结果

Java - "JUC" ReentrantLock获取锁
Thread-0 produce( 60) --> left=  0, inc= 60, size= 60
Thread-1 produce(120) --> left= 80, inc= 40, size=100
Thread-2 consume( 90) <-- left=  0, dec= 90, size= 10
Thread-3 consume(150) <-- left=140, dec= 10, size=  0
Thread-4 produce(110) --> left= 10, inc=100, size=100
Thread-3 consume(150) <-- left= 40, dec=100, size=  0
Thread-4 produce(110) --> left=  0, inc= 10, size= 10
Thread-3 consume(150) <-- left= 30, dec= 10, size=  0
Thread-1 produce(120) --> left=  0, inc= 80, size= 80
Thread-3 consume(150) <-- left=  0, dec= 30, size= 50
Java - "JUC" ReentrantLock获取锁

 

ReentrantLock与synchronized的区别

前面提到ReentrantLock提供了比synchronized更加灵活和强大的锁机制,那么它的灵活和强大之处在哪里呢?他们之间又有什么相异之处呢?

首先他们肯定具有相同的功能和内存语义。

1、与synchronized相比,ReentrantLock提供了更多,更加全面的功能,具备更强的扩展性。例如:时间锁等候,可中断锁等候,锁投票。

2、ReentrantLock还提供了条件Condition,对线程的等待、唤醒操作更加详细和灵活,所以在多个条件变量和高度竞争锁的地方,ReentrantLock更加适合(以后会阐述Condition)。

3、ReentrantLock提供了可轮询的锁请求。它会尝试着去获取锁,如果成功则继续,否则可以等到下次运行时处理,而synchronized则一旦进入锁请求要么成功要么阻塞,所以相比synchronized而言,ReentrantLock会不容易产生死锁些。

4、ReentrantLock支持更加灵活的同步代码块,但是使用synchronized时,只能在同一个synchronized块结构中获取和释放。注:ReentrantLock的锁释放一定要在finally中处理,否则可能会产生严重的后果。

5、ReentrantLock支持中断处理,且性能较synchronized会好些。

 

 

ReentrantLock数据结构

ReentrantLock的UML类图

Java - "JUC" ReentrantLock获取锁

从上图我们可以看到,ReentrantLock实现Lock接口,Sync与ReentrantLock是组合关系,且FairSync(公平锁)、NonfairySync(非公平锁)是Sync的子类。Sync继承AQS(AbstractQueuedSynchronizer)。在具体分析lock时,我们需要了解几个概念:

1. AQS -- 指AbstractQueuedSynchronizer类。
    AQS是java中管理“锁”的抽象类,锁的许多公共方法都是在这个类中实现。AQS是独占锁(例如,ReentrantLock)和共享锁(例如,Semaphore)的公共父类。

2. AQS锁的类别 -- 分为“独占锁”和“共享锁”两种。
    (01) 独占锁 -- 锁在一个时间点只能被一个线程锁占有。根据锁的获取机制,它又划分为“公平锁”和“非公平锁”。公平锁,是按照通过CLH等待线程按照先来先得的规则,公平的获取锁;而非公平锁,则当线程要获取锁时,它会无视CLH等待队列而直接获取锁。独占锁的典型实例子是ReentrantLock,此外,ReentrantReadWriteLock.WriteLock也是独占锁。
    (02) 共享锁 -- 能被多个线程同时拥有,能被共享的锁。JUC包中的ReentrantReadWriteLock.ReadLock,CyclicBarrier, CountDownLatch和Semaphore都是共享锁。这些锁的用途和原理,在以后的章节再详细介绍。

3. CLH队列 -- Craig, Landin, and Hagersten lock queue
    CLH队列是AQS中“等待锁”的线程队列。在多线程中,为了保护竞争资源不被多个线程同时操作而起来错误,我们常常需要通过锁来保护这些资源。在独占锁中,竞争资源在一个时间点只能被一个线程锁访问;而其它线程则需要等待。CLH就是管理这些“等待锁”的线程的队列。
    CLH是一个非阻塞的 FIFO 队列。也就是说往里面插入或移除一个节点的时候,在并发条件下不会阻塞,而是通过自旋锁和 CAS 保证节点插入和移除的原子性。

4. CAS函数 -- Compare And Swap 
    CAS函数,是比较并交换函数,它是原子操作函数;即,通过CAS操作的数据都是以原子方式进行的。例如,compareAndSetHead(), compareAndSetTail(), compareAndSetNext()等函数。它们共同的特点是,这些函数所执行的动作是以原子的方式进行的。

 

获取公平锁(基于JDK1.7.0_40)

通过前面“Java多线程系列--“JUC锁”02之 互斥锁ReentrantLock”的“示例1”,我们知道,获取锁是通过lock()函数。下面,我们以lock()对获取公平锁的过程进行展开。

1. lock()

lock()在ReentrantLock.java的FairSync类中实现,它的源码如下:

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

说明:“当前线程”实际上是通过acquire(1)获取锁的。
        这里说明一下“1”的含义,它是设置“锁的状态”的参数。对于“独占锁”而言,锁处于可获取状态时,它的状态值是0;锁被线程初次获取到了,它的状态值就变成了1。
        由于ReentrantLock(公平锁/非公平锁)是可重入锁,所以“独占锁”可以被单个线程多此获取,每获取1次就将锁的状态+1。也就是说,初次获取锁时,通过acquire(1)将锁的状态值设为1;再次获取锁时,将锁的状态值设为2;依次类推...这就是为什么获取锁时,传入的参数是1的原因了。
        可重入就是指锁可以被单个线程多次获取。

 

2. acquire()

acquire()在AQS中实现的,它的源码如下:

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

(01) “当前线程”首先通过tryAcquire()尝试获取锁。获取成功的话,直接返回;尝试失败的话,进入到等待队列排序等待(前面还有可能有需要线程在等待该锁)。
(02) “当前线程”尝试失败的情况下,先通过addWaiter(Node.EXCLUSIVE)来将“当前线程”加入到"CLH队列(非阻塞的FIFO队列)"末尾。CLH队列就是线程等待队列。
(03) 再执行完addWaiter(Node.EXCLUSIVE)之后,会调用acquireQueued()来获取锁。由于此时ReentrantLock是公平锁,它会根据公平性原则来获取锁。
(04) “当前线程”在执行acquireQueued()时,会
进入到CLH队列中休眠等待,直到获取锁了才返回!如果“当前线程”在休眠等待过程中被中断过,acquireQueued会返回true,此时"当前线程"会调用selfInterrupt()来自己给自己产生一个中断。至于为什么要自己给自己产生一个中断,后面再介绍。

上面是对acquire()的概括性说明。下面,我们将该函数分为4部分来逐步解析。
一. tryAcquire()
二. addWaiter()
三. acquireQueued()
四. selfInterrupt()

 

一. tryAcquire()

1. tryAcquire()

公平锁的tryAcquire()在ReentrantLock.java的FairSync类中实现,源码如下:

Java - "JUC" ReentrantLock获取锁
protected final boolean tryAcquire(int acquires) {
    // 获取“当前线程”
    final Thread current = Thread.currentThread();
    // 获取“独占锁”的状态
    int c = getState();
    // c=0意味着“锁没有被任何线程锁拥有”,
    if (c == 0) {
        // 若“锁没有被任何线程锁拥有”,
        // 则判断“当前线程”是不是CLH队列中的第一个线程线程,
        // 若是的话,则获取该锁,设置锁的状态,并切设置锁的拥有者为“当前线程”。
        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;
}
Java - "JUC" ReentrantLock获取锁

说明根据代码,我们可以分析出,tryAcquire()的作用就是尝试去获取锁。注意,这里只是尝试!
         尝试成功的话,返回true;尝试失败的话,返回false,后续再通过其它办法来获取该锁。后面我们会说明,在尝试失败的情况下,是如何一步步获取锁的。

 

2. hasQueuedPredecessors()

hasQueuedPredecessors()在AQS中实现,源码如下:

Java - "JUC" ReentrantLock获取锁
public final boolean hasQueuedPredecessors() {
    Node t = tail; 
    Node h = head;
    Node s;
    return h != t &&
        ((s = h.next) == null || s.thread != Thread.currentThread());
}
Java - "JUC" ReentrantLock获取锁

说明: 通过代码,能分析出,hasQueuedPredecessors() 是通过判断"当前线程"是不是在CLH队列的队首,来返回AQS中是不是有比“当前线程”等待更久的线程。下面对head、tail和Node进行说明。

 

3. Node的源码

Node就是CLH队列的节点。Node在AQS中实现,它的数据结构如下:

Java - "JUC" ReentrantLock获取锁
private transient volatile Node head;    // CLH队列的队首
private transient volatile Node tail;    // CLH队列的队尾

// CLH队列的节点
static final class Node {
    static final Node SHARED = new Node();
    static final Node EXCLUSIVE = null;

    // 线程已被取消,对应的waitStatus的值
    static final int CANCELLED =  1;
    // “当前线程的后继线程需要被unpark(唤醒)”,对应的waitStatus的值。
    // 一般发生情况是:当前线程的后继线程处于阻塞状态,而当前线程被release或cancel掉,因此需要唤醒当前线程的后继线程。
    static final int SIGNAL    = -1;
    // 线程(处在Condition休眠状态)在等待Condition唤醒,对应的waitStatus的值
    static final int CONDITION = -2;
    // (共享锁)其它线程获取到“共享锁”,对应的waitStatus的值
    static final int PROPAGATE = -3;

    // waitStatus为“CANCELLED, SIGNAL, CONDITION, PROPAGATE”时分别表示不同状态,
    // 若waitStatus=0,则意味着当前线程不属于上面的任何一种状态。
    volatile int waitStatus;

    // 前一节点
    volatile Node prev;

    // 后一节点
    volatile Node next;

    // 节点所对应的线程
    volatile Thread thread;

    // nextWaiter是“区别当前CLH队列是 ‘独占锁’队列 还是 ‘共享锁’队列 的标记”
    // 若nextWaiter=SHARED,则CLH队列是“独占锁”队列;
    // 若nextWaiter=EXCLUSIVE,(即nextWaiter=null),则CLH队列是“共享锁”队列。
    Node nextWaiter;

    // “共享锁”则返回true,“独占锁”则返回false。
    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
    }

    // 构造函数。thread是节点所对应的线程,mode是用来表示thread的锁是“独占锁”还是“共享锁”。
    Node(Thread thread, Node mode) {     // Used by addWaiter
        this.nextWaiter = mode;
        this.thread = thread;
    }

    // 构造函数。thread是节点所对应的线程,waitStatus是线程的等待状态。
    Node(Thread thread, int waitStatus) { // Used by Condition
        this.waitStatus = waitStatus;
        this.thread = thread;
    }
}
Java - "JUC" ReentrantLock获取锁

说明
Node是CLH队列的节点,代表“等待锁的线程队列”。
(01) 每个Node都会一个线程对应。
(02) 每个Node会通过prev和next分别指向上一个节点和下一个节点,这分别代表上一个等待线程和下一个等待线程。
(03) Node通过waitStatus保存线程的等待状态。
(04) Node通过nextWaiter来区分线程是“独占锁”线程还是“共享锁”线程。如果是“独占锁”线程,则nextWaiter的值为EXCLUSIVE;如果是“共享锁”线程,则nextWaiter的值是SHARED。

 

4. compareAndSetState()

compareAndSetState()在AQS中实现。它的源码如下:

protected final boolean compareAndSetState(int expect, int update) {
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

说明: compareAndSwapInt() 是sun.misc.Unsafe类中的一个本地方法。对此,我们需要了解的是 compareAndSetState(expect, update) 是以原子的方式操作当前线程;若当前线程的状态为expect,则设置它的状态为update。

 

5. setExclusiveOwnerThread()

setExclusiveOwnerThread()在AbstractOwnableSynchronizer.java中实现,它的源码如下:

// exclusiveOwnerThread是当前拥有“独占锁”的线程
private transient Thread exclusiveOwnerThread;
protected final void setExclusiveOwnerThread(Thread t) {
    exclusiveOwnerThread = t;
}

说明:setExclusiveOwnerThread()的作用就是,设置线程t为当前拥有“独占锁”的线程。

 

6. getState(), setState()

getState()和setState()都在AQS中实现,源码如下:

Java - "JUC" ReentrantLock获取锁
// 锁的状态
private volatile int state;
// 设置锁的状态
protected final void setState(int newState) {
    state = newState;
}
// 获取锁的状态
protected final int getState() {
    return state;
}
Java - "JUC" ReentrantLock获取锁

说明:state表示锁的状态,对于“独占锁”而已,state=0表示锁是可获取状态(即,锁没有被任何线程锁持有)。由于java中的独占锁是可重入的,state的值可以>1。

 

小结tryAcquire()的作用就是让“当前线程”尝试获取锁。获取成功返回true,失败则返回false。

 

二. addWaiter(Node.EXCLUSIVE)

addWaiter(Node.EXCLUSIVE)的作用是,创建“当前线程”的Node节点,且Node中记录“当前线程”对应的锁是“独占锁”类型,并且将该节点添加到CLH队列的末尾。

1.addWaiter()

addWaiter()在AQS中实现,源码如下:

Java - "JUC" ReentrantLock获取锁
private Node addWaiter(Node mode) {
    // 新建一个Node节点,节点对应的线程是“当前线程”,“当前线程”的锁的模型是mode。
    Node node = new Node(Thread.currentThread(), mode);
    Node pred = tail;
    // 若CLH队列不为空,则将“当前线程”添加到CLH队列末尾
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 若CLH队列为空,则调用enq()新建CLH队列,然后再将“当前线程”添加到CLH队列中。
    enq(node);
    return node;
}
Java - "JUC" ReentrantLock获取锁

说明对于“公平锁”而言,addWaiter(Node.EXCLUSIVE)会首先创建一个Node节点,节点的类型是“独占锁”(Node.EXCLUSIVE)类型。然后,再将该节点添加到CLH队列的末尾。

 

2. compareAndSetTail()

compareAndSetTail()在AQS中实现,源码如下:

private final boolean compareAndSetTail(Node expect, Node update) {
    return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}

说明compareAndSetTail也属于CAS函数,也是通过“本地方法”实现的。compareAndSetTail(expect, update)会以原子的方式进行操作,它的作用是判断CLH队列的队尾是不是为expect,是的话,就将队尾设为update。

 

3. enq()

enq()在AQS中实现,源码如下:

Java - "JUC" ReentrantLock获取锁
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}
Java - "JUC" ReentrantLock获取锁

说明: enq()的作用很简单。如果CLH队列为空,则新建一个CLH表头;然后将node添加到CLH末尾。否则,直接将node添加到CLH末尾。

 

小结:addWaiter()的作用,就是将当前线程添加到CLH队列中。这就意味着将当前线程添加到等待获取“锁”的等待线程队列中了。

 

三. acquireQueued()

前面,我们已经将当前线程添加到CLH队列中了。而acquireQueued()的作用就是逐步的去执行CLH队列的线程,如果当前线程获取到了锁,则返回;否则,当前线程进行休眠,直到唤醒并重新获取锁了才返回。下面,我们看看acquireQueued()的具体流程。

 

1. acquireQueued()

acquireQueued()在AQS中实现,源码如下:

Java - "JUC" ReentrantLock获取锁
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        // interrupted表示在CLH队列的调度中,
        // “当前线程”在休眠时,有没有被中断过。
        boolean interrupted = false;
        for (;;) {
            // 获取上一个节点。
            // node是“当前线程”对应的节点,这里就意味着“获取上一个等待锁的线程”。
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
Java - "JUC" ReentrantLock获取锁

说明:acquireQueued()的目的是从队列中获取锁。

 

在这个for循环中,LZ不是很明白为什么要加p==head,Java多线程系列--“JUC锁”03之 公平锁(一)这篇博客有一个较好的解释如下:

p == head && tryAcquire(arg) 
首先,判断“前继节点”是不是CHL表头。如果是的话,则通过tryAcquire()尝试获取锁。 
其实,这样做的目的是为了“让当前线程获取锁”,但是为什么需要先判断p==head呢?理解这个对理解“公平锁”的机制很重要,因为这么做的原因就是为了保证公平性! 
      (a) 前面,我们在shouldParkAfterFailedAcquire()我们判断“当前线程”是否需要阻塞; 
      (b) 接着,“当前线程”阻塞的话,会调用parkAndCheckInterrupt()来阻塞线程。当线程被解除阻塞的时候,我们会返回线程的中断状态。而线程被解决阻塞,可能是由于“线程被中断”,也可能是由于“其它线程调用了该线程的unpark()函数”。 
      (c) 再回到p==head这里。如果当前线程是因为其它线程调用了unpark()函数而被唤醒,那么唤醒它的线程,应该是它的前继节点所对应的线程(关于这一点,后面在“释放锁”的过程中会看到)。 OK,是前继节点调用unpark()唤醒了当前线程! 
此时,再来理解p==head就很简单了:当前继节点是CLH队列的头节点,并且它释放锁之后;就轮到当前节点获取锁了。然后,当前节点通过tryAcquire()获取锁;获取成功的话,通过setHead(node)设置当前节点为头节点,并返回。 
       总之,如果“前继节点调用unpark()唤醒了当前线程”并且“前继节点是CLH表头”,此时就是满足p==head,也就是符合公平性原则的。否则,如果当前线程是因为“线程被中断”而唤醒,那么显然就不是公平了。这就是为什么说p==head就是保证公平性!

 

 

2. shouldParkAfterFailedAcquire()

shouldParkAfterFailedAcquire()在AQS中实现,源码如下:

Java - "JUC" ReentrantLock获取锁
// 返回“当前线程是否应该阻塞”
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 前继节点的状态
    int ws = pred.waitStatus;
    // 如果前继节点是SIGNAL状态,则意味这当前线程需要被unpark唤醒。此时,返回true。
    if (ws == Node.SIGNAL)
        return true;
    // 如果前继节点是“取消”状态,则设置 “当前节点”的 “当前前继节点”  为  “‘原前继节点’的前继节点”。
    if (ws > 0) {
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 如果前继节点为“0”或者“共享锁”状态,则设置前继节点为SIGNAL状态。
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
Java - "JUC" ReentrantLock获取锁

说明
(01) 关于waitStatus请参考下表(中扩号内为waitStatus的值),更多关于waitStatus的内容,可以参考前面的Node类的介绍。

CANCELLED[1]  -- 当前线程已被取消 SIGNAL[-1]    -- “当前线程的后继线程需要被unpark(唤醒)”。一般发生情况是:当前线程的后继线程处于阻塞状态,而当前线程被release或cancel掉,因此需要唤醒当前线程的后继线程。 CONDITION[-2] -- 当前线程(处在Condition休眠状态)在等待Condition唤醒 PROPAGATE[-3] -- (共享锁)其它线程获取到“共享锁” [0]           -- 当前线程不属于上面的任何一种状态。

(02) shouldParkAfterFailedAcquire()通过以下规则,判断“当前线程”是否需要被阻塞。

规则1:如果前继节点状态为SIGNAL,表明当前节点需要被unpark(唤醒),此时则返回true。 规则2:如果前继节点状态为CANCELLED(ws>0),说明前继节点已经被取消,则通过先前回溯找到一个有效(非CANCELLED状态)的节点,并返回false。 规则3:如果前继节点状态为非SIGNAL、非CANCELLED,则设置前继的状态为SIGNAL,并返回false。

如果“规则1”发生,即“前继节点是SIGNAL”状态,则意味着“当前线程”需要被阻塞。接下来会调用parkAndCheckInterrupt()阻塞当前线程,直到当前先被唤醒才从parkAndCheckInterrupt()中返回。

 

3. parkAndCheckInterrupt())

parkAndCheckInterrupt()在AQS中实现,源码如下:

private final boolean parkAndCheckInterrupt() {
    // 通过LockSupport的park()阻塞“当前线程”。
    LockSupport.park(this);
    // 返回线程的中断状态。
    return Thread.interrupted();
}

说明parkAndCheckInterrupt()的作用是阻塞当前线程,并且返回“线程被唤醒之后”的中断状态。
它会先通过LockSupport.park()阻塞“当前线程”,然后通过Thread.interrupted()返回线程的中断状态。

这里介绍一下线程被阻塞之后如何唤醒。一般有2种情况:
第1种情况:unpark()唤醒。“前继节点对应的线程”使用完锁之后,通过unpark()方式唤醒当前线程。
第2种情况:中断唤醒。其它线程通过interrupt()中断当前线程。

补充:LockSupport()中的park(),unpark()的作用 和 Object中的wait(),notify()作用类似,是阻塞/唤醒。
它们的用法不同,park(),unpark()是轻量级的,而wait(),notify()是必须先通过Synchronized获取同步锁。
关于LockSupport,我们会在之后的章节再专门进行介绍!

 

4. 再次tryAcquire()

了解了shouldParkAfterFailedAcquire()和parkAndCheckInterrupt()函数之后。我们接着分析acquireQueued()的for循环部分。

Java - "JUC" ReentrantLock获取锁
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
    setHead(node);
    p.next = null; // help GC
    failed = false;
    return interrupted;
}
Java - "JUC" ReentrantLock获取锁

说明
(01) 通过node.predecessor()获取前继节点。predecessor()就是返回node的前继节点,若对此有疑惑可以查看下面关于Node类的介绍。
(02) p == head && tryAcquire(arg)
       首先,判断“前继节点”是不是CHL表头。如果是的话,则通过tryAcquire()尝试获取锁。
       其实,这样做的目的是为了“让当前线程获取锁”,但是为什么需要先判断p==head呢?理解这个对理解“公平锁”的机制很重要,因为这么做的原因就是为了保证公平性!
       (a) 前面,我们在shouldParkAfterFailedAcquire()我们判断“当前线程”是否需要阻塞;
       (b) 接着,“当前线程”阻塞的话,会调用parkAndCheckInterrupt()来阻塞线程。当线程被解除阻塞的时候,我们会返回线程的中断状态。而线程被解决阻塞,可能是由于“线程被中断”,也可能是由于“其它线程调用了该线程的unpark()函数”。
       (c) 再回到p==head这里。如果当前线程是因为其它线程调用了unpark()函数而被唤醒,那么唤醒它的线程,应该是它的前继节点所对应的线程(关于这一点,后面在“释放锁”的过程中会看到)。 OK,是前继节点调用unpark()唤醒了当前线程!
            此时,再来理解p==head就很简单了:当前继节点是CLH队列的头节点,并且它释放锁之后;就轮到当前节点获取锁了。然后,当前节点通过tryAcquire()获取锁;获取成功的话,通过setHead(node)设置当前节点为头节点,并返回。
       总之,如果“前继节点调用unpark()唤醒了当前线程”并且“前继节点是CLH表头”,此时就是满足p==head,也就是符合公平性原则的。否则,如果当前线程是因为“线程被中断”而唤醒,那么显然就不是公平了。这就是为什么说p==head就是保证公平性!


小结acquireQueued()的作用就是“当前线程”会根据公平性原则进行阻塞等待,直到获取锁为止;并且返回当前线程在等待过程中有没有并中断过。

 

四. selfInterrupt()

selfInterrupt()是AQS中实现,源码如下:

private static void selfInterrupt() {
    Thread.currentThread().interrupt();
}

说明selfInterrupt()的代码很简单,就是“当前线程”自己产生一个中断。但是,为什么需要这么做呢?
这必须结合acquireQueued()进行分析。如果在acquireQueued()中,当前线程被中断过,则执行selfInterrupt();否则不会执行。

在acquireQueued()中,即使是线程在阻塞状态被中断唤醒而获取到cpu执行权利;但是,如果该线程的前面还有其它等待锁的线程,根据公平性原则,该线程依然无法获取到锁。它会再次阻塞! 该线程再次阻塞,直到该线程被它的前面等待锁的线程锁唤醒;线程才会获取锁,然后“真正执行起来”!
也就是说,在该线程“成功获取锁并真正执行起来”之前,它的中断会被忽略并且中断标记会被清除! 因为在parkAndCheckInterrupt()中,我们线程的中断状态时调用了Thread.interrupted()。该函数不同于Thread的isInterrupted()函数,isInterrupted()仅仅返回中断状态,而interrupted()在返回当前中断状态之后,还会清除中断状态。 正因为之前的中断状态被清除了,所以这里需要调用selfInterrupt()重新产生一个中断!

 

小结selfInterrupt()的作用就是当前线程自己产生一个中断。

 

总结

再回过头看看acquire()函数,它最终的目的是获取锁!

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

(01) 先是通过tryAcquire()尝试获取锁。获取成功的话,直接返回;尝试失败的话,再通过acquireQueued()获取锁。
(02) 尝试失败的情况下,会先通过addWaiter()来将“当前线程”加入到"CLH队列"末尾;然后调用acquireQueued(),在CLH队列中排序等待获取锁,在此过程中,线程处于休眠状态。直到获取锁了才返回。 如果在休眠等待过程中被中断过,则调用selfInterrupt()来自己产生一个中断。

 

首先通过tryAcquire方法尝试获取锁,如果成功直接返回,否则通过acquireQueued()再次获取。在acquireQueued()中会先通过addWaiter将当前线程加入到CLH队列的队尾,在CLH队列中等待。在等待过程中线程处于休眠状态,直到成功获取锁才会返回。如下:

Java - "JUC" ReentrantLock获取锁

 

 

非公平锁(NonfairSync):lock

非公平锁NonfairSync的lock()与公平锁的lock()在获取锁的流程上是一直的,但是由于它是非公平的,所以获取锁机制还是有点不同。通过前面我们了解到公平锁在获取锁时采用的是公平策略(CLH队列),而非公平锁则采用非公平策略它无视等待队列,直接尝试获取。如下:

 

非公平锁NonfairSync的lock()与公平锁的lock()在获取锁的流程上是一直的,但是由于它是非公平的,所以获取锁机制还是有点不同。通过前面我们了解到公平锁在获取锁时采用的是公平策略(CLH队列),而非公平锁则采用非公平策略它无视等待队列,直接尝试获取。如下:

Java - "JUC" ReentrantLock获取锁
final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }
Java - "JUC" ReentrantLock获取锁

lock()通过compareAndSetState尝试设置所状态,若成功直接将锁的拥有者设置为当前线程(简单粗暴),否则调用acquire()尝试获取锁;

说明
lock()会先通过compareAndSet(0, 1)来判断“锁”是不是空闲状态。是的话,“当前线程”直接获取“锁”;否则的话,调用acquire(1)获取锁。
(01) compareAndSetState()是CAS函数,它的作用是比较并设置当前锁的状态。若锁的状态值为0,则设置锁的状态值为1。
(02) setExclusiveOwnerThread(Thread.currentThread())的作用是,设置“当前线程”为“锁”的持有者。

“公平锁”和“非公平锁”关于lock()的对比

公平锁   -- 公平锁的lock()函数,会直接调用acquire(1)。 非公平锁 -- 非公平锁会先判断当前锁的状态是不是空闲,是的话,就不排队,而是直接获取锁。

 

acquire

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

在非公平锁中acquire()的实现和公平锁一模一样,但是他们尝试获取锁的机制不同(也就是tryAcquire()的实现不同)。

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

tryAcquire内部调用nonfairyTryAcquire:

Java - "JUC" ReentrantLock获取锁
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) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
Java - "JUC" ReentrantLock获取锁

与公平锁相比,非公平锁的不同之处就体现在if(c==0)的条件代码块中:

Java - "JUC" ReentrantLock获取锁
//----------------非公平锁-----  
  if (c == 0) {
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
  //----------------公平锁-----  
 if (c == 0) {
                if (!hasQueuedPredecessors() &&
                    compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
Java - "JUC" ReentrantLock获取锁

是否已经发现了不同之处。公平锁中要通过hasQueuedPredecessors()来判断该线程是否位于CLH队列中头部,是则获取锁;而非公平锁则不管你在哪个位置都直接获取锁。

 

说明
根据代码,我们可以分析出,tryAcquire()的作用就是尝试去获取锁。
(01) 如果“锁”没有被任何线程拥有,则通过CAS函数设置“锁”的状态为acquires,同时,设置“当前线程”为锁的持有者,然后返回true。
(02) 如果“锁”的持有者已经是当前线程,则将更新锁的状态即可。
(03) 如果不术语上面的两种情况,则认为尝试失败。

 

ReentrantLock示例

通过对比“示例1”和“示例2”,我们能够清晰的认识lock和unlock的作用

示例1

Java - "JUC" ReentrantLock获取锁
 1 import java.util.concurrent.locks.Lock;
 2 import java.util.concurrent.locks.ReentrantLock;
 3 
 4 // LockTest1.java
 5 // 仓库
 6 class Depot { 
 7     private int size;        // 仓库的实际数量
 8     private Lock lock;        // 独占锁
 9 
10     public Depot() {
11         this.size = 0;
12         this.lock = new ReentrantLock();
13     }
14 
15     public void produce(int val) {
16         lock.lock();
17         try {
18             size += val;
19             System.out.printf("%s produce(%d) --> size=%d\n", 
20                     Thread.currentThread().getName(), val, size);
21         } finally {
22             lock.unlock();
23         }
24     }
25 
26     public void consume(int val) {
27         lock.lock();
28         try {
29             size -= val;
30             System.out.printf("%s consume(%d) <-- size=%d\n", 
31                     Thread.currentThread().getName(), val, size);
32         } finally {
33             lock.unlock();
34         }
35     }
36 }; 
37 
38 // 生产者
39 class Producer {
40     private Depot depot;
41     
42     public Producer(Depot depot) {
43         this.depot = depot;
44     }
45 
46     // 消费产品:新建一个线程向仓库中生产产品。
47     public void produce(final int val) {
48         new Thread() {
49             public void run() {
50                 depot.produce(val);
51             }
52         }.start();
53     }
54 }
55 
56 // 消费者
57 class Customer {
58     private Depot depot;
59     
60     public Customer(Depot depot) {
61         this.depot = depot;
62     }
63 
64     // 消费产品:新建一个线程从仓库中消费产品。
65     public void consume(final int val) {
66         new Thread() {
67             public void run() {
68                 depot.consume(val);
69             }
70         }.start();
71     }
72 }
73 
74 public class LockTest1 {  
75     public static void main(String[] args) {  
76         Depot mDepot = new Depot();
77         Producer mPro = new Producer(mDepot);
78         Customer mCus = new Customer(mDepot);
79 
80         mPro.produce(60);
81         mPro.produce(120);
82         mCus.consume(90);
83         mCus.consume(150);
84         mPro.produce(110);
85     }
86 }
Java - "JUC" ReentrantLock获取锁

运行结果

Thread-0 produce(60) --> size=60
Thread-1 produce(120) --> size=180
Thread-3 consume(150) <-- size=30
Thread-2 consume(90) <-- size=-60
Thread-4 produce(110) --> size=50

结果分析
(01) Depot 是个仓库。通过produce()能往仓库中生产货物,通过consume()能消费仓库中的货物。通过独占锁lock实现对仓库的互斥访问:在操作(生产/消费)仓库中货品前,会先通过lock()锁住仓库,操作完之后再通过unlock()解锁。
(02) Producer是生产者类。调用Producer中的produce()函数可以新建一个线程往仓库中生产产品。
(03) Customer是消费者类。调用Customer中的consume()函数可以新建一个线程消费仓库中的产品。
(04) 在主线程main中,我们会新建1个生产者mPro,同时新建1个消费者mCus。它们分别向仓库中生产/消费产品。
根据main中的生产/消费数量,仓库最终剩余的产品应该是50。运行结果是符合我们预期的!

这个模型存在两个问题:
(01) 现实中,仓库的容量不可能为负数。但是,此模型中的仓库容量可以为负数,这与现实相矛盾!
(02) 现实中,仓库的容量是有限制的。但是,此模型中的容量确实没有限制的!
这两个问题,我们稍微会讲到如何解决。现在,先看个简单的示例2;通过对比“示例1”和“示例2”,我们能更清晰的认识lock(),unlock()的用途。

 

示例2

Java - "JUC" ReentrantLock获取锁
 1 import java.util.concurrent.locks.Lock;
 2 import java.util.concurrent.locks.ReentrantLock;
 3 
 4 // LockTest2.java
 5 // 仓库
 6 class Depot { 
 7     private int size;        // 仓库的实际数量
 8     private Lock lock;        // 独占锁
 9 
10     public Depot() {
11         this.size = 0;
12         this.lock = new ReentrantLock();
13     }
14 
15     public void produce(int val) {
16 //        lock.lock();
17 //        try {
18             size += val;
19             System.out.printf("%s produce(%d) --> size=%d\n", 
20                     Thread.currentThread().getName(), val, size);
21 //        } catch (InterruptedException e) {
22 //        } finally {
23 //            lock.unlock();
24 //        }
25     }
26 
27     public void consume(int val) {
28 //        lock.lock();
29 //        try {
30             size -= val;
31             System.out.printf("%s consume(%d) <-- size=%d\n", 
32                     Thread.currentThread().getName(), val, size);
33 //        } finally {
34 //            lock.unlock();
35 //        }
36     }
37 }; 
38 
39 // 生产者
40 class Producer {
41     private Depot depot;
42     
43     public Producer(Depot depot) {
44         this.depot = depot;
45     }
46 
47     // 消费产品:新建一个线程向仓库中生产产品。
48     public void produce(final int val) {
49         new Thread() {
50             public void run() {
51                 depot.produce(val);
52             }
53         }.start();
54     }
55 }
56 
57 // 消费者
58 class Customer {
59     private Depot depot;
60     
61     public Customer(Depot depot) {
62         this.depot = depot;
63     }
64 
65     // 消费产品:新建一个线程从仓库中消费产品。
66     public void consume(final int val) {
67         new Thread() {
68             public void run() {
69                 depot.consume(val);
70             }
71         }.start();
72     }
73 }
74 
75 public class LockTest2 {  
76     public static void main(String[] args) {  
77         Depot mDepot = new Depot();
78         Producer mPro = new Producer(mDepot);
79         Customer mCus = new Customer(mDepot);
80 
81         mPro.produce(60);
82         mPro.produce(120);
83         mCus.consume(90);
84         mCus.consume(150);
85         mPro.produce(110);
86     }
87 }
Java - "JUC" ReentrantLock获取锁

(某一次)运行结果

Thread-0 produce(60) --> size=-60
Thread-4 produce(110) --> size=50
Thread-2 consume(90) <-- size=-60
Thread-1 produce(120) --> size=-60
Thread-3 consume(150) <-- size=-60

结果说明
“示例2”在“示例1”的基础上去掉了lock锁。在“示例2”中,仓库中最终剩余的产品是-60,而不是我们期望的50。原因是我们没有实现对仓库的互斥访问。

 

示例3

在“示例3”中,我们通过Condition去解决“示例1”中的两个问题:“仓库的容量不可能为负数”以及“仓库的容量是有限制的”。
解决该问题是通过Condition。Condition是需要和Lock联合使用的:通过Condition中的await()方法,能让线程阻塞[类似于wait()];通过Condition的signal()方法,能让唤醒线程[类似于notify()]。

Java - "JUC" ReentrantLock获取锁
  1 import java.util.concurrent.locks.Lock;
  2 import java.util.concurrent.locks.ReentrantLock;
  3 import java.util.concurrent.locks.Condition;
  4 
  5 // LockTest3.java
  6 // 仓库
  7 class Depot {
  8     private int capacity;    // 仓库的容量
  9     private int size;        // 仓库的实际数量
 10     private Lock lock;        // 独占锁
 11     private Condition fullCondtion;            // 生产条件
 12     private Condition emptyCondtion;        // 消费条件
 13 
 14     public Depot(int capacity) {
 15         this.capacity = capacity;
 16         this.size = 0;
 17         this.lock = new ReentrantLock();
 18         this.fullCondtion = lock.newCondition();
 19         this.emptyCondtion = lock.newCondition();
 20     }
 21 
 22     public void produce(int val) {
 23         lock.lock();
 24         try {
 25              // left 表示“想要生产的数量”(有可能生产量太多,需多此生产)
 26             int left = val;
 27             while (left > 0) {
 28                 // 库存已满时,等待“消费者”消费产品。
 29                 while (size >= capacity)
 30                     fullCondtion.await();
 31                 // 获取“实际生产的数量”(即库存中新增的数量)
 32                 // 如果“库存”+“想要生产的数量”>“总的容量”,则“实际增量”=“总的容量”-“当前容量”。(此时填满仓库)
 33                 // 否则“实际增量”=“想要生产的数量”
 34                 int inc = (size+left)>capacity ? (capacity-size) : left;
 35                 size += inc;
 36                 left -= inc;
 37                 System.out.printf("%s produce(%3d) --> left=%3d, inc=%3d, size=%3d\n", 
 38                         Thread.currentThread().getName(), val, left, inc, size);
 39                 // 通知“消费者”可以消费了。
 40                 emptyCondtion.signal();
 41             }
 42         } catch (InterruptedException e) {
 43         } finally {
 44             lock.unlock();
 45         }
 46     }
 47 
 48     public void consume(int val) {
 49         lock.lock();
 50         try {
 51             // left 表示“客户要消费数量”(有可能消费量太大,库存不够,需多此消费)
 52             int left = val;
 53             while (left > 0) {
 54                 // 库存为0时,等待“生产者”生产产品。
 55                 while (size <= 0)
 56                     emptyCondtion.await();
 57                 // 获取“实际消费的数量”(即库存中实际减少的数量)
 58                 // 如果“库存”<“客户要消费的数量”,则“实际消费量”=“库存”;
 59                 // 否则,“实际消费量”=“客户要消费的数量”。
 60                 int dec = (size<left) ? size : left;
 61                 size -= dec;
 62                 left -= dec;
 63                 System.out.printf("%s consume(%3d) <-- left=%3d, dec=%3d, size=%3d\n", 
 64                         Thread.currentThread().getName(), val, left, dec, size);
 65                 fullCondtion.signal();
 66             }
 67         } catch (InterruptedException e) {
 68         } finally {
 69             lock.unlock();
 70         }
 71     }
 72 
 73     public String toString() {
 74         return "capacity:"+capacity+", actual size:"+size;
 75     }
 76 }; 
 77 
 78 // 生产者
 79 class Producer {
 80     private Depot depot;
 81     
 82     public Producer(Depot depot) {
 83         this.depot = depot;
 84     }
 85 
 86     // 消费产品:新建一个线程向仓库中生产产品。
 87     public void produce(final int val) {
 88         new Thread() {
 89             public void run() {
 90                 depot.produce(val);
 91             }
 92         }.start();
 93     }
 94 }
 95 
 96 // 消费者
 97 class Customer {
 98     private Depot depot;
 99     
100     public Customer(Depot depot) {
101         this.depot = depot;
102     }
103 
104     // 消费产品:新建一个线程从仓库中消费产品。
105     public void consume(final int val) {
106         new Thread() {
107             public void run() {
108                 depot.consume(val);
109             }
110         }.start();
111     }
112 }
113 
114 public class LockTest3 {  
115     public static void main(String[] args) {  
116         Depot mDepot = new Depot(100);
117         Producer mPro = new Producer(mDepot);
118         Customer mCus = new Customer(mDepot);
119 
120         mPro.produce(60);
121         mPro.produce(120);
122         mCus.consume(90);
123         mCus.consume(150);
124         mPro.produce(110);
125     }
126 }
Java - "JUC" ReentrantLock获取锁

(某一次)运行结果

Java - "JUC" ReentrantLock获取锁
Thread-0 produce( 60) --> left=  0, inc= 60, size= 60
Thread-1 produce(120) --> left= 80, inc= 40, size=100
Thread-2 consume( 90) <-- left=  0, dec= 90, size= 10
Thread-3 consume(150) <-- left=140, dec= 10, size=  0
Thread-4 produce(110) --> left= 10, inc=100, size=100
Thread-3 consume(150) <-- left= 40, dec=100, size=  0
Thread-4 produce(110) --> left=  0, inc= 10, size= 10
Thread-3 consume(150) <-- left= 30, dec= 10, size=  0
Thread-1 produce(120) --> left=  0, inc= 80, size= 80
Thread-3 consume(150) <-- left=  0, dec= 30, size= 50
Java - "JUC" ReentrantLock获取锁