简介:面向需要深入理解C++内存管理的中高级开发者,这份代码包给出一个两级内存池的可运行实现,重点处理多线程环境下频繁申请与释放小块内存时的系统调用开销、碎片与锁竞争问题。设计上采用全局一级内存池配合每线程独立的二级内存池:小块优先在线程本地分配,不足时再向一级池申请;释放后的块回收到池中复用,而非直接交还操作系统。工程共33个文件、约19KB,以17个头文件和9个cpp源文件为主体,另有sln/vcxproj工程文件、txt说明、bat构建脚本及lib静态库,便于直接编译、调试和嵌入测试工程。已有1269人学习。读者可据此梳理池初始化、分配、释放、并发控制与空闲块组织方式,并针对每块指针大小的额外开销、池耗尽后的动态扩展、碎片整理和细粒度锁等方向继续改进,适合作为内存池课程设计、面试准备和性能优化实践的参考样例。
1. 一个C++内存池实现:从 new/delete 的抖动到可复现的固定块分配器
某在线服务压测时,QPS 卡在一个平台期,火焰图里最宽的不是业务算法,而是 new/delete 背后的分配器锁和页故障。C++内存池实现要解决的不是“比 malloc 快一点”这么简单,而是把高频小对象的分配从通用堆里拆出来,用固定块、分级 size class 和线程缓存换可预测的延迟。它适合连接对象、任务节点、协议包这类生命周期短、尺寸集中的场景;不适合大块、长生命周期或尺寸完全随机的分配。下面按一个能落地的内存池实现推进:先定边界,再写 free list,接上 placement new,最后处理并发、回收和压测验收。
2. 先定边界:固定块、分级块与对象池该怎么选
内存池最容易翻车的地方不是链表写错,而是边界没定清楚:拿一个固定块池去接所有尺寸的分配,最后内部碎片和扩容逻辑会把收益吃光。先分清三种形态——固定块池、分级池、对象池——再决定代码长什么样,比上来就抄一个无锁栈靠谱得多。
2.1 内存池的三种边界:固定块池、分级池、对象池
固定块池只服务一种块大小,allocate就是从空闲链表弹一个节点,deallocate就是压回去,时间复杂度 O(1),代码短、缓存友好。代价是尺寸绑定:业务里有 24 字节的任务节点和 96 字节的协议包,就得建两个池,或者在上层做映射。它最适合“对象尺寸固定、分配频率高、生命周期短”的热路径,比如定时器节点、消息队列节点、短连接上下文。
分级池是一组固定块池,按 16、32、48、64 这样的 size class 分桶。分配时把请求大小向上取整到某个 class,释放时再找回 class 索引。它比固定块池通用,但多了两个成本:内部碎片,以及“释放时怎么知道这块属于哪个 class”的元数据。常见做法是把 class 索引放在块头,或者用页对齐加页表反查;前者实现简单但会吃掉 8 到 16 字节,后者对对齐和平台页大小更敏感。
对象池则是在原始内存池上再包一层构造和析构。它绑定类型T,create里做 placement new,destroy里先调~T()再归还内存。对象池解决的是“对象生命周期管理”,不是“任意大小内存分配”。如果业务对象内部还会自己new字符串或容器,那部分不归池管,别指望一个对象池把整条链路的分配都吃掉。
| 方案 | 适用场景 | 优点 | 代价 | 关键参数 |
|---|---|---|---|---|
| 固定块池 | 单一尺寸高频小对象 | O(1)、实现短、局部性好 | 多尺寸要建多个池 | block_size、blocks_per_slab |
| 分级池 | 多尺寸小对象 | 通用、碎片可控 | 元数据、class 设计复杂 | class 间距、最大小块 |
| 对象池 | 固定类型对象复用 | 构造析构封装、类型安全 | 类型绑定、异常安全要处理 | T、对齐、构造参数 |
选型时先看业务分配尺寸分布。我一般会抓一段线上或压测期间的分配大小直方图,按 p50、p90、p99 定 class,而不是拍脑袋按 2 的幂一路切到底。超过一两 KB 的分配、长生命周期缓存、大块 buffer,直接回退通用分配更稳;内存池的收益集中在小块、高频、短生命周期这三个条件同时成立的时候。
2.2 最小固定块池:用 free list 串起 slab 里的空闲块
先从最小可用版本开始:一次向系统申请一大块 slab,切成等长 block,每个空闲 block 的前几个字节存一个FreeNode*,把所有空闲块串成单链表。分配就是弹出头节点,释放就是压回头节点。这个结构不解决扩容、线程安全和对象析构,但它是后面分级池和线程缓存的底座。
#include <cstddef> #include <cstdlib> #include <new> class FixedBlockPool { public: FixedBlockPool(std::size_t block_size, std::size_t blocks_per_slab) : block_size_(align_up(block_size < sizeof(FreeNode) ? sizeof(FreeNode) : block_size, alignof(std::max_align_t))), blocks_per_slab_(blocks_per_slab) { slab_ = static_cast<std::byte*>(std::malloc(block_size_ * blocks_per_slab_)); if (!slab_) throw std::bad_alloc(); for (std::size_t i = 0; i < blocks_per_slab_; ++i) { push(slab_ + i * block_size_); } } ~FixedBlockPool() { std::free(slab_); } FixedBlockPool(const FixedBlockPool&) = delete; FixedBlockPool& operator=(const FixedBlockPool&) = delete; void* allocate() { if (!free_) return nullptr; // 生产环境应扩 slab 或回退通用分配 FreeNode* n = free_; free_ = n->next; n->~FreeNode(); return n; } void deallocate(void* p) { if (!p) return; push(p); } private: struct FreeNode { FreeNode* next; }; static std::size_t align_up(std::size_t n, std::size_t a) { // a 必须是 2 的幂;这里依赖 max_align_t 的对齐满足该条件 return (n + a - 1) & ~(a - 1); } void push(void* p) { free_ = new (p) FreeNode{free_}; } std::byte* slab_ = nullptr; std::size_t block_size_ = 0; std::size_t blocks_per_slab_ = 0; FreeNode* free_ = nullptr; };这段代码的关键点有三个。第一,block_size_不能小于sizeof(FreeNode),否则空闲链表指针都放不下;再按alignof(std::max_align_t)向上取整,保证 slab 内第 i 个块的起始地址仍然满足普通类型的对齐要求。第二,push用 placement new 在空闲块上构造FreeNode,allocate弹出后调用析构,避免把“未构造的内存”当对象用。第三,std::malloc返回的 slab 基址满足max_align_t对齐,但block_size_必须是该对齐的整数倍,否则第二块开始就会错位。
参数上,blocks_per_slab决定单次申请的大小和扩容粒度。太小会频繁申请 slab,太大则低峰期占着内存不还。常见起点是让block_size_ * blocks_per_slab_落在 64 KB 到 256 KB 之间;如果对象很小、分配极频繁,可以适当加大到 1 MB 以内,但要用压测看 RSS 和页故障。allocate返回nullptr表示当前 slab 用尽,生产代码要在这里接“新申请 slab 并挂到池上”的逻辑,不能直接让业务崩掉。
2.3 分级池的尺寸表怎么定:8 字节起步还是按业务分位
分级池的第一版通常从 16 字节起步,按 16、32、48、64、96、128、192、256、384、512 这样切。为什么不从 8 字节开始?在 64 位环境里,空闲链表指针本身就要 8 字节,块太小放不下元数据;而且很多类型要求 8 或 16 字节对齐,块起始地址不对齐,后面 placement new 就会踩 UBSan。把最小块定在 16 字节,能同时容纳指针和常见对齐要求,代价是小对象会有一点内部碎片。
| class 索引 | 块大小 | 典型对象 | 说明 |
|---|---|---|---|
| 0 | 16 | 小节点、索引项 | 至少能放下空闲链表指针 |
| 1 | 32 | 小任务、短消息 | 内部碎片可控 |
| 2 | 48 | 协议头、键值对 | 按 16 字节对齐 |
| 3 | 64 | 常见小对象 | 半个缓存行 |
| 4 | 96 | 中等对象 | 避免 class 过密 |
| 5 | 128 | 常见小对象上限 | 超过后看业务 |
| 6 | 192 | 稍大对象 | 与 128 拉开间距 |
| 7 | 256 | 中大对象 | 接近回退阈值 |
| 8 | 384 | 大一点的小对象 | 控制碎片 |
| 9 | 512 | 小对象上限 | 超过回退通用分配 |
class 之间的间距不要一味追求“省内存”。切得太密,空闲链表多、缓存行换入换出频繁,压测时可能出现 QPS 不升反降;切得太疏,内部碎片变大,RSS 会上去。我一般按 1.25 到 1.5 倍的间距起步,再根据业务直方图合并低频 class。最大小块先定在 512 或 1024 字节,超过就回退通用分配,别让池子变成“什么都管、什么都管不好”的黑匣子。
#include <cstddef> #include <iterator> constexpr std::size_t kAlign = alignof(std::max_align_t); constexpr std::size_t kMaxSmall = 512; std::size_t round_up(std::size_t n, std::size_t a) { return (n + a - 1) / a * a; } std::size_t size_class_index(std::size_t n) { static constexpr std::size_t classes[] = { 16, 32, 48, 64, 96, 128, 192, 256, 384, 512 }; n = round_up(n, kAlign); for (std::size_t i = 0; i < std::size(classes); ++i) { if (n <= classes[i]) return i; } return std::size(classes); // 等于 class 数量表示超出小对象范围 }这段映射代码只解决“请求大小到 class 索引”的正向查找。真正释放时还要反向找回 class:常见做法是在用户指针前面留一个 8 或 16 字节的块头,存 class 索引和 magic;或者把 slab 按页对齐,用页号查元数据表。块头方案实现快,但会破坏用户指针对齐,必须把块头也按max_align_t取整;页表方案对齐友好,但多一次查表和页大小假设。选哪个取决于你对可移植性和内存开销的要求,别在没压测前就拍板。
3. 把分配器接到业务里:placement new、RAII 与对齐分配
固定块池返回的是原始内存,不是对象。业务代码如果直接reinterpret_cast<T*>就用,构造没发生、析构没调用,单测可能全绿,ASan 一开就现形。这一层要把“内存分配”和“对象生命周期”分开,再用 RAII 把归还动作包住。
3.1 原始内存池与对象池:placement new 的构造/析构边界
对象池的职责很明确:create先向原始池要一块足够大的内存,再用 placement new 构造T;destroy先显式调用~T(),再把内存还给原始池。构造过程中如果抛异常,必须把内存归还,否则每抛一次就漏一块。析构函数原则上不要抛异常,否则destroy后半段的归还会被跳过。
#include <new> #include <utility> template <class T, class RawPool> class ObjectPool { public: explicit ObjectPool(RawPool& raw) : raw_(raw) {} template <class... Args> T* create(Args&&... args) { void* mem = raw_.allocate(sizeof(T), alignof(T)); if (!mem) return nullptr; try { return new (mem) T(std::forward<Args>(args)...); } catch (...) { raw_.deallocate(mem, sizeof(T), alignof(T)); throw; } } void destroy(T* p) { if (!p) return; p->~T(); raw_.deallocate(p, sizeof(T), alignof(T)); } private: RawPool& raw_; };这里要求RawPool提供allocate(size, align)和deallocate(ptr, size, align)两个接口,而不是只接收一个void*。为什么释放还要传大小和对齐?因为分级池需要 size 找回 class,over-aligned 类型需要对齐信息选择正确的释放路径。把这两个参数一路传下去,比在块头里猜要可靠。create的完美转发让构造函数参数原样传给T,destroy里先析构再归还,顺序不能反;反过来就是典型的 use-after-free 温床。
还有一个常见误用:把池返回的指针直接delete。池内存不是new出来的,delete会走通用分配器的元数据,轻则崩溃,重则静默破坏堆。业务侧要么统一用ObjectPool::destroy,要么用自定义 deleter 包成std::unique_ptr<T, PoolDeleter>,让 RAII 兜住异常路径和提前 return。
3.2 对齐与大小计算:align_up、max_align_t 和 over-aligned 类型
普通类型的对齐通常不超过max_align_t,但alignas(64)的结构体、SIMD 向量、某些缓存行对齐对象会超过。固定块池按max_align_t切块,只能保证普通类型对齐;遇到 over-aligned 类型,必须单独走 aligned new 或平台对齐分配,不能硬塞进普通块。
#include <cstddef> #include <cstdlib> #include <new> bool is_power_of_two(std::size_t n) { return n != 0 && (n & (n - 1)) == 0; } std::size_t align_up(std::size_t n, std::size_t a) { if (!is_power_of_two(a)) return n; // 生产代码应断言或抛错 return (n + a - 1) & ~(a - 1); } void* allocate_aligned(std::size_t size, std::size_t align) { if (align > alignof(std::max_align_t)) { return ::operator new(size, std::align_val_t{align}); } return std::malloc(size); } void deallocate_aligned(void* p, std::size_t align) { if (!p) return; if (align > alignof(std::max_align_t)) { ::operator delete(p, std::align_val_t{align}); } else { std::free(p); } }align_up只负责把大小向上取整到对齐倍数,它不会移动指针;真正决定块起始地址的是 slab 基址加上i * block_size_。所以 slab 基址必须满足最大对齐,block_size_也必须是该对齐的倍数,两个条件缺一不可。allocate_aligned里的分支是常见兜底:普通对齐走malloc,over-aligned 走 aligned new。释放时必须用匹配的路径,不能operator new配free。参数上,align必须是 2 的幂;sizeof(T)不包含尾部填充到对齐的部分,块大小要按align_up(sizeof(T), alignof(T))算,别只传sizeof(T)。
注意:分级池的块头如果放在用户指针前面,块头大小也要参与对齐计算,否则用户指针可能从 16 字节对齐变成 8 字节对齐,over-aligned 类型直接触发 UBSan。
3.3 线程缓存:每线程 free list 与批量搬移
固定块池和分级池如果共用一个全局锁,线程一多,锁竞争会把收益吃回去。常见做法是给每个线程一个thread_local缓存:线程本地先 pop/push,本地空了或超过上限,再跟中央池批量搬一批。批量大小batch决定锁竞争和内存占用的平衡,缓存上限max_cached防止低峰期线程占着大量空闲块不放。
#include <cstddef> #include <new> struct FreeNode { FreeNode* next; }; struct ThreadCache { static constexpr std::size_t kBatch = 32; static constexpr std::size_t kMaxCached = 256; FreeNode* free_list[16]{}; std::size_t count[16]{}; }; thread_local ThreadCache t_cache; void* pop(FreeNode*& head, std::size_t& count) { if (!head) return nullptr; FreeNode* n = head; head = n->next; --count; n->~FreeNode(); return n; } bool try_push(FreeNode*& head, std::size_t& count, void* p, std::size_t max_cached) { if (count >= max_cached) return false; head = new (p) FreeNode{head}; ++count; return true; } void* thread_cache_alloc(CentralPool& central, std::size_t cls) { auto& head = t_cache.free_list[cls]; auto& cnt = t_cache.count[cls]; if (void* p = pop(head, cnt)) return p; void* batch[ThreadCache::kBatch]; std::size_t n = central.refill(cls, batch, ThreadCache::kBatch); for (std::size_t i = 1; i < n; ++i) { try_push(head, cnt, batch[i], ThreadCache::kMaxCached); } return n ? batch[0] : nullptr; } void thread_cache_free(CentralPool& central, void* p, std::size_t cls) { if (!try_push(t_cache.free_list[cls], t_cache.count[cls], p, ThreadCache::kMaxCached)) { central.deallocate(p, cls); // 超出本地缓存上限,直接回中央池 } }这段代码只展示线程缓存的骨架,CentralPool的refill和deallocate需要接真实的分片空闲链表。关键参数是kBatch和kMaxCached:kBatch太小,锁竞争还在;太大,一次搬移的块多,低峰期线程缓存膨胀。kMaxCached太小,频繁回中央池;太大,线程退出或空闲时内存不释放。常见起点是kBatch = 16~64、kMaxCached = 128~512,再用多线程压测看 p99 和 RSS 调。线程缓存不是越多越好,超过业务并发度的缓存只会变成延迟释放的内存。
4. 并发与回收:无锁栈、分片锁和跨线程归还的取舍
内存池的并发设计有两个坑:一是以为 CAS 无锁一定比锁快,二是忘了空闲块也会跨线程流动。线程 A 分配、线程 B 释放是常态,如果 B 把块直接 push 到 A 的本地缓存,链表就可能在无锁状态下成环;如果所有归还都走中央锁,锁竞争又会回来。先看无锁栈的边界,再看分片锁和回收策略。
4.1 无锁 free list 的 ABA 问题:为什么教学代码一上并发就翻车
无锁空闲链表最经典的写法是 Treiber stack:push用 CAS 把头节点换成新节点,pop用 CAS 把头节点换成head->next。单线程看没问题,多线程下会碰到 ABA:线程 A 读到头节点是 X,准备把 X 弹出;线程 B 先弹出 X,又弹出 Y,再把 X 压回去;线程 A 的 CAS 仍然成功,但此时X->next已经变了,链表结构可能错乱。
#include <atomic> struct Node { Node* next; }; std::atomic<Node*> free_head{nullptr}; void push(Node* n) { Node* old = free_head.load(std::memory_order_relaxed); do { n->next = old; } while (!free_head.compare_exchange_weak( old, n, std::memory_order_release, std::memory_order_relaxed)); } Node* pop() { Node* old = free_head.load(std::memory_order_acquire); while (old && !free_head.compare_exchange_weak( old, old->next, std::memory_order_acquire, std::memory_order_relaxed)) { } return old; }这段代码能说明 CAS 循环的形状,但不能直接上生产。ABA 只是第一层问题;第二层是内存回收:你弹出节点后,另一个线程可能还在读它,什么时候真正复用这块内存需要 hazard pointer、epoch 回收或 tagged pointer。tagged pointer 用高位计数配合 128 位 CAS,能缓解 ABA,但可移植性和调试成本都不低。我的血泪经验是:如果团队里没有人能长期维护无锁内存回收,优先用分片锁或线程缓存,把复杂度压到可测试的范围。
注意:无锁不等于无等待。CAS 失败重试、缓存行争用、伪共享都会让 p99 抖动;用无锁栈之前先用压测证明锁是瓶颈,而不是凭感觉。
4.2 分片锁与线程缓存回收:把竞争拆到 8/16 个桶
分片锁的思路很朴素:把中央空闲链表拆成 16 或 32 个桶,每个桶一把锁,按线程 ID 或 class 索引哈希到桶上。线程大部分时间只碰自己的桶,锁竞争自然下降。桶结构要按缓存行对齐,避免两个热点桶落在同一个 64 字节缓存行里互相拖累。
#include <array> #include <cstddef> #include <mutex> #include <new> struct alignas(64) Shard { std::mutex mu; FreeNode* head = nullptr; }; class ShardedPool { public: static constexpr std::size_t kShards = 16; void push(void* p, std::size_t shard) { auto* n = new (p) FreeNode{nullptr}; Shard& s = shards_[shard & (kShards - 1)]; std::lock_guard<std::mutex> lk(s.mu); n->next = s.head; s.head = n; } void* pop(std::size_t shard) { Shard& s = shards_[shard & (kShards - 1)]; std::lock_guard<std::mutex> lk(s.mu); if (!s.head) return nullptr; FreeNode* n = s.head; s.head = n->next; n->~FreeNode(); return n; } private: std::array<Shard, kShards> shards_; };分片数一般取 2 的幂,方便用位运算取模。shard可以由(class_index * 131 + thread_seq) & (kShards - 1)算出来,让同一线程的同一 class 尽量落在固定桶。分片不是越多越好:16 到 32 个桶足够覆盖常见并发,继续加只会增加内存和缓存压力。跨线程归还时,释放线程可能哈希到不同桶,这没问题,因为桶本身有锁;但线程本地缓存不能直接跨线程 push,必须
本文还有配套的精品资源,点击获取