多线程之读写锁原理

今天主要通过多读单写的例子来说下读写锁的原理


概念
多读单写，简单说，就是对资源的访问分为两种状态，一种是读操作，另一种是写操作。由应用程序提示锁应该做哪种操作。当为读模式时，所有的写动作被悬挂，而读请求被允许通过，而写动作时，所有操作被悬挂。并且，读写切换时，有足够的状态等待，直到真正安全时，才会切换动作。


如下图所示：


 
 
业务场景举例
比如现在有 A、B、C、D、E、F、G 6个线程，其中A、B、C、G 4个线程之行读请求，E、F 2个线程之行写请求，如何保证读写安全？

分析：

1、读写请求是可以在多个线程进行的

2、写请求时，所有的请求都会被停止即悬挂

解决：使用读写锁

代码：

demo里面的代码就是业务场景的表达，即有多个线程同时执行读写请求的业务场景

- (void)demo {
    self.queue = dispatch_queue_create("rw_queue", DISPATCH_QUEUE_CONCURRENT);
    
    for (int i = 0; i < 2; i++) {
        dispatch_async(self.queue, ^{
            [self read:1];
        });
        
        dispatch_async(self.queue, ^{
            [self read:2];
        });
        
        dispatch_barrier_async(self.queue, ^{
            [self write];
        });
        
        dispatch_async(self.queue, ^{
            [self read:3];
        });
        
    }
}

下面的 read 和 write 方法里，就是读写锁的使用

- (void)read {
    pthread_rwlock_rdlock(&_lock);
    
    sleep(1);
    NSLog(@"%s", __func__);
    
    pthread_rwlock_unlock(&_lock);
}
 
- (void)write
{
    pthread_rwlock_wrlock(&_lock);
    
    sleep(1);
    NSLog(@"%s", __func__);
    
    pthread_rwlock_unlock(&_lock);
}

读写锁的原理
在 AQS 中，通过 int 类型的全局变量 state 来表示同步状态，即用 state 来表示锁。

ReentrantReadWriteLock 也是通过 AQS 来实现锁的，但是 ReentrantReadWriteLock有两把锁：读锁和写锁，它们保护的都是同一个资源，那么如何用一个共享变量来区分锁是写锁还是读锁呢？答案就是按位拆分。

由于 state 是 int 类型的变量，在内存中占用4个字节，也就是32位。将其拆分为两部分：高16位和低16位，其中高16位用来表示读锁状态，低16位用来表示写锁状态。

  • 当设置读锁成功时，就将高16位加1，释放读锁时，将高16位减1；
  • 当设置写锁成功时，就将低16位加1，释放写锁时，将第16位减1；

如下图所示:

 

写锁加锁的原理

获取写锁的流程
c == 0表示锁还没有被任何线程占用
w 写锁的数量

  1. 如果 c==0，标记锁成功后，表述获取写锁成功
  2. 如果 c!=0 && w==0，表示读锁在占用锁，所以获取锁失败
  3. 如果 c!=0 && w!=0,表示写锁在占用锁，此时就需要判断访问该锁的线程是否和占用该锁的线程为同一线程，如果不为同一线程就返回失败；如果为同一线程，则判断重入的数量，数量为超过就返回成功，否则抛出异常 
protected final boolean tryAcquire(int acquires) {
Thread current = Thread.currentThread();
int c = getState();
// exclusiveCount()方法的作用是将同步变量与0xFFFF做&运算，计算结果就是写锁的数量。
// 因此w的值的含义就是写锁的数量
int w = exclusiveCount(c);
// 如果c不为0就表示锁被占用了，但是占用的是写锁还是读锁呢？这个时候就需要根据w的值来判断了。
// 如果c等于0就表示此时锁还没有被任何线程占用，那就让线程直接去尝试获取锁
if (c != 0) {
    // (Note: if c != 0 and w == 0 then shared count != 0)
    //
    /**
     * 1. 如果w为0，说明写锁的数量为0，而此时又因为c不等于0，说明锁被占用，但是不是写锁，那么此时锁的状态一定是读锁，
     * 既然是读锁状态，那么写锁此时来获取锁时，就肯定失败，因此当w等于0时，tryAcquire()方法返回false。
     * 2. 如果w不为0，说明此时锁的状态时写锁，接着进行current != getExclusiveOwnerThread()判断，判断持有锁的线程是否是当前线程
     * 如果不是当前线程，那么tryAcquire()返回false；如果是当前线程，那么就进行后面的逻辑。为什么是当前线程持有锁，就还能执行后面的逻辑呢？
     * 因为读写锁是支持重入的。
     */
    if (w == 0 || current != getExclusiveOwnerThread())
        return false;
    // 下面一行代码是判断，写锁的重入次数或不会超过最大限制，这个最大限制是：2的16次方减1
    // 为什么是2的16次方减1呢？因为state的低16位存放的是写锁，因此写锁数量的最大值是2的16次方减1
    if (w + exclusiveCount(acquires) > MAX_COUNT)
        throw new Error("Maximum lock count exceeded");
    // Reentrant acquire
    setState(c + acquires);
    return true;
}
/**
 * 1. writerShouldBlock()方法的作用是判断当前线程是否应该阻塞，对于公平的写锁和非公平写锁的具体实现不一样。
 * 对于非公平写锁而言，直接返回false，因为非公平锁获取锁之前不需要去判断是否排队
 * 对于公平锁写锁而言，它会判断同步队列中是否有人在排队，有人排队，就返回true，表示当前线程需要阻塞。无人排队就返回false。
 *
 * 2. 当writerShouldBlock()返回true时，表示当前线程还不能直接获取锁，因此tryAcquire()方法直接返回false。
 * 当writerShouldBlock()返回false时，表示当前线程可以尝试去获取锁，因此会执行if判断中后面的逻辑，即通过CAS方法尝试去修改同步变量的值，
 * 如果修改同步变量成功，则表示当前线程获取到了锁，最终tryAcquire()方法会返回true。如果修改失败，那么tryAcquire()会返回false，表示获取锁失败。
 *
 */
if (writerShouldBlock() ||
    !compareAndSetState(c, c + acquires))
    return false;
setExclusiveOwnerThread(current);
return true;
}

读锁加锁的原理

获取读锁的流程
c == 0 表示锁还没有被任何线程占用
r 读锁的数量
w = exclusiveCount(c) 写锁的数量

  1. 如果c!=0 && w!=0,表示写锁在占用锁，改线程就未获取到读锁所以立即执行fullTryAcquireShared(current);
  2. 如果c!=0 && r!=0,表示锁被写线程占用
protected final int tryAcquireShared(int unused) {
Thread current = Thread.currentThread();
int c = getState();
// exclusiveCount(c)返回的是写锁的数量，如果它不为0，说明写锁被占用，如果此时占用写锁的线程不是当前线程，就返回-1，表示获取锁失败
if (exclusiveCount(c) != 0 &&
    getExclusiveOwnerThread() != current)
    return -1;
// r表示的是读锁的数量
int r = sharedCount(c);
/**
 * 在下面的代码中进行了三个判断：
 * 1、读锁是否应该排队。如果没有人排队，就进行if后面的判断。有人排队，就不会进行if后面的判断，而是最终调用fullTryAcquireShared()方法
 * 2、读锁数量是否超过最大值。（最大数量为2的16次方-1）
 * 3、尝试修改同步变量的值
 */
if (!readerShouldBlock() &&
    r < MAX_COUNT &&
    compareAndSetState(c, c + SHARED_UNIT)) {
    // 读锁数量为0时，就将当前线程设置为firstReader，firstReaderHoldCount=1
    if (r == 0) {
        firstReader = current;
        firstReaderHoldCount = 1;
    } else if (firstReader == current) {
        // 读锁数量不为0且firstReader(第一次获取读的线程)为当前线程，就将firstReaderHoldCount累加
        firstReaderHoldCount++;
    } else {
        // 读锁数量不为0，且第一个获取到读锁的线程不是当前线程
        // 下面这一段逻辑就是保存当前线程获取读锁的次数，如何保存的呢？
        // 通过ThreadLocal来实现的，readHolds就是一个ThreadLocal的实例
        HoldCounter rh = cachedHoldCounter;
        if (rh == null || rh.tid != getThreadId(current))
            cachedHoldCounter = rh = readHolds.get();
        else if (rh.count == 0)
            readHolds.set(rh);
        rh.count++;
    }
    // 返回1表示获取读锁成功
    return 1;
}
// 当if中的三个判断均不满足时，就会执行到这儿，调用fullTryAcquireShared()方法尝试获取锁
return fullTryAcquireShared(current);
}
  1. 如果 r==0, firstReader = current
  2. 如果 r!=0 && firstReader为当前线程，firstReaderHoldCount++
  3. 如果第一个获取到读锁的线程不是当前线程就记录当前线程的获取锁的数量，并让请求线程获得锁

读锁获取锁失败后会循环的去执行下面这个方法，直到满足相应的条件才会 return 退出，否则一直循环

 final int fullTryAcquireShared(Thread current) {
    /*
     * This code is in part redundant with that in
     * tryAcquireShared but is simpler overall by not
     * complicating tryAcquireShared with interactions between
     * retries and lazily reading hold counts.
     */
    HoldCounter rh = null;
    // for死循环，直到满足相应的条件才会return退出，否则一直循环
    for (;;) {
        int c = getState();
        // 锁的状态为写锁时，持有锁的线程不等于当期那线程，就说明当前线程获取锁失败，返回-1
        if (exclusiveCount(c) != 0) {
            if (getExclusiveOwnerThread() != current)
                return -1;
            // else we hold the exclusive lock; blocking here
            // would cause deadlock.
        } else if (readerShouldBlock()) {
            // Make sure we're not acquiring read lock reentrantly
            if (firstReader == current) {
                // assert firstReaderHoldCount > 0;
            } else {
                if (rh == null) {
                    rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current)) {
                        rh = readHolds.get();
                        if (rh.count == 0)
                            readHolds.remove();
                    }
                }
                if (rh.count == 0)
                    return -1;
            }
        }
        if (sharedCount(c) == MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        // 尝试设置同步变量的值，只要设置成功了，就表示当前线程获取到了锁，然后就设置锁的获取次数等相关信息
        if (compareAndSetState(c, c + SHARED_UNIT)) {
            if (sharedCount(c) == 0) {
                firstReader = current;
                firstReaderHoldCount = 1;
            } else if (firstReader == current) {
                firstReaderHoldCount++;
            } else {
                if (rh == null)
                    rh = cachedHoldCounter;
                if (rh == null || rh.tid != getThreadId(current))
                    rh = readHolds.get();
                else if (rh.count == 0)
                    readHolds.set(rh);
                rh.count++;
                cachedHoldCounter = rh; // cache for release
            }
            return 1;
        }
    }
}

问题
当有100个线程来并发的进行读写请求，其中有99个线程是进行读请求，只有一个线程是进行写请求（假设写请求的编号为20）

  1. 先有1-19线程进行了读请求
  2. 然后第20线程进行了写请求
  3. 又来21-100线程80个线程进行读请求
  4. 结果是第20线程等到所有读线程执行完了才能执行写请求
  5. 从而导致写锁饥饿问题

总结
多读单写在实际开发过程中是非常常见的，不同的开发语言有不同的解决方式，但是大体的实现思路是差不多的。我们会使用读写锁，但是其读写锁的原理也需要明白和理解。

欢迎关注【无量测试之道】公众号，回复【领取资源】
Python编程学习资源干货、
Python+Appium框架APP的UI自动化、
Python+Selenium框架Web的UI自动化、
Python+Unittest框架API自动化、

资源和代码 免费送啦~
文章下方有公众号二维码，可直接微信扫一扫关注即可。

备注：我的个人公众号已正式开通，致力于测试技术的分享，包含：大数据测试、功能测试，测试开发，API接口自动化、测试运维、UI自动化测试等，微信搜索公众号：“无量测试之道”，或扫描下方二维码：

 

 添加关注，让我们一起共同成长！

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/Wu13241454771/p/14596461.html