Python基础入门之魔法方法与异常处
833 2023-04-03 04:01:53
真实的多线程业务开发中,最常用到的逻辑就是数据的读写,ReentrantLock虽然具有完全互斥排他的效果(即同一时间只有一个线程正在执行lock后面的任务),
这样做虽然保证了实例变量的线程安全性,但效率却是非常低下的。所以在JDK中提供了一种读写锁ReentrantReadWriteLock类,使用它可以加快运行效率。
读写锁表示两个锁,一个是读操作相关的锁,称为共享锁;另一个是写操作相关的锁,称为排他锁。
下面我们通过代码去验证下读写锁之间的互斥性
回到顶部首先创建一个对象,分别定义一个加读锁方法和一个加写锁的方法,
?1234567891011121314151617181920212223242526272829public
class
MyDomain3 {
private
ReentrantReadWriteLock lock =
new
ReentrantReadWriteLock();
public
void
testReadLock() {
try
{
lock.readLock().lock();
System.out.println(System.currentTimeMillis() +
" 获取读锁"
);
Thread.sleep(
1000
);
}
catch
(InterruptedException e) {
e.printStackTrace();
}
finally
{
lock.readLock().unlock();
}
}
public
void
testWriteLock() {
try
{
lock.writeLock().lock();
System.out.println(System.currentTimeMillis() +
" 获取写锁"
);
Thread.sleep(
1000
);
}
catch
(InterruptedException e) {
e.printStackTrace();
}
finally
{
lock.writeLock().unlock();
}
}
}
创建线程类1 调用加读锁方法
?12345678910111213public
class
Mythread3_1
extends
Thread {
private
MyDomain3 myDomain3;
public
Mythread3_1(MyDomain3 myDomain3) {
this
.myDomain3 = myDomain3;
}
@Override
public
void
run() {
myDomain3.testReadLock();
}
}
?12345678910
@Test
public
void
test3()
throws
InterruptedException {
MyDomain3 myDomain3 =
new
MyDomain3();
Mythread3_1 readLock =
new
Mythread3_1(myDomain3);
Mythread3_1 readLock2 =
new
Mythread3_1(myDomain3);
readLock.start();
readLock2.start();
Thread.sleep(
3000
);
}
执行结果:
?121639621812838
获取读锁
1639621812839
获取读锁
可以看出两个读锁几乎同时执行,说明读和读之间是共享的,因为读操作不会有线程安全问题。
创建线程类2,调用加写锁方法
?12345678910111213public
class
Mythread3_2
extends
Thread {
private
MyDomain3 myDomain3;
public
Mythread3_2(MyDomain3 myDomain3) {
this
.myDomain3 = myDomain3;
}
@Override
public
void
run() {
myDomain3.testWriteLock();
}
}
?1234567891011@Test
public
void
test3()
throws
InterruptedException {
MyDomain3 myDomain3 =
new
MyDomain3();
Mythread3_2 writeLock =
new
Mythread3_2(myDomain3);
Mythread3_2 writeLock2 =
new
Mythread3_2(myDomain3);
writeLock.start();
writeLock2.start();
Thread.sleep(
3000
);
}
执行结果:
?121639622063226
获取写锁
1639622064226
获取写锁
从时间上看,间隔是1000ms即1s,说明写锁和写锁之间互斥。
再用线程1和线程2分别调用读锁与写锁
?1234567891011@Test
public
void
test3()
throws
InterruptedException {
MyDomain3 myDomain3 =
new
MyDomain3();
Mythread3_1 readLock =
new
Mythread3_1(myDomain3);
Mythread3_2 writeLock =
new
Mythread3_2(myDomain3);
readLock.start();
writeLock.start();
Thread.sleep(
3000
);
}
执行结果:
?121639622338402
获取读锁
1639622339402
获取写锁
从时间上看,间隔是1000ms即1s,和代码里面是一致的,证明了读和写之间是互斥的。
注意一下,"读和写互斥"和"写和读互斥"是两种不同的场景,但是证明方式和结论是一致的,所以就不证明了。
最终测试结果下:
1、读和读之间不互斥,因为读操作不会有线程安全问题
2、写和写之间互斥,避免一个写操作影响另外一个写操作,引发线程安全问题
3、读和写之间互斥,避免读操作的时候写操作修改了内容,引发线程安全问题
总结起来就是,多个Thread可以同时进行读取操作,但是同一时刻只允许一个Thread进行写入操作。
回到顶部
读写锁中的Sync也是同样实现了AQS,回想ReentrantLock中自定义同步器的实现,同步状态表示锁被一个线程重复获取的次数,
而读写锁的自定义同步器需要在同步状态(一个整型变量)上维护多个读线程和一个写线程的状态,使得该状态的设计成为读写锁实现的关键。
读写锁将变量切分成了两个部分,高16位表示读,低16位表示写
当前同步状态表示一个线程已经获取了写锁,且重进入了两次,同时也连续获取了两次读锁。读写锁是如何迅速确定读和写各自的状态呢?
?123456789static
final
int
SHARED_SHIFT =
16
;
static
final
int
SHARED_UNIT = (
1
<< SHARED_SHIFT);
static
final
int
MAX_COUNT = (
1
<< SHARED_SHIFT) -
1
;
static
final
int
EXCLUSIVE_MASK = (
1
<< SHARED_SHIFT) -
1
;
/** Returns the number of shared holds represented in count */
static
int
sharedCount(
int
c) {
return
c >>> SHARED_SHIFT; }
/** Returns the number of exclusive holds represented in count */
static
int
exclusiveCount(
int
c) {
return
c & EXCLUSIVE_MASK; }
其实是通过位运算。假设当前同步状态值为c,写状态等于c & EXCLUSIVE_MASK (c&0x0000FFFF(将高16位全部抹去)),
读状态等于c>>>16(无符号补0右移16位)。当写状态增加1时,等于c+1,当读状态增加1时,等于c+(1<<16),也就是c+0x00010000。
根据状态的划分能得出一个推论:c不等于0时,当写状态(c & 0x0000FFFF)等于0时,则读状态(c>>>16)大于0,即读锁已被获取。
通过上面的测试,我们知道写锁是一个支持重入的排它锁,看下源码是如何实现写锁的获取
?12345678910111213141516171819202122232425262728293031protected
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);
if
(c !=
0
) {
// (Note: if c != 0 and w == 0 then shared count != 0)
if
(w ==
0
|| current != getExclusiveOwnerThread())
return
false
;
if
(w + exclusiveCount(acquires) > MAX_COUNT)
throw
new
Error(
"Maximum lock count exceeded"
);
// Reentrant acquire
setState(c + acquires);
return
true
;
}
if
(writerShouldBlock() ||
!compareAndSetState(c, c + acquires))
return
false
;
setExclusiveOwnerThread(current);
return
true
;
}
第3行到第11行,简单说了下整个方法的实现逻辑,这里要夸一下,这段注释就很容易的让人知道代码的功能。下面我们分析一下,
第13到第15行,分别拿到了当前线程对象current,lock的加锁状态值c 以及写锁的值w,c!=0 表明 当前处于有锁状态,
再继续分析第16行到25行,有个关键的Note:(Note: if c != 0 and w == 0 then shared count != 0):简单说就是:如果一个有锁状态但是没有写锁,那么肯定加了读锁。
第18行if条件,就是判断加了读锁,但是当前线程不是锁拥有的线程,那么获取锁失败,证明读写锁互斥。
第20行到第25行,走到这步,说明 w !=0 ,已经获取了写锁,只要不超过写锁最大值,那么增加写状态然后就可以成功获取写锁。
如果代码走到第26行,说明c==0,当前没有加任何锁,先执行 writerShouldBlock()方法,此方法用来判断写锁是否应该阻塞,
这块是对公平与非公平锁会有不同的逻辑,对于非公平锁,直接返回false,不需要阻塞,
下面是公平锁执行的判断
?12345678910public
final
boolean
hasQueuedPredecessors() {
// The correctness of this depends on head being initialized
// before tail and on head.next being accurate if the current
// thread is first in queue.
Node t = tail;
// Read fields in reverse initialization order
Node h = head;
Node s;
return
h != t &&
((s = h.next) ==
null
|| s.thread != Thread.currentThread());
}
对于公平锁需要判断当前等待队列中是否存在 等于当前线程并且正在排队等待获取锁的线程。
写锁的释放与ReentrantLock的释放过程基本类似,每次释放均减少写状态,当写状态为0时表示写锁已被释放,
从而等待的读写线程能够继续访问读写锁,同时前次写线程的修改对后续读写线程可见。
读锁是一个支持重进入的共享锁,它能够被多个线程同时获取。JDK源码如下:
?123456789101112131415161718192021222324252627protected
final
int
tryAcquireShared(
int
unused) {
Thread current = Thread.currentThread();
int
c = getState();
if
(exclusiveCount(c) !=
0
&&
getExclusiveOwnerThread() != current)
return
-
1
;
int
r = sharedCount(c);
if
(!readerShouldBlock() &&
r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {
if
(r ==
0
) {
firstReader = current;
firstReaderHoldCount =
1
;
}
else
if
(firstReader == current) {
firstReaderHoldCount++;
}
else
{
HoldCounter rh = cachedHoldCounter;
if
(rh ==
null
|| rh.tid != getThreadId(current))
cachedHoldCounter = rh = readHolds.get();
else
if
(rh.count ==
0
)
readHolds.set(rh);
rh.count++;
}
return
1
;
}
return
fullTryAcquireShared(current);
}
第4行到第6行,如果写锁被其他线程持有,则直接返回false,获取读锁失败,证明不同线程间写读互斥。
第8行,readerShouldBlock() 获取读锁是否应该阻塞,这儿也同样要区分公平锁和非公平锁,
公平锁模式需要判断当前等待队列中是否存在 等于当前线程并且正在排队等待获取锁的线程,存在则获取读锁需要等待。
非公平锁模式需要判断当前等待队列中第一个是等待写锁的,则方法返回true,获取读锁需要等待。
fullTryAcquireShared() 主要是处理读锁获取的完整版本,它处理tryAcquireShared()中没有处理的CAS错误和可重入读锁的处理逻辑。
回到顶部
1:《Java并发编程的艺术》