synchronized 和 Lock

锁的相关概念介绍

可重入锁

如果锁具备可重入性,则称作为可重入锁。像synchronized和ReentrantLock都是可重入锁,可重入性在我看来实际上表明了锁的分配机制:基于线程的分配,而不是基于方法调用的分配。举个简单的例子,当一个线程执行到某个synchronized方法时,比如说method1,而在method1中会调用另外一个synchronized方法method2,此时线程不必重新去申请锁,而是可以直接执行方法method2。

看下面这段代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
/**
* LockTest class
*
* @author 蒋时华
* @date 2017/09/22
*/
public class LockTest {

public static void main(String[] args) {
final LockTest test = new LockTest();

new Thread(()->{
test.get();
}).start();

new Thread(()->{
test.get();
}).start();

}

public synchronized void get() {
method2();
}

public synchronized void method2() {

}

}

你认为结果会是怎样的呢,会不会造成死锁呢?

上述代码中的两个方法method1和method2都用synchronized修饰了,假如某一时刻,线程A执行到了method1,此时线程A获取了这个对象的锁,而由于method2也是synchronized方法,假如synchronized不具备可重入性,此时线程A需要重新申请锁。但是这就会造成一个问题,因为线程A已经持有了该对象的锁,而又在申请获取该对象的锁,这样就会线程A一直等待永远不会获取到的锁。
而由于synchronized和Lock都具备可重入性,所以不会发生上述现象

可中断锁

可中断锁:顾名思义,就是可以相应中断的锁。
在Java中,synchronized就不是可中断锁,而Lock是可中断锁。
如果某一线程A正在执行锁中的代码,另一线程B正在等待获取该锁,可能由于等待时间过长,线程B不想等待了,想先处理其他事情,我们可以让它中断自己或者在别的线程中中断它,这种就是可中断锁。
在前面演示lockInterruptibly()的用法时已经体现了Lock的可中断性。

公平锁

公平锁即尽量以请求锁的顺序来获取锁。比如同是有多个线程在等待一个锁,当这个锁被释放时,等待时间最久的线程(最先请求的线程)会获得该所,这种就是公平锁。
非公平锁即无法保证锁的获取是按照请求锁的顺序进行的。这样就可能导致某个或者一些线程永远获取不到锁。
在Java中,synchronized就是非公平锁,它无法保证等待的线程获取锁的顺序。
而对于ReentrantLock和ReentrantReadWriteLock,它默认情况下是非公平锁,但是可以设置为公平锁。

在ReentrantLock中定义了2个静态内部类,一个是NotFairSync,一个是FairSync,分别用来实现非公平锁和公平锁。
我们可以在创建ReentrantLock对象时,通过以下方式来设置锁的公平性:

1
ReentrantLock lock = new ReentrantLock(true);


另外在ReentrantLock类中定义了很多方法,比如:

  • isFair() //判断锁是否是公平锁
  • isLocked() //判断锁是否被任何线程获取了
  • isHeldByCurrentThread() //判断锁是否被当前线程获取了
  • hasQueuedThreads() //判断是否有线程在等待该锁

在ReentrantReadWriteLock中也有类似的方法,同样也可以设置为公平锁和非公平锁。不过要记住,ReentrantReadWriteLock并未实现Lock接口,它实现的是ReadWriteLock接口。

读写锁

读写锁将对一个资源(比如文件)的访问分成了2个锁,一个读锁和一个写锁。
正因为有了读写锁,才使得多个线程之间的读操作不会发生冲突。
ReadWriteLock就是读写锁,它是一个接口,ReentrantReadWriteLock实现了这个接口。
可以通过readLock()获取读锁,通过writeLock()获取写锁。

synchronized

synchronized可以保证方法或者代码块在运行时,同一时刻只有一个方法可以进入到临界区,只有一个线程可以执行某个方法或某个代码块,同时synchronized可以保证一个线程的变化可见(可见性),即可以代替volatile,保证共享变量的内存可见性

常见的几种应用是:

  • synchronized(this)、synchronize方法
  • synchronized(class)、synchronized静态同步方法
  • synchronized(Object)

synchronized(this)、synchronize方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
* Lock1 class
*
* @author 蒋时华
* @date 2017/09/22
*/
public class SynchronizedTest {

public static void main(String[] args) throws InterruptedException {
SynchronizedTest synchronizedTest = new SynchronizedTest();
ExecutorService executorService = Executors.newFixedThreadPool(100);

executorService.execute(()->{
try {
synchronizedTest.test();
} catch (Exception e) {
e.printStackTrace();
}
});

// 保证第一个线程先启动并执行
Thread.sleep(1000);

executorService.execute(()->{
try {
synchronizedTest.test2();
} catch (Exception e) {
e.printStackTrace();
}
});
}

public void test() throws InterruptedException {
synchronized (this){
System.out.println("test start");
Thread.sleep(5000);
System.out.println("test end");
}
}
public void test2() {
synchronized (this){
System.out.println("test22222222 start");
System.out.println("test22222222 end");
}
}

}

执行结果

1
2
3
4
test start
test end
test22222222 start
test22222222 end

将test2方法的锁去掉

1
2
3
4
public void test2() {
System.out.println("test22222222 start");
System.out.println("test22222222 end");
}

执行结果

1
2
3
4
test start
test22222222 start
test22222222 end
test end

将test2改成方法锁

1
2
3
4
public synchronized void test2() {
System.out.println("test22222222 start");
System.out.println("test22222222 end");
}

执行结果

1
2
3
4
test start
test end
test22222222 start
test22222222 end

总结

上面的结果我们能看到即使 test方法 耗时较长,test2方法 也并不会获得执行的机会,
如果将 test2方法 的同步锁去掉,test2方法 就可以执行
如果将 test2方法 改成方法锁,执行结果和 synchronized(this) 结果相同
so,
synchronized (this)使用的对象监视器该对象自身, 当一个线程访问SynchronizedTest中的一个synchronized (this)同步代码块时,其它线程对同一个SynchronizedTest中的synchronized (this)【包括本方法和其它被修饰的方法】同步代码块的访问将是堵塞,实现了代码顺序的同步执行
synchronized(this) == synchronize方法 【前提是this不是写在其他线程里的,如果是写在其他线程里的,则代表这个线程对象】

synchronized(Object)

object
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* Lock1 class
*
* @author 蒋时华
* @date 2017/09/22
*/
public class SynchronizedTest {

private Object objectLock = new Object();

public static void main(String[] args) throws InterruptedException {
SynchronizedTest synchronizedTest = new SynchronizedTest();
ExecutorService executorService = Executors.newFixedThreadPool(100);

executorService.execute(()->{
try {
synchronizedTest.test("线程A");
} catch (Exception e) {
e.printStackTrace();
}
});

// 保证第一个线程先启动并执行
Thread.sleep(1000);

executorService.execute(()->{
try {
synchronizedTest.test("线程B");
} catch (Exception e) {
e.printStackTrace();
}
});
}

public void test(String msg) throws InterruptedException {
synchronized (objectLock){
System.out.println(msg + " test start");
Thread.sleep(5000);
System.out.println(msg + " test end");
}
}

}

执行结果

1
2
3
4
线程A test start
线程A test end
线程B test start
线程B test end

将 objectLock 锁对象放入到 方法内

1
2
3
4
5
6
7
8
public void test(String msg) throws InterruptedException {
Object objectLock = new Object();
synchronized (objectLock){
System.out.println(msg + " test start");
Thread.sleep(5000);
System.out.println(msg + " test end");
}
}

执行结果

1
2
3
4
线程A test start
线程B test start
线程A test end
线程B test end

总结

object 和 this 本质上是一样的,只不过this是当前对象,而object是我们额外引入的对象

synchronized(class)、synchronized静态同步方法

class
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
/**
* LockTest class
*
* @author 蒋时华
* @date 2017/09/22
*/
public class LockTest {

public static void main(String[] args) {

LockTest lockTest = new LockTest();
LockTest lockTest2 = new LockTest();

new Thread(()->{
synchronized (LockTest.class){
lockTest.print();
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();

new Thread(()->{
synchronized (LockTest.class){
lockTest2.print();
}
}).start();

}

public void print(){
System.out.println("父类正在执行");
}

}

执行结果发现 线程2 也是无法执行的,至到 线程1 释放锁

静态同步synchronized方法

把 print 方法改成静态锁方法,发现结果和锁class是一样的

总结

静态同步synchronized方法 默认锁的是当前的.class对象,
通过实例可以发现 class 锁的当前对象的所有实例,区别于this,this只是锁当前的实例对象
so, synchronized(class) == synchronized静态同步方法

Lock

概念

synchronized是java中的一个关键字,也就是说是Java语言内置的特性。那么为什么会出现Lock呢?
如果一个代码块被synchronized修饰了,当一个线程获取了对应的锁,并执行该代码块时,其他线程便只能一直等待,等待获取锁的线程释放锁,而这里获取锁的线程释放锁只会有两种情况:

  • 获取锁的线程执行完了该代码块,然后线程释放对锁的占有
  • 线程执行发生异常,此时JVM会让线程自动释放锁。

那么如果这个获取锁的线程由于要等待IO或者其他原因(比如调用sleep方法)被阻塞了,但是又没有释放锁,其他线程便只能干巴巴地等待,试想一下,这多么影响程序执行效率。
因此就需要有一种机制可以不让等待的线程一直无期限地等待下去(比如只等待一定的时间或者能够响应中断),通过Lock就可以办到。
再举个例子:当有多个线程读写文件时,读操作和写操作会发生冲突现象,写操作和写操作会发生冲突现象,但是读操作和读操作不会发生冲突现象。
但是采用synchronized关键字来实现同步的话,就会导致一个问题:
如果多个线程都只是进行读操作,所以当一个线程在进行读操作时,其他线程只能等待无法进行读操作。
因此就需要一种机制来使得多个线程都只是进行读操作时,线程之间不会发生冲突,通过Lock就可以办到。

另外,通过Lock可以知道线程有没有成功获取到锁。这个是synchronized无法办到的。
总结一下,也就是说Lock提供了比synchronized更多的功能。但是要注意以下几点:

  • Lock不是Java语言内置的,synchronized是Java语言的关键字,因此是内置特性。Lock是一个类,通过这个类可以实现同步访问;
  • Lock和synchronized有一点非常大的不同,采用synchronized不需要用户去手动释放锁,当synchronized方法或者synchronized代码块执行完之后,系统会自动让线程释放对锁的占用;而Lock则必须要用户去手动释放锁,如果没有主动释放锁,就有可能导致出现死锁现象。

首先要说明的就是Lock,通过查看Lock的源码可知,Lock是一个接口:

1
2
3
4
5
6
7
8
public interface Lock {
void lock();
void lockInterruptibly() throws InterruptedException;
boolean tryLock();
boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
void unlock();
Condition newCondition();
}

lock()、tryLock()、tryLock(long time, TimeUnit unit)和lockInterruptibly()是用来获取锁的
unLock()方法是用来释放锁的

在Lock中声明了四个方法来获取锁,那么这四个方法有何区别呢?
首先lock()方法是平常使用得最多的一个方法,就是用来获取锁。如果锁已被其他线程获取,则进行等待

lock()

由于在前面讲到如果采用Lock,必须主动去释放锁,并且在发生异常时,不会自动释放锁。因此一般来说,使用Lock必须在try{}catch{}块中进行,并且将释放锁的操作放在finally块中进行,以保证锁一定被被释放,防止死锁的发生。

通常使用Lock来进行同步的话,是以下面这种形式去使用的:

1
2
3
4
5
6
7
8
9
Lock lock = ...;
lock.lock();
try{
//处理任务
}catch(Exception ex){

}finally{
lock.unlock(); //释放锁
}
tryLock()、tryLock(long time, TimeUnit unit)

tryLock()方法是有返回值的,它表示用来尝试获取锁,如果获取成功,则返回true,如果获取失败(即锁已被其他线程获取),则返回false,也就说这个方法无论如何都会立即返回。在拿不到锁时不会一直在那等待。
tryLock(long time, TimeUnit unit)方法和tryLock()方法是类似的,只不过区别在于这个方法在拿不到锁时会等待一定的时间,在时间期限之内如果还拿不到锁,就返回false。如果如果一开始拿到锁或者在等待期间内拿到了锁,则返回true

所以,一般情况下通过tryLock来获取锁时是这样使用的:

1
2
3
4
5
6
7
8
9
10
11
12
Lock lock = ...;
if(lock.tryLock()) {
try{
//处理任务
}catch(Exception ex){

}finally{
lock.unlock(); //释放锁
}
}else {
//如果不能获取锁,则直接做其他事情
}
lockInterruptibly()

lockInterruptibly()方法比较特殊,当通过这个方法去获取锁时,如果线程正在等待获取锁,则这个线程能够响应中断,即中断线程的等待状态。也就使说,当两个线程同时通过lock.lockInterruptibly()想获取某个锁时,假若此时线程A获取到了锁,而线程B只有在等待,那么对线程B调用threadB.interrupt()方法能够中断线程B的等待过程。

由于lockInterruptibly()的声明中抛出了异常,所以lock.lockInterruptibly()必须放在try块中或者在调用lockInterruptibly()的方法外声明抛出InterruptedException。

因此lockInterruptibly()一般的使用形式如下:

1
2
3
4
5
6
7
8
9
public void method() throws InterruptedException {
lock.lockInterruptibly();
try {
//.....
}
finally {
lock.unlock();
}
}

注意:
当一个线程获取了锁之后,是不会被interrupt()方法中断的。单独调用interrupt()方法不能中断正在运行过程中的线程,只能中断阻塞过程中的线程。

ReentrantLock

ReentrantLock,意思是“可重入锁”,ReentrantLock是唯一实现了Lock接口的类,并且ReentrantLock提供了更多的方法。
具体的使用和synchronized 差不多,都需要注意锁的对象,不要把lock对象定义在方法中,导致线程不能共享锁

lock()、tryLock()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
import java.util.ArrayList;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

**
* LockTest class
*
* @author 蒋时华
* @date 2017/09/22
*/
public class LockTest {

private ArrayList<Integer> arrayList = new ArrayList<Integer>();
Lock lock = new ReentrantLock();

public static void main(String[] args) {
final LockTest test = new LockTest();

new Thread(()->{
test.lock(Thread.currentThread());
}).start();

new Thread(()->{
test.lock(Thread.currentThread());
}).start();
}

public void lock(Thread thread) {
lock.lock();
try {
System.out.println(thread.getName()+"得到了锁");

// TODO: 2019/10/22 执行业务逻辑
} catch (Exception e) {
// TODO: handle exception
}finally {
System.out.println(thread.getName()+"释放了锁");
lock.unlock();
}
}

public void tryLock(Thread thread) {
if(lock.tryLock()) {
try {
System.out.println(thread.getName()+"得到了锁");

// TODO: 2019/10/22 执行业务逻辑
} catch (Exception e) {
// TODO: handle exception
}finally {
System.out.println(thread.getName()+"释放了锁");
lock.unlock();
}
} else {
System.out.println(thread.getName()+"获取锁失败");
}
}

}

lockInterruptibly()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
* LockTest class
*
* @author 蒋时华
* @date 2017/09/22
*/
public class LockTest {

Lock lock = new ReentrantLock();

public static void main(String[] args) throws InterruptedException {
final LockTest test = new LockTest();

Thread thread1 = new Thread(() -> {
try {
test.lockInterruptibly(Thread.currentThread());
} catch (InterruptedException e) {
System.out.println(Thread.currentThread().getName() + "被中断");
}
});
thread1.start();


Thread thread2 = new Thread(() -> {
try {
test.lockInterruptibly(Thread.currentThread());
} catch (InterruptedException e) {
System.out.println(Thread.currentThread().getName() + "被中断");
}
});
thread2.start();


Thread.sleep(2000);

thread2.interrupt();

Thread.sleep(10000);
}

public void lockInterruptibly(Thread thread) throws InterruptedException {
//注意,如果需要正确中断等待锁的线程,必须将获取锁放在外面,然后将InterruptedException抛出
lock.lockInterruptibly();
try {
System.out.println(thread.getName()+"得到了锁");

Thread.sleep(5000);

// TODO: 2019/10/22 执行业务逻辑
}
finally {
System.out.println(Thread.currentThread().getName()+"执行finally");
lock.unlock();
System.out.println(thread.getName()+"释放了锁");
}
}

}

运行之后,发现thread2能够被正确中断。

ReadWriteLock

ReadWriteLock也是一个接口,在它里面只定义了两个方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public interface ReadWriteLock {
/**
* Returns the lock used for reading.
*
* @return the lock used for reading.
*/
Lock readLock();

/**
* Returns the lock used for writing.
*
* @return the lock used for writing.
*/
Lock writeLock();
}

一个用来获取读锁,一个用来获取写锁。也就是说将文件的读写操作分开,分成2个锁来分配给线程,从而使得多个线程可以同时进行读操作。下面的ReentrantReadWriteLock实现了ReadWriteLock接口。

ReentrantReadWriteLock

ReentrantReadWriteLock里面提供了很多丰富的方法,不过最主要的有两个方法:readLock()和writeLock()用来获取读锁和写锁。
假如有多个线程要同时进行读操作的话,先看一下synchronized达到的效果:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
* LockTest class
*
* @author 蒋时华
* @date 2017/09/22
*/
public class LockTest {

private ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();

public static void main(String[] args) {
final LockTest test = new LockTest();

new Thread(()->{
test.get(Thread.currentThread());
}).start();

new Thread(()->{
test.get(Thread.currentThread());
}).start();

}

public synchronized void get(Thread thread) {
long start = System.currentTimeMillis();
while(System.currentTimeMillis() - start <= 1) {
System.out.println(thread.getName()+"正在进行读操作");
}
System.out.println(thread.getName()+"读操作完毕");
}

}

这段程序的输出结果会是,直到thread1执行完读操作之后,才会打印thread2执行读操作的信息。

1
2
3
4
5
6
7
8
9
10
11
12
13
Thread-0正在进行读操作
Thread-0正在进行读操作
Thread-0正在进行读操作
Thread-0正在进行读操作
Thread-0正在进行读操作
Thread-0读操作完毕
Thread-1正在进行读操作
Thread-1正在进行读操作
Thread-1正在进行读操作
Thread-1正在进行读操作
Thread-1正在进行读操作
Thread-1正在进行读操作
Thread-1读操作完毕

而改成用读写锁的话:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
* LockTest class
*
* @author 蒋时华
* @date 2017/09/22
*/
public class LockTest {

private ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();

public static void main(String[] args) {
final LockTest test = new LockTest();

new Thread(()->{
test.get(Thread.currentThread());
}).start();

new Thread(()->{
test.get(Thread.currentThread());
}).start();

}

public void get(Thread thread) {
rwl.readLock().lock();
try {
long start = System.currentTimeMillis();

while(System.currentTimeMillis() - start <= 1) {
System.out.println(thread.getName()+"正在进行读操作");
}
System.out.println(thread.getName()+"读操作完毕");
} finally {
rwl.readLock().unlock();
}
}

}

执行结果

1
2
3
4
5
6
7
Thread-0正在进行读操作
Thread-0正在进行读操作
Thread-1正在进行读操作
Thread-0读操作完毕
Thread-1正在进行读操作
Thread-1正在进行读操作
Thread-1读操作完毕

说明thread1和thread2在同时进行读操作。
这样就大大提升了读操作的效率。
不过要注意的是,如果有一个线程已经占用了读锁,则此时其他线程如果要申请写锁,则申请写锁的线程会一直等待释放读锁。
如果有一个线程已经占用了写锁,则此时其他线程如果申请写锁或者读锁,则申请的线程会一直等待释放写锁。

Condition 类

Condition与重入锁是通过lock.newCondition()方法产生一个与当前重入锁绑定的Condtion实例,我们通知该实例来控制线程的等待与通知。该接口的所有方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

public interface Condition {
//当前线程进入等待状态,直到被通知(signal)或者被中断时,当前线程进入运行状态,从await()返回。与Object.wait()类似。
void await() throws InterruptedException;

//当前线程进入等待状态,直到被通知,对中断不做响应;
//线程在调用condition.await()后处于await状态,此时调用thread.interrupt()会报错
//但是使用condition.awaitUninterruptibly()后,调用thread.interrupt()则不会报错
void awaitUninterruptibly();

//nanosTimeout指定该方法等待信号的的最大时间(单位为纳秒)。若指定时间内收到signal()或signalALL()则返回nanosTimeout减去已经等待的时间;
//若指定时间内有其它线程中断该线程,则抛出InterruptedException并清除当前线程的打断状态;
//增加了超时响应,返回值表示当前剩余的时间,如果在nanosTimeout之前被唤醒,返回值 = nanosTimeout - 实际消耗的时间,返回值 <= 0表示超时;
long awaitNanos(long nanosTimeout) throws InterruptedException;

//与await()基本一致,唯一不同点在于,返回值返回true/false,在time之前被唤醒,返回true,超时返回false。
boolean await(long time, TimeUnit unit) throws InterruptedException;

//适用条件与行为与awaitNanos(long nanosTimeout)完全一样,唯一不同点在于它不是等待指定时间,而是等待由参数指定的某一时刻。
boolean awaitUntil(Date deadline) throws InterruptedException;

//唤醒一个在 await()等待队列中的线程。与Object.notify()相似
void signal();

//唤醒 await()等待队列中所有的线程。与object.notifyAll()相似
void signalAll();
}

CountDownLatch

  • countDownLatch这个类使一个线程等待其他线程各自执行完毕后再执行。
  • 是通过一个计数器来实现的,计数器的初始值是线程的数量。每当一个线程执行完毕后,计数器的值就-1,当计数器的值为0时,表示所有线程都执行完毕,然后在闭锁上等待的线程就可以恢复工作了。

countDownLatch类中只提供了一个构造器:

1
2
//参数count为计数值
public CountDownLatch(int count) { };

类中有三个方法是最重要的:

1
2
3
4
5
6
//调用await()方法的线程会被挂起,它会等待直到count值为0才继续执行
public void await() throws InterruptedException { };
//和await()类似,只不过等待一定的时间后count值还没变为0的话就会继续执行
public boolean await(long timeout, TimeUnit unit) throws InterruptedException { };
//将count值减1
public void countDown() { };

CyclicBarrier

1
2
3
4
5
6
//调用await方法的线程告诉CyclicBarrier自己已经到达同步点,然后当前线程被阻塞。直到parties个参与线程调用了await方法,CyclicBarrier同样提供带超时时间的await和不带超时时间的await方法
await()
//broken标识该当前CyclicBarrier是否已经处于中断状态。
//默认barrier是没有损坏的。当barrier损坏了或者有一个线程中断了,则通过breakBarrier()来终止所有的线程
breakBarrier()
...

dowait(boolean, long)方法,它也是CyclicBarrier的核心方法,该方法定义如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
// 获取独占锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 当前代
final Generation g = generation;
// 如果这代损坏了,抛出异常
if (g.broken)
throw new BrokenBarrierException();

// 如果线程中断了,抛出异常
if (Thread.interrupted()) {
// 将损坏状态设置为true
// 并通知其他阻塞在此栅栏上的线程
breakBarrier();
throw new InterruptedException();
}

// 获取下标
int index = --count;
// 如果是 0,说明最后一个线程调用了该方法
if (index == 0) { // tripped
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
// 执行栅栏任务
if (command != null)
command.run();
ranAction = true;
// 更新一代,将count重置,将generation重置
// 唤醒之前等待的线程
nextGeneration();
return 0;
} finally {
// 如果执行栅栏任务的时候失败了,就将损坏状态设置为true
if (!ranAction)
breakBarrier();
}
}

// loop until tripped, broken, interrupted, or timed out
for (;;) {
try {
// 如果没有时间限制,则直接等待,直到被唤醒
if (!timed)
trip.await();
// 如果有时间限制,则等待指定时间
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
// 当前代没有损坏
if (g == generation && ! g.broken) {
// 让栅栏失效
breakBarrier();
throw ie;
} else {
// 上面条件不满足,说明这个线程不是这代的
// 就不会影响当前这代栅栏的执行,所以,就打个中断标记
Thread.currentThread().interrupt();
}
}

// 当有任何一个线程中断了,就会调用breakBarrier方法
// 就会唤醒其他的线程,其他线程醒来后,也要抛出异常
if (g.broken)
throw new BrokenBarrierException();

// g != generation表示正常换代了,返回当前线程所在栅栏的下标
// 如果 g == generation,说明还没有换代,那为什么会醒了?
// 因为一个线程可以使用多个栅栏,当别的栅栏唤醒了这个线程,就会走到这里,所以需要判断是否是当前代。
// 正是因为这个原因,才需要generation来保证正确。
if (g != generation)
return index;

// 如果有时间限制,且时间小于等于0,销毁栅栏并抛出异常
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
// 释放独占锁
lock.unlock();
}
}

dowait(boolean, long)方法的主要逻辑处理比较简单,如果该线程不是最后一个调用await方法的线程,则它会一直处于等待状态,除非发生以下情况:

  • 最后一个线程到达,即index == 0
  • 某个参与线程等待超时
  • 某个参与线程被中断
  • 调用了CyclicBarrier的reset()方法。该方法会将屏障重置为初始状态
    在上面的源代码中,我们可能需要注意Generation 对象,在上述代码中我们总是可以看到抛出BrokenBarrierException异常,那么什么时候抛出异常呢?如果一个线程处于等待状态时,如果其他线程调用reset(),或者调用的barrier原本就是被损坏的,则抛出BrokenBarrierException异常。同时,任何线程在等待时被中断了,则其他所有线程都将抛出BrokenBarrierException异常,并将barrier置于损坏状态。

同时,Generation描述着CyclicBarrier的更新换代。在CyclicBarrier中,同一批线程属于同一代。当有parties个线程到达barrier之后,generation就会被更新换代。其中broken标识该当前CyclicBarrier是否已经处于中断状态。

1
2
3
private static class Generation {
boolean broken = false;
}

默认barrier是没有损坏的。当barrier损坏了或者有一个线程中断了,则通过breakBarrier()来终止所有的线程:

1
2
3
4
5
private void breakBarrier() {
generation.broken = true;
count = parties;
trip.signalAll();
}

在breakBarrier()中除了将broken设置为true,还会调用signalAll将在CyclicBarrier处于等待状态的线程全部唤醒。

当所有线程都已经到达barrier处(index == 0),则会通过nextGeneration()进行更新换地操作,在这个步骤中,做了三件事:唤醒所有线程,重置count,generation:

1
2
3
4
5
6
7
private void nextGeneration() {
// signal completion of last generation
trip.signalAll();
// set up next generation
count = parties;
generation = new Generation();
}

除了上面讲到的栅栏更新换代以及损坏状态,我们在使用CyclicBarrier时还要要注意以下几点:

  • CyclicBarrier使用独占锁来执行await方法,并发性可能不是很高
  • 如果在等待过程中,线程被中断了,就抛出异常。但如果中断的线程所对应的CyclicBarrier不是这代的,比如,在最后一次线程执行signalAll后,并且更新了这个“代”对象。在这个区间,这个线程被中断了,那么,JDK认为任务已经完成了,就不必在乎中断了,只需要打个标记。该部分源码已在dowait(boolean, long)方法中进行了注释。
  • 如果线程被其他的CyclicBarrier唤醒了,那么g肯定等于generation,这个事件就不能return了,而是继续循环阻塞。反之,如果是当前CyclicBarrier唤醒的,就返回线程在CyclicBarrier的下标。完成了一次冲过栅栏的过程。该部分源码已在dowait(boolean, long)方法中进行了注释。

我们自定义的工作线程必须要等所有参与线程开始之后才可以执行,我们可以使用CyclicBarrier类来帮助我们完成。从程序的执行结果中也可以看出,所有的工作线程都运行await()方法之后都到达了栅栏位置,然后,工作线程才开始执行业务处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
public class CyclicBarrier {

private static class Generation {
boolean broken = false;
}

// 独占锁成员
private final ReentrantLock lock = new ReentrantLock();
// 条件成员
private final Condition trip = lock.newCondition();
// 必须满足障碍条件的线程个数
private final int parties;
// 当障碍条件满足会被自动执行的任务
private final Runnable barrierCommand;
// 当前世代(cyclicBarrier可重复利用,每一次利用是一个世代)
private Generation generation = new Generation();
// 满足障碍锁条件还需要的阻塞线程个数
private int count;

...
..
.
// 构造函数
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
// 初始化障碍条件总数
this.parties = parties;
// 初始化阻塞线程个数
this.count = parties;
// 初始化条件满足时自动执行任务
this.barrierCommand = barrierAction;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
public int await() throws InterruptedException, BrokenBarrierException {
try {
// 调用dowait来实现。
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen;
}
}


private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
// 获取独占锁。下面的操作都是同步的。
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 设置当前世代
final Generation g = generation;

if (g.broken)
throw new BrokenBarrierException();

if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
// 由于本线程调用await(),则需要的阻塞线程个数-1。
int index = --count;
// 若为0则代表障碍条件满足
if (index == 0) { // tripped
boolean ranAction = false;
try {
// 执行注册的自动执行任务
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
// 设置世代为下一世代,以方便障碍锁的二次利用。
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}

// 通过循环
for (;;) {
try {
// 若没有阻塞时间限制,则阻塞
if (!timed)
trip.await();
// 若有阻塞时间限制,则阻塞相应的时间
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
if (g == generation && ! g.broken) {
breakBarrier();
throw ie;
} else {
Thread.currentThread().interrupt();
}
}

if (g.broken)
throw new BrokenBarrierException();
// 阻塞线程被唤醒后,若generation已被更新则代表障碍条件达成,线程继续执行。
if (g != generation)
return index;
// 阻塞线程若超过了阻塞时间,被唤醒后,则抛出异常
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
// 释放同步锁。
lock.unlock();
}
}


.....
...
..
// 更新当前障碍锁世代。
private void nextGeneration() {
// 唤醒所有在当前障碍锁上阻塞的线程
trip.signalAll();
// reset障碍锁条件
count = parties;
// 初始化新世代
generation = new Generation();
}

CyclicBarrier和CountDownLatch的区别

  • CountDownLatch的计数器只能使用一次,而CyclicBarrier的计数器可以使用reset()方法重置,可以使用多次,所以CyclicBarrier能够处理更为复杂的场景;
  • CyclicBarrier还提供了一些其他有用的方法,比如getNumberWaiting()方法可以获得CyclicBarrier阻塞的线程数量,isBroken()方法用来了解阻塞的线程是否被中断;
  • CountDownLatch允许一个或多个线程等待一组事件的产生,而CyclicBarrier用于等待其他线程运行到栅栏位置。
  • CountDownLatch : 一个线程(或者多个), 等待另外N个线程完成某个事情之后才能执行。 CyclicBarrier : N个线程相互等待,任何一个线程完成之前,所有的线程都必须等待。这样应该就清楚一点了,对于CountDownLatch来说,重点是那个“一个线程”, 是它在等待, 而另外那N的线程在把“某个事
  • 情”做完之后可以继续等待,可以终止。而对于CyclicBarrier来说,重点是那N个线程,他们之间任何一个没有完成,所有的线程都必须等待。

Lock和synchronized的选择

总结来说,Lock和synchronized有以下几点不同:

  • Lock是一个接口,而synchronized是Java中的关键字,synchronized是内置的语言实现;
  • synchronized在发生异常时,会自动释放线程占有的锁,因此不会导致死锁现象发生;而Lock在发生异常时,如果没有主动通过unLock()去释放锁,则很可能造成死锁现象,因此使用Lock时需要在finally块中释放锁;
  • Lock可以让等待锁的线程响应中断,而synchronized却不行,使用synchronized时,等待的线程会一直等待下去,不能够响应中断;
  • 通过Lock可以知道有没有成功获取锁,而synchronized却无法办到。
  • Lock可以提高多个线程进行读操作的效率。
    在性能上来说,如果竞争资源不激烈,两者的性能是差不多的,而当竞争资源非常激烈时(即有大量线程同时竞争),此时Lock的性能要远远优于synchronized。所以说,在具体使用时要根据适当情况选择。
Jeff-Eric wechat