★ FEATURED ARTICLE 一、高性能基础库 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_hash_map.h先记住一件事Cyber 里到处在查「这个 ID 对应谁」。例如 Channel 号 → 缓冲区、节点号 → 名字。这些 ID 是整数查得很勤还经常好几个线程同时查、同时改。std::unordered_map 一把大锁也能用但锁一抢大家都排队。AtomicHashMap的做法是表很小、不扩容、不删除用原子操作硬改指针。换来的是热路径上少排队。先把几个最容易被问到的结论写死后文展开在 CyberRT 里干什么给整数 IDnode / channel / service / task做并发查找例如GlobalData、DataDispatcher。支不支持并发读写支持Set/Get/Has并发没有 Delete。无锁怎么保证线程安全不用 mutex用std::atomic CAScompare_exchange_strong失败整轮重试。扩容和冲突桶数编译期固定默认 128须为 2 的幂不扩容冲突用桶内有序拉链链地址法不是开放定址。怎么用apollo::cyber::base::AtomicHashMapint,std::stringmap;map.Set(5,std::string(hello));// 插入或覆盖std::string v;if(map.Has(5)map.Get(5,v)){// v hello}一这张表长什么样固定 128 个桶2 的幂用key 127定位。桶里是升序链表哑头head_ 若干Entry。value 不放在节点里只挂指针方便 CAS 换值。图Set(5,hello)落到table_[5]。虚线框是哑头head_实线框是有序Entry绿色是堆上的 value。改值只 CAS 换指针。二实现方法定位、CAS、重试无锁不是「不用同步」而是不用互斥锁。同步收到 CAS 这一条不可打断的指令上。失败就整轮重来系统整体还能推进。对外入口很短算桶再交给桶上的Insert。voidSet(K key,constVvalue){uint64_tindexkeymode_num_;// 5 127 → 5table_[index].Insert(key,value);}Insert先Find再 CAS。Find在升序链上走带回插入/更新要用的前后指针。链为哑头 → 3 → 7 → 10时找谁返回prevtarget接下来7true37CAS 换value_ptr5false37CAS 插到 3 和 7 中间12false10空CAS 接到 10 后面命中则换值new一份VCAS 把value_ptr从旧指针换成新的成功就delete旧对象。未命中则先new_entry-next target再 CASprev-next从target换成new_entry。必须先接好后继再挂到前驱否则会断链。CAS 失败不要死磕这一行continue回到最外层while (true)重新Find。同一 key 并发写最后一次成功的留下。三读这段源码要用的 C 知识点结构、步骤看完之后剩下的卡点几乎都是语言知识列出用到的以下四个知识点供大家参考。1左值引用、右值引用先分清两件事表达式是左值还是右值以及参数写成什么引用。左值右值直观有名字、能取地址如变量s临时量、std::move(s)引用写法T/const TT这份代码Entry(K, const V)拷贝Entry(K, V)尽量移动容易混的点const int是左值引用int是右值引用。int a b合法当且仅当b是可修改的int左值int a 10不合法。const int可以绑临时量非 const 的int不行。函数参数就算写成V value有了名字之后函数里的value是左值。所以要std::forwardV(value)或std::move才能继续当右值用。std::forward按原来的值类别转发出去。std::move是无条件变成右值。这条V重载里两者效果接近const V那条不该写成forwardV拷贝和移动是两条路不是繁简关系。explicit Entry(K key)禁止Entry e some_key这种隐式转换只能显式构造。默认构造new V()不是拷贝只是造一个空的/零的V。对应Set(key)。2compare_exchange_strong签名可以记成boolcompare_exchange_strong(Texpected,T desired,memory_order success,memory_order failure);语义若当前值 expected改成desired并返回 true否则不改把expected写成最新值返回 false。比较和赋值中间插不进别人的写。这份代码里两处 CAS更新value_ptrold_val_ptr→new_value插入prev-nexttarget→new_entry有人抢先改了怎么办返回 falsecontinue回到while (true)顶部不是回到 CAS 那一行。插入失败即使没写continue也会掉出else再进下一轮循环。失败时expected会被覆盖——这是接口约定。这里失败后整轮Find不依赖那个旧变量。3memory_order这一组原子性来自std::atomic的load/store/ CAS。memory_order_*只规定可见顺序。关键字用在这份代码里的意思releasestore对象构造完再公布指针acquireload拿到指针后能看见发布前的写入acq_relCAS 成功又读又发既看见别人也让别人看见自己relaxedCAS 失败没改出去只要原子性acquire不是「加载完再发布」它是读端。发布是release。memory_order_acquire也不会让一次普通赋值变成原子操作。写成int* p q;再怎么写 memory_order 都没用。成对记忆store(ptr, release)↔load(acquire)。CAS 成功acq_rel、失败relaxed是无锁代码的常见写法。4std::atomicstd::atomicV*value_ptr{nullptr};std::atomicEntry*next{nullptr}; {nullptr}是默认空指针。带 key 的构造函数里再store(new V(...), release)。为什么 value 不直接做成V而做成原子指针换值时只 CAS 指针不用拆节点。next同理插入时 CAS 一条边。atomic管的是指针这个字本身的读写对象有没有构造完要靠上面的release/acquire。四核心函数Has、Find、Insert桶里就是一条按 key 升序的链表。Has/Find是查找Insert是「找到了改值没找到插节点」。无锁只多了两件事指针用atomic读写改链用 CAS失败就整轮重找。Insert只看右值版V。另外两个算法一样一个拷贝、一个默认构造。Has// 判断链表中是否存在该节点boolHas(K key){// acquire读到指针时节点已经构造完Entry*m_targethead_-next.load(std::memory_order_acquire);while(Entry*targetm_target){// 空指针则到链尾退出if(target-keykey){m_targettarget-next.load(std::memory_order_acquire);continue;// 有序链还没走到}else{returntarget-keykey;// 相等有更大后面不可能再有}}returnfalse;}有序链上走比 key 小就继续否则看等不等于。更大就可以停后面不会再变小。Find// 找到 key 应插入位置的前驱 prev 和当前 targetboolFind(K key,Entry**prev_ptr,Entry**target_ptr){Entry*prevhead_;// 哑头保证永远有前驱Entry*m_targethead_-next.load(std::memory_order_acquire);while(Entry*targetm_target){// 走到空则链尾if(target-keykey){*prev_ptrprev;*target_ptrtarget;returntrue;// 命中后面改 value_ptr}elseif(target-keykey){*prev_ptrprev;*target_ptrtarget;returnfalse;// 应插在 prev 和 target 之间}else{prevtarget;m_targettarget-next.load(std::memory_order_acquire);}}*prev_ptrprev;*target_ptrnullptr;// 接到尾巴returnfalse;}还是有序查找多带回prev/target给插入当挂钩。哑头head_保证永远有前驱。相等找到了更大应插在prev和target之间走到空接到尾巴Insert(K, V)// 插入右值少一次拷贝voidInsert(K key,Vvalue){Entry*prevnullptr;Entry*targetnullptr;Entry*new_entrynullptr;V*new_valuenullptr;while(true){// CAS 失败则整轮重找if(Find(key,prev,target)){// key 已存在只换 value 指针if(!new_value){new_valuenewV(std::forwardV(value));// 移动避免拷贝}autoold_val_ptrtarget-value_ptr.load(std::memory_order_acquire);// CAS若仍是 old则换成 new成功 acq_rel 发布失败 relaxedif(target-value_ptr.compare_exchange_strong(old_val_ptr,// expected期望仍是刚才读到的旧指针new_value,// desired要换成的新 V*std::memory_order_acq_rel,// 成功既看见别人也让别人看见自己std::memory_order_relaxed)){// 失败没改出去只要原子性deleteold_val_ptr;// CAS 成功才删旧值if(new_entry){deletenew_entry;new_entrynullptr;}return;}continue;// 有人抢先改了回到 while 重新 Find}else{// key 不存在链表插入if(!new_entry){new_entrynewEntry(key,value);}new_entry-next.store(target,std::memory_order_release);// 先接好后继// CASprev-next 仍是 target 才挂上 new_entryif(prev-next.compare_exchange_strong(target,// expected挂钩还没被别人改过new_entry,// desired新节点std::memory_order_acq_rel,std::memory_order_relaxed)){// 插入成功prev → new_entry → targetif(new_value){deletenew_value;new_valuenullptr;}return;}// 有人抢先插了下一轮重新 Find}}}找到了CAS 换value_ptr等于改链表节点上的值。没找到先让新节点next指向target再 CAS 把prev-next改成新节点——普通链表插入只是用 CAS 抢这一步。forward是为了移动而不是拷贝。new_value/new_entry只造一次失败留着下一轮用。CAS 失败就回到while (true)重新Find因为链可能已经变了。常见问题AtomicHashMap 在 CyberRT 里有什么作用给热路径上的「整型 ID → 对象」做查找节点名、Channel、回调表。它是cyber/base里的容器不是业务模块。原理是什么怎么实现无锁固定桶数组 桶内有序链表。改值 CASvalue_ptr插节点 CASprev-next。失败就while (true)里重新Find。相比普通哈希表好在哪不是单线程一定更快而是少一把全局锁。多线程同时Set/Get时不必互相堵住。代价是不扩容、不删除、key 必须是整数。base 库里它如何设计源码细节有哪些AtomicHashMap只算桶下标Bucket管链Entry存key、原子value_ptr、原子next。对外Has/Get/Set对内FindInsert。详见上文一四。高性能基础库里的无锁哈希表怎么用见文首示例Set插入或覆盖Has查询Get取值。Get(key, V**)返回内部指针不要delete也不要拿太久。支持并发读写吗线程安全怎么保证Set/Get/Has可并发。安全靠原子指针和 CAS不是读写锁。同一 key 同时写最后一次 CAS 成功的值留下。扩容和冲突处理是怎么做的不扩容。冲突走链地址法同一下标进同一条升序链表。链太长只会变慢不会自动 rehash。这是博主第一次发中间件类笔记用来记下自己学中间件的过程。Cyber RT 虽已发布多年把热路径容器做成固定大小、无锁、按整数 ID 查找这种取舍今天读起来仍值得学。
阅读完成 · 觉得有帮助？
