一、高性能基础库 base ├── 1无锁哈希表 AtomicHashMap ├── 2原子读写锁 AtomicRWLock ← 本文 ├── 3读写锁守卫 RWLockGuard ← 本文 ├── 4可重入读写锁 ReentrantRWLock ├── 5有界队列 BoundedQueue ├── 6等待策略 WaitStrategy ├── 7无界队列 UnboundedQueue ├── 8线程安全队列 ThreadSafeQueue ├── 9线程池 ThreadPool ├── 10对象池 ObjectPool ├── 11并发对象池 CCObjectPool ├── 12信号槽 Signal ├── 13区间遍历 FOR_EACH └── 14基础宏 macros源文件cyber/base/atomic_rw_lock.h、cyber/base/rw_lock_guard.h上一篇的哈希表自己改指针表里面没有锁。Cyber 里还有调度器的协程表、拓扑图、Clock普通容器读的人多、改的人少。一把mutex会让读者也排队。下面这把锁允许好几个线程同时读写的时候必须一个人独占。先看最小用法。加锁、解锁你都不用手写花括号结束就会解开usingapollo::cyber::base::AtomicRWLock;usingapollo::cyber::base::ReadLockGuard;usingapollo::cyber::base::WriteLockGuard;AtomicRWLock lock;// 默认写优先{ReadLockGuardAtomicRWLockg(lock);// 进来加读锁// 读共享数据。别的线程也可以同时拿读锁}// 出去自动解读锁{WriteLockGuardAtomicRWLockg(lock);// 进来加写锁// 改共享数据。这时别人读、写都进不来}// 出去自动解写锁ReadLock()是私有的所以你不能写lock.ReadLock()。为什么要私有、守卫里那几行在干什么放到第五节。先搞清楚锁里那两个数字。一两个数字lock_num_和排队人数锁的全部状态就是一个原子整数lock_num_再加一个「有多少人正在抢写锁」的计数。常量只是起名RW_LOCK_FREE 0WRITE_EXCLUSIVE -1。lock_num_名字含义别人可以做什么0空闲没人持锁读者可以变成1写者可以变成-11, 2, 3…读者人数几个线程正在读读者可以再1写者必须等到回到0-1写者独占恰好一个线程在写读者、写者都要等另外还有write_lock_wait_num_已经执行了 WriteLock、但还没把lock_num_改成-1的人数。它不是「已经持有写锁」只是报名。默认write_first_ true这个数大于 0新读者就不许再1。对应到源码成员// 0表示没人占用锁staticconstint32_tRW_LOCK_FREE0;// -1表示有人占用写锁staticconstint32_tWRITE_EXCLUSIVE-1;// 最大重试次数staticconstuint32_tMAX_RETRY_TIMES5;// ...// 等待写锁的线程数std::atomicuint32_twrite_lock_wait_num_{0};// 锁的占用情况std::atomicint32_tlock_num_{0};// 是否优先写锁boolwrite_first_true;数字看懂之后再用门口来记迁移先有表再有这句话0是空屋子正数是屋里几个读者-1是一个写者把门锁上。写者进门前先在门口挂个号write_lock_wait_num_就是挂了几个号。这一节的问题正数为什么表示读者、要用-1表示写者因为一个整数要同时表达「几个读者」和「有没有写者」。读者用加一减一写者单独占一个负数避免和「一个读者」的1混在一起。这把锁给谁用调度器、拓扑仓库、Clock。哈希表自己 CAS不用它。二场景一读者进门默认写优先时读锁是两层循环内层等到「现在能读」外层用 CAS 把人数加一。inlinevoidAtomicRWLock::ReadLock(){uint32_tretry_times0;int32_tlock_numlock_num_.load();// 判断当前是否可读if(write_first_){do{// 如果锁的占用情况小于0或者等待写锁的线程数大于0则等待while(lock_numRW_LOCK_FREE||write_lock_wait_num_.load()0){if(retry_timesMAX_RETRY_TIMES){// 告诉操作系统让出CPU时间片std::this_thread::yield();retry_times0;}lock_numlock_num_.load();}// 代表没人占用写锁可以占用读锁}while(!lock_num_.compare_exchange_weak(lock_num,lock_num1,std::memory_order_acq_rel,std::memory_order_relaxed));}else{// 读优先do{while(lock_numRW_LOCK_FREE){if(retry_timesMAX_RETRY_TIMES){// saving cpustd::this_thread::yield();retry_times0;}lock_numlock_num_.load();}}while(!lock_num_.compare_exchange_weak(lock_num,lock_num1,std::memory_order_acq_rel,std::memory_order_relaxed));}}yield()当前线程对操作系统说「先跑别人」。不是把锁解开也不是睡固定时间。读优先AtomicRWLock lock(false)内层只判断lock_num 0不管有没有人报名写。默认构造不走那条。带数值走一遍没有写者假设现在lock_num_ 2两个读者write_lock_wait_num_ 0。线程 A 来拿读锁。lock_num读到2。内层条件2 0为假wait 0为假不进内层 while。CAS期望仍是2想改成3。没人抢成功。while (!true)为假函数返回。现在lock_num_ 3三个读者。CAS 可以先记成一句话原子量还等于我手里的期望值才改成新值。这里期望是局部变量lock_num2新值是3。带数值走一遍有人在抢写锁同一时刻线程 B 已经进了WriteLock报名成功write_lock_wait_num_ 1但屋里还有读者B 还没把数改成-1。线程 A 再来读。lock_num可能仍是2。内层wait 0为真卡在这一行的 while 里反复load满 5 次yield。A 不会去执行后面的 CAS所以不会出现「写者在排队读者人数还在涨」。等 B 抢到-1再释放wait回到 0 且lock_num_回到 0 之后A 才能 CAS0 → 1。这就是写优先写者还没拿到锁只报了名新读者已经被挡住。否则读者可以一直1写者永远看不到0。外层do-while还防一种情况内层刚跳出别人抢先把数改了。CAS 失败则!false为真再进内层看门。这一节的问题真正上读锁的是哪一句最后的 CAS。前面的 while 只是等。线程 A 在写者报名时卡在哪while (lock_num 0 || wait 0)。三场景二写者进门报名是关键inlinevoidAtomicRWLock::WriteLock(){int32_trw_lock_freeRW_LOCK_FREE;uint32_tretry_times0;// 等待写锁的线程数加1write_lock_wait_num_.fetch_add(1);// 尝试获取写锁while(!lock_num_.compare_exchange_weak(rw_lock_free,WRITE_EXCLUSIVE,std::memory_order_acq_rel,std::memory_order_relaxed)){// rw_lock_free will change after CAS fail, so init aginrw_lock_freeRW_LOCK_FREE;if(retry_timesMAX_RETRY_TIMES){// saving cpustd::this_thread::yield();retry_times0;}}// 等待写锁的线程数减1write_lock_wait_num_.fetch_sub(1);}compare_exchange_weak(期望, 新值)成功lock_num_从0变成-1返回true。前面有!while 结束。失败原子量不变但会把你传入的rw_lock_free改成当前真实值返回false。最后这一点对初学者最容易漏失败改的是你手里那个局部变量不是「把锁改坏」。所以循环里必须再赋成0下次还是「我期望空闲」。带数值走一遍开始lock_num_ 2wait 0。线程 B 要写。wait变成1。之后新读者都会卡在上一节的内层 while。CAS期望0实际是2失败。rw_lock_free被改成2。若不写回0下一轮会拿期望2去比可能误把「两个读者」换成-1把读锁拆掉。所以要rw_lock_free 0。两个读者陆续离开2 → 1 → 0。B 再 CAS0 → -1成功。wait减回0。B 独占。报名为什么必须在 CAS 之前若先抢0 → -1、抢不到再报名中间会有空档新读者可以继续1写者可能一直饿着。先报名写优先才成立。weak和哈希表里的strong都是 CAS。weak允许「值其实没变也失败一次」这把锁外层本来就要重试用weak即可换成strong也对。假失败时同样会改期望值同样要写回0。这一节的问题为什么还没拿到锁报名就能挡读者因为读者看的是wait 0不是「已经是-1」。while (!cas)何时为真CAS 失败屋里不空或 weak 空失败就继续转成功才停。四场景三解锁inlinevoidAtomicRWLock::ReadUnlock(){lock_num_.fetch_sub(1);}inlinevoidAtomicRWLock::WriteUnlock(){lock_num_.fetch_add(1);}fetch_sub/fetch_add自己是原子的不用再 CAS。当前谁走之后为什么3一个读者2人数减一1最后一个读者0屋子空了写者可以进-1写者0-1 1 0不是再减一写者若也减一会变成-2读者会把「小于 0」一直当成有人在写锁就坏了。这四处才会改lock_num_读 CAS 成功、写 CAS 成功、读解锁、写解锁。空转里的load只看不改。同一线程再拿一次写锁lock_num_已经是-1自己再 CAS0 → -1永远失败会空转等自己。这把锁不能重入那是4ReentrantRWLock的事。五守卫为什么例子里要写ReadLockGuard上面四句加解锁都是private。类开头写了友元只有守卫能调friendclassReadLockGuardAtomicRWLock;friendclassWriteLockGuardAtomicRWLock;所以不是「建议用守卫」是只能用守卫。这也逼你把「加」和「解」绑在同一个栈对象上避免ReadLock()之后return把ReadUnlock()跳过。RAII资源跟对象的寿命走。对象造出来就拿资源对象销毁就还资源。{ReadLockGuardAtomicRWLockg(lock);// 构造 → ReadLock()// ...}// 析构 → ReadUnlock()即使中间 return 也会走这里源码templatetypenameRWLockclassReadLockGuard{public:explicitReadLockGuard(RWLocklock):rw_lock_(lock){rw_lock_.ReadLock();}~ReadLockGuard(){rw_lock_.ReadUnlock();}private:ReadLockGuard(constReadLockGuard)delete;ReadLockGuardoperator(constReadLockGuard)delete;RWLockrw_lock_;};