并发编程——ReentrantReadWriteLock

简介: 因为ReentrantLock是互斥锁,如果有一个操作是读多写少,同时还需要保证线程安全,那么使用ReentrantLock会导致效率比较低。因为多个线程在对同一个数据进行读操作时,也不会造成线程安全问题。

@[TOC]

为什么要出现读写锁

因为ReentrantLock是互斥锁,如果有一个操作是读多写少,同时还需要保证线程安全,那么使用ReentrantLock会导致效率比较低。因为多个线程在对同一个数据进行读操作时,也不会造成线程安全问题。所以出现了ReentrantReadWriteLock锁:

  • 读读操作是共享的。
  • 写写操作是互斥的。
  • 读写操作是互斥的。
  • 写读操作是互斥的。

单个线程获取写锁后,再次获取读锁,可以拿到。(写读可重入)

单个线程获取读锁后,再次获取写锁,拿不到。(读写不可重入)

使用方式:

public class XxxTest {
   
    // 读写锁
    static ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

    // 写锁
    static ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();

    // 读锁
    static ReentrantReadWriteLock.ReadLock readLock = lock.readLock();

    public static void main(String[] args) throws InterruptedException {
   
        readLock.lock();
        try {
   
            System.out.println("拿到读锁!");
        } finally {
   
            readLock.unlock();
        }

        writeLock.lock();
        try {
   
            System.out.println("拿到写锁!");
        } finally {
   
            writeLock.unlock();
        }
    }
}

读写锁的核心思想

ReentrantReadWriteLock基于AQS实现的,很多功能的实现和ReentrantLock类似。还是基于AQS的state来确定当前线程是否拿到锁资源。将state的高16位作为读锁的标识,将state的低16位作为写锁的标识

锁重入问题:

  • 写锁重入:因为写操作和其他操作是互斥的,代表同一时间,只有一个线程持有着写锁,只要锁重入,就对低位+1即可。
  • 读锁重入:读锁的重入不能仿照写锁的方式,因为写锁属于互斥锁,同一时间只会有一个线程持有写锁,但是读锁是共享锁,同一时间会有多个线程持有读锁。所以每个获取到读锁的线程,记录锁重入的方式都是基于自己的ThreadLocal存储锁重入次数。

读锁重入修改state,只是记录当前线程锁重入的次数,需要基于ThreadLocal记录。

state二进制表示:00000000 00000000 00000000 00000000

将state的高16位作为读锁的标识,将state的低16位作为写锁的标识。

写锁:00000000 00000000 00000000 00000001

写锁重入,低16位+1:00000000 00000000 00000000 00000010

读锁:

00000000 00000001 00000000 00000000

读锁重入,高16位+1:

00000000 00000010 00000000 00000000

每个读操作的线程,在获取读锁时,都需要开辟一个ThreadLocal。读写锁为了优化这个事情,做了两手操作:

  • 第一个拿到读锁的线程,不用ThreadLocal记录重入次数,在读写锁内有有一个firstRead记录重入次数
  • 记录了最后一个拿到读锁的线程的重入次数,交给cachedHoldCounter属性标识,可以避免频繁的在锁重入时,从TL中获取。

写锁的操作

写锁加锁-acquire

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

tryAcquire:尝试获取锁资源,能否以CAS的方式将state 从0 改为 1,改成功,拿锁成功。
addWaiter:将当前没按到锁资源的,封装成Node,排到AQS里。
acquireQueued:当前排队的能否竞争锁资源,不能挂起线程阻塞。

因为都是AQS的实现,主要看tryAcquire

// state,高16:读,低16:写
00000000 00000000 00000000 00000000

00000000 00000001 00000000 00000000 - SHARED_UNIT

00000000 00000000 11111111 11111111 - MAX_COUNT

00000000 00000000 11111111 11111111 - EXCLUSIVE_MASK
&
00000000 00000000 00000000 00000001 

static 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;

// 只拿到表示读锁的高16位。
static int sharedCount(int c)    {
    return c >>> SHARED_SHIFT; }
// 只拿到表示写锁的低16位。
static int exclusiveCount(int c) {
    return c & EXCLUSIVE_MASK; }


// 读写锁的写锁,获取流程
protected final boolean tryAcquire(int acquires) {
   
    // 拿到当前线程
    Thread current = Thread.currentThread();
    // 拿到state
    int c = getState();
    // 拿到了写锁的低16位标识w
    int w = exclusiveCount(c);
    // c != 0:要么有读操作拿着锁,要么有写操作拿着锁
    if (c != 0) {
   
        // 如果w == 0,代表没有写锁,拿不到
        // 如果w != 0,代表有写锁,看一下拿占用写锁是不是当前线程,如果不是,拿不到
        if (w == 0 || current != getExclusiveOwnerThread())
            return false;
        // 到这,说明肯定是写锁,并且是当前线程持有
        // 判断对低位 + 1,是否会超过MAX_COUNT,超过抛Error
        if (w + exclusiveCount(acquires) > MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        // 如果没超过锁重入次数, + 1,返回true,拿到锁资源。
        setState(c + acquires);
        return true;
    }
    // 到这,说明c == 0
    // 读写锁也分为公平锁和非公平锁
    // 公平:看下排队不,排队就不抢了
    // 走hasQueuedPredecessors方法,有排队的返回true,没排队的返回false
    // 非公平:直接抢!
    // 方法实现直接返回false
    if (writerShouldBlock() ||
        // 以CAS的方式,将state从0修改为 1
        !compareAndSetState(c, c + acquires))
        // 要么不让抢,要么CAS操作失败,返回false
        return false;
    // 将当前持有互斥锁的线程,设置为自己
    setExclusiveOwnerThread(current);
    return true;
}

addWaiter和acquireQueued和ReentrantLock看的一样,都是AQS自身提供的方法。

写锁-释放锁操作

读写锁的释放操作,跟ReentrantLock一致,只是需要单独获取低16位,判断是否为0,为0就释放成功

// 写锁的释放锁
public final boolean release(int arg) {
   
    // 只有tryRealse是读写锁重新实现的方法,其他的和ReentrantLock一致
    if (tryRelease(arg)) {
   
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

// 读写锁的真正释放
protected final boolean tryRelease(int releases) {
   
    // 判断释放锁的线程是不是持有锁的线程
    if (!isHeldExclusively())
        // 不是抛异常
        throw new IllegalMonitorStateException();
    // 对state - 1
    int nextc = getState() - releases;
    // 拿着next从获取低16位的值,判断是否为0
    boolean free = exclusiveCount(nextc) == 0;
    // 返回true
    if (free)
        // 将持有互斥锁的线程信息置位null
        setExclusiveOwnerThread(null);
    // 将-1之后的nextc复制给state
    setState(nextc);
    return free;
}

读锁的操作

读锁的加锁操作

// 读锁加锁操作
public final void acquireShared(int arg) {
   

    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

tryAcquireShared,尝试获取锁资源,获取到返回1,没获取到返回-1
doAcquireShared,前面没拿到锁,这边需要排队

// tryAcquireShared方法
protected final int tryAcquireShared(int unused) {
   
    // 获取当前线程
    Thread current = Thread.currentThread();
    // 拿到state
    int c = getState();
    // 那写锁标识,如果 !=0,代表有写锁
    if (exclusiveCount(c) != 0 &&
        // 如果持有写锁的不是当前线程,排队去!
        getExclusiveOwnerThread() != current)
        // 排队!
        return -1;
    // 没有写锁!
    // 获取读锁信息
    int r = sharedCount(c);
    // 公平锁: 有人排队,返回true,直接拜拜,没人排队,返回false
    // 非公平锁:正常的逻辑是非公平直接抢,因为是读锁,每次抢占只要CAS成功,必然成功
    // 这就会出现问题,写操作无法在读锁的情况抢占资源,导致写线程饥饿,一直阻塞
    // 非公平锁会查看next是否是写锁的,如果是,返回true,如果不是返回false
    if (!readerShouldBlock() &&
        // 查看读锁是否已经达到了最大限制
        r < MAX_COUNT &&
        // 以CAS的方式,对state的高16位+1
        compareAndSetState(c, c + SHARED_UNIT)) {
   
        // 拿到锁资源成功!!!
        if (r == 0) {
   
            // 第一个拿到锁资源的线程,用first存储
            firstReader = current;
            firstReaderHoldCount = 1;
        } else if (firstReader == current) {
   
            // 锁重入,第一个拿到读锁的线程,直接对firstReaderHoldCount++记录重入的次数
            firstReaderHoldCount++;
        } else {
   
            // 不是第一个拿到锁资源的
            // 先拿到cachedHoldCounter,最后一个线程的重入次数
            HoldCounter rh = cachedHoldCounter;
            // rh == null: 第二个拿到读锁的!
            // 或者发现之前有最后一个来的,但是不我,将我设置为最后一个。
            if (rh == null || rh.tid != getThreadId(current))
                // 获取自己的重入次数,并赋值给cachedHoldCounter
                cachedHoldCounter = rh = readHolds.get();
            // 之前拿过,现在如果为0,赋值给TL
            else if (rh.count == 0)
                readHolds.set(rh);
            // 重入次数+1,
            // 第一个:可能是第一次拿
            // 第二个:可能是重入操作
            rh.count++;
        }
        return 1;
    }
    return fullTryAcquireShared(current);
}

// 通过tryAcquireShared没拿到锁资源,也没返回-1,就走这
final int fullTryAcquireShared(Thread current) {
   
    HoldCounter rh = null;
    for (;;) {
   
        // 拿state
        int c = getState();
        // 现在有互斥锁,不是自己,拜拜!
        if (exclusiveCount(c) != 0) {
   
            if (getExclusiveOwnerThread() != current)
                return -1;

        // 公平:有排队的,进入逻辑。   没排队的,过!
        // 非公平:head的next是写不,是,进入逻辑。   如果不是,过!
        } else if (readerShouldBlock()) {
   
            // 这里代码特别乱,因为这里的代码为了处理JDK1.5的内存泄漏问题,修改过~
            // 这个逻辑里不会让你拿到锁,做被阻塞前的准备
            if (firstReader == current) {
   
                // 什么都不做
            } else {
   
                if (rh == null) {
   
                    // 获取最后一个拿到读锁资源的
                    rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current)) {
   
                        // 拿到我自己的记录重入次数的。
                        rh = readHolds.get();
                        // 如果我的次数是0,绝对不是重入操作!
                        if (rh.count == 0)
                            // 将我的TL中的值移除掉,不移除会造成内存泄漏
                            readHolds.remove();
                    }
                }
                // 如果我的次数是0,绝对不是重入操作!
                if (rh.count == 0)
                    // 返回-1,等待阻塞吧!
                    return -1;
            }
        }
        // 超过读锁的最大值了没?
        if (sharedCount(c) == MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        // 到这,就CAS竞争锁资源
        if (compareAndSetState(c, c + SHARED_UNIT)) {
   
            // 跟tryAcquireShared一模一样
            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; 
            }
            return 1;
        }
    }
}

加锁-扔到队列准备阻塞操作

// 没拿到锁,准备挂起
private void doAcquireShared(int arg) {
   
    // 将当前线程封装为Node,当前Node为共享锁,并添加到队列的模式
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
   
        boolean interrupted = false;
        for (;;) {
   
            // 获取上一个节点
            final Node p = node.predecessor();
            if (p == head) {
   
                // 如果我的上一个是head,尝试再次获取锁资源
                int r = tryAcquireShared(arg);
                if (r >= 0) {
   
                    // 如果r大于等于0,代表获取锁资源成功
                    // 唤醒AQS中我后面的要获取读锁的线程(SHARED模式的Node)
                    setHeadAndPropagate(node, r);
                    p.next = null; 
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            // 能否挂起当前线程,需要保证我前面Node的状态为-1,才能执行后面操作
            if (shouldParkAfterFailedAcquire(p, node) &&
                //LockSupport.park挂起~~
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
   
        if (failed)
            cancelAcquire(node);
    }
}

总结

ReentrantReadWriteLock的使用可以提高并发性,特别适用于读操作远多于写操作的情况。它相对于排他锁,因为排他锁在同一时间只允许一个线程访问。ReentrantReadWriteLock允许多个读线程同时访问,但不允许写线程和读线程同时访问,也不允许写线程和写线程同时访问。

相关文章
|
2月前
|
XML Java 数据安全/隐私保护
彻底搞懂 Spring Boot 自动配置原理:从源码拆解到手写 Starter,零废话全干货
本文深入解析SpringBoot自动配置原理,基于SpringBoot 3.4.2版本详细拆解了自动配置的执行流程。主要内容包括:1)自动配置的本质是基于条件注解的动态JavaConfig配置类;2)核心执行流程通过AutoConfigurationImportSelector实现;3)SpringBoot 3.x采用新的自动配置注册方式;4)重点讲解了@Conditional系列条件注解的使用场景与常见坑点;5)通过开发自定义加密Starter实战演示完整实现过程。
1085 3
|
2月前
|
人工智能 数据挖掘 Python
Python 3.9.5 安装教程:详细步骤+新手入门介绍
本教程详解Python 3.9.5安装全流程:下载解压、管理员运行、自定义安装、路径设置、创建IDLE快捷方式,助你快速完成环境搭建,轻松开启编程学习之旅。(238字)
|
1月前
|
JSON 搜索推荐 定位技术
无 Embedding、无向量数据库的 RAG 方法:PageIndex 技术解析
PageIndex是无向量、基于推理的RAG框架,通过构建文档层次化目录树,让大模型像人类专家一样逐层推理导航,精准定位答案,支持可追溯、高相关性检索,专长于财报、法律、政策等结构化长文档。
357 0
无 Embedding、无向量数据库的 RAG 方法:PageIndex 技术解析
|
2月前
|
Linux 网络安全 API
OpenClaw 阿里云+本地系统部署保姆级图文步骤+千问/Coding Plan配置+Skill体系与实战场景分享
OpenClaw(内部代号Clawdbot)作为2026年主流的本地AI助手平台,核心价值在于通过**技能体系**实现任务自动化、流程串联与跨场景协作。很多用户虽然完成基础部署,但因技能选择混乱、模型配置错误、运行环境不稳定,导致效率不升反降。本文基于稳定实践,完整覆盖**阿里云轻量服务器部署、本地macOS/Linux/Windows11部署、阿里云千问大模型接入、免费Coding Plan模型配置、官方推荐10类核心技能、5类高频实战场景**,并附带可直接复制的命令与常见问题解决方案,实现从环境搭建到高效落地的全流程闭环。
329 8
|
5月前
|
负载均衡 Java Nacos
Gateway服务网关
网关是微服务的统一入口,实现请求路由、权限控制、限流与负载均衡。基于SpringCloud Gateway可构建高性能响应式网关,支持断言与过滤器工厂,灵活实现路由匹配及请求处理,并可通过全局过滤器扩展功能,解决跨域等问题。
|
数据库连接 数据库
多线程事务失效的原因
【5月更文挑战第16天】多线程事务失效的原因
837 0
|
XML Java 数据格式
Spring容器的本质
本文主要讨论Spring容器最核心的机制,用最少的代码讲清楚Spring容器的本质。
|
缓存 监控 负载均衡
将近2万字的Dubbo原理解析,彻底搞懂dubbo
市面上有很多基于RPC思想实现的框架,比如有Dubbo。今天就从Dubbo的SPI机制、服务注册与发现源码及网络通信过程去深入剖析下Dubbo。
30686 9
|
存储 安全 Java
最爱问的高频ConcurrentHashMap原理,你会了吗?
ConcurrentHashMap 是 Java 中的线程安全散列表实现,允许多个线程同时访问和修改数据。它在 JDK 1.7 中通过分段锁机制将 HashMap 分为多个段,每个段使用独立的锁来保证线程安全;而在 JDK 1.8 中则采用 CAS 和 synchronized 结合的方式,提高了并发性能。与 HashMap 相比,ConcurrentHashMap 是线程安全的,支持更高的并发性能,且不支持 null 键和值。CAS(Compare-and-Swap)是一种无锁原子操作,用于确保多线程环境下的数据一致性,避免竞态条件。
822 5
|
开发框架 Dubbo 应用服务中间件
微服务开发框架-----Apache Dubbo
这篇文章介绍了Apache Dubbo微服务开发框架,它提供RPC通信和微服务治理能力,支持服务发现、负载均衡和流量治理等功能,并强调了Dubbo在微服务规模化实践和企业级治理方面的优势。
微服务开发框架-----Apache Dubbo