【C++高并发内存池实战】从 new/malloc 升级到定长内存池，让分配像喝水一样快

weixin_43925427

1116人浏览 · 2025-11-27 15:45:08

weixin_43925427 · 2025-11-27 15:45:08 发布

一、为什么还要折腾“内存池”？

先说结论：
在高并发场景下，如果你频繁 new/delete 相同尺寸的小对象，用定长内存池，性能和抖动会好很多。

传统 new/malloc 在高压场景的几个痛点：

系统调用 & 管理开销大
- 系统分配器要维护整块堆空间、元数据、合并/切分空闲块，逻辑很重。
- 频繁分配/释放小块时，开销完全不小。
内存碎片严重
- 你要 64 字节，系统可能给你 80、96……
- 长期运行后，碎片多了，整体内存占用会远高于你“逻辑上”的需求。
多线程锁竞争
- 大多数系统分配器内部要加锁或使用复杂的同步策略。
- 多线程高并发 new/delete，互相抢锁，延迟突刺就来了。
行为不可控
- 你很难预估一次分配的真实开销，也很难对齐业务的生命周期管理。

定长内存池做的事情本质上就一句话：

“我要一大块内存，切成 N 份固定大小的小块，自己管理 free list，分配释放都 O(1)，还尽量少加锁。”

二、先从单线程定长内存池写起

在这里插入图片描述

先别急着上多线程，所有复杂的并发优化，都是从简单版本长出来的。

我们假设一个典型场景：

你有大量 struct Node { ... };，大小固定，比如 64 字节；
这些 Node 会频繁创建和销毁；
你不希望每次都 new / delete。

2.1 核心设计

单线程定长内存池的核心数据结构可以极度朴素：

一块连续内存，被切成 N 个小块；
每个小块里前几个字节拿来当指针：形成一个单向链表；
allocate()：从链表头取一个块；
deallocate(p)：把块塞回链表头。

示意代码（核心逻辑）：

#include <cstddef>
#include <cstdint>
#include <vector>
#include <stdexcept>

class FixedSizePool {
public:
    FixedSizePool(std::size_t blockSize, std::size_t blockCount)
        : m_blockSize(adjustBlockSize(blockSize)),
          m_blockCount(blockCount),
          m_memory(nullptr),
          m_freeList(nullptr)
    {
        if (m_blockSize == 0 || m_blockCount == 0) {
            throw std::invalid_argument("blockSize and blockCount must be > 0");
        }
        initPool();
    }

    ~FixedSizePool() {
        ::operator delete(m_memory, std::align_val_t(alignof(std::max_align_t)));
    }

    void* allocate() {
        if (!m_freeList) {
            // 可以考虑扩容，这里简单点：直接抛异常
            throw std::bad_alloc{};
        }
        Node* node = m_freeList;
        m_freeList = m_freeList->next;
        return node;
    }

    void deallocate(void* ptr) {
        if (!ptr) return;
        Node* node = static_cast<Node*>(ptr);
        node->next = m_freeList;
        m_freeList = node;
    }

    std::size_t blockSize() const { return m_blockSize; }

private:
    struct Node {
        Node* next;
    };

    std::size_t m_blockSize;
    std::size_t m_blockCount;
    void*       m_memory;   // 整块大内存
    Node*       m_freeList; // 空闲链表头

    static std::size_t adjustBlockSize(std::size_t sz) {
        // 至少要能放下一个指针
        if (sz < sizeof(Node)) sz = sizeof(Node);
        // 简单对齐：向上对齐到 pointer 大小
        constexpr std::size_t align = alignof(std::max_align_t);
        if (sz % align != 0) {
            sz += align - (sz % align);
        }
        return sz;
    }

    void initPool() {
        std::size_t totalSize = m_blockSize * m_blockCount;
        m_memory = ::operator new(totalSize, std::align_val_t(alignof(std::max_align_t)));

        // 切块并串成 free list
        std::byte* p = static_cast<std::byte*>(m_memory);
        m_freeList = nullptr;
        for (std::size_t i = 0; i < m_blockCount; ++i) {
            Node* node = reinterpret_cast<Node*>(p + i * m_blockSize);
            node->next = m_freeList;
            m_freeList = node;
        }
    }
};

这个版本有几个特点：

分配/释放都是 O(1)，只改一个指针；
内部不涉及系统层面的碎片管理；
但 只适合单线程使用（free list 是非线程安全的）。

三、让内存池“扛住”多线程：从锁到无锁

单线程版本跑在多线程环境下，马上就会炸：多个线程同时操作 m_freeList，数据结构直接被写坏。

多线程方案常见有三条路：

粗暴加锁：全局 mutex 把 allocate/deallocate 包起来；
线程本地缓存（thread-local pool）：每个线程有自己的 free list，减少锁争用；
无锁结构：例如基于 std::atomic 的无锁栈/队列做 free list。

我们按演化顺序说。

3.1 方案一：全局锁（简单但不香）

最直接的写法：

#include <mutex>

class FixedSizePoolMT {
public:
    FixedSizePoolMT(std::size_t blockSize, std::size_t blockCount)
        : m_pool(blockSize, blockCount) {}

    void* allocate() {
        std::lock_guard<std::mutex> lock(m_mutex);
        return m_pool.allocate();
    }

    void deallocate(void* p) {
        std::lock_guard<std::mutex> lock(m_mutex);
        m_pool.deallocate(p);
    }

private:
    FixedSizePool m_pool;
    std::mutex    m_mutex;
};

优点：

实现爽快，不容易出 bug；
对很多中等并发场景其实就够用。

缺点：

高并发下，所有线程抢同一个锁，扩展性有限；
延迟抖动明显。

如果你现在项目里的 QPS 不高，这个版本已经是“白嫖优化”：比系统分配器更可控，代码又不复杂。

3.2 方案二：线程本地内存池（推荐）

在这里插入图片描述

真正“高并发友好”的做法，是：绝大部分分配释放只在当前线程自己的 free list 内完成，完全无锁：

每个线程有一个 FixedSizePool 或一个小的 free list“缓存”；
当本地池耗尽时，再从“全局大池”拉一批块；
当本地池太多空闲时，反向还给全局池。

下面给一个简化版的线程本地方案：每个线程一份池，不做全局回收，逻辑简单好理解：

class ThreadLocalFixedPool {
public:
    static void* allocate(std::size_t size) {
        auto& pool = getPool(size);
        return pool.allocate();
    }

    static void deallocate(void* p, std::size_t size) {
        auto& pool = getPool(size);
        pool.deallocate(p);
    }

private:
    // 每个线程维护自己的一份 pool
    static FixedSizePool& getPool(std::size_t size) {
        // 对不同 size 分类是更完整方案，这里简化：只支持单一 size
        thread_local FixedSizePool pool(size, 1024); // 每线程 1024 块
        return pool;
    }
};

用法：

struct MyObject {
    int a;
    double b;
    // ...
};

void* operator new(std::size_t sz) {
    return ThreadLocalFixedPool::allocate(sz);
}

void operator delete(void* p, std::size_t sz) noexcept {
    ThreadLocalFixedPool::deallocate(p, sz);
}

特点：

对于 MyObject 这种大量在同一线程创建/销毁的对象，非常友好；
线程之间完全无锁；
缺点是：跨线程释放（A 线程申请、B 线程 delete）会出问题，这个简单版没处理。

真正生产级的实现（比如 TCMalloc、JEMalloc）会有：

每线程的 small cache；
中央缓存负责不同 size class 池之间的批量分发和回收；
跨线程释放则先回收到“目标线程”的结构或者中央缓存中。

但对多数业务来说，你只要保证对象在哪个线程申请，就在哪个线程释放，这种 thread-local 小池就已经很香了。

四、一个完整的“定长对象池”封装示例

上面都是对 void* 级别的池管理。实际业务里，我们更希望有一个模板对象池，负责构造/析构：

#include <new>
#include <utility>

template <typename T>
class ObjectPool {
public:
    explicit ObjectPool(std::size_t count)
        : m_pool(sizeof(T), count) {}

    template <typename... Args>
    T* create(Args&&... args) {
        void* mem = m_pool.allocate();
        // 在池中做 placement new
        return new (mem) T(std::forward<Args>(args)...);
    }

    void destroy(T* obj) {
        if (!obj) return;
        obj->~T();
        m_pool.deallocate(obj);
    }

private:
    FixedSizePool m_pool;
};

使用示意：

#include <thread>
#include <vector>
#include <iostream>

struct Node {
    int x;
    double y;
};

int main() {
    ObjectPool<Node> pool(100000); // 10 万个 Node

    const int threadCount = 8;
    const int perThreadOps = 100000;

    auto worker = [&pool, perThreadOps]() {
        std::vector<Node*> nodes;
        nodes.reserve(perThreadOps);
        for (int i = 0; i < perThreadOps; ++i) {
            Node* p = pool.create(i, i * 0.1);
            nodes.push_back(p);
        }
        for (Node* p : nodes) {
            pool.destroy(p);
        }
    };

    std::vector<std::thread> threads;
    for (int i = 0; i < threadCount; ++i) {
        threads.emplace_back(worker);
    }
    for (auto& t : threads) t.join();

    std::cout << "done\n";
    return 0;
}

这段用的是“带锁版池 + 多线程”，如果你把底层换成 thread-local 版本，在高并发压测里差距会非常明显。

五、和 new/malloc 对比性能：怎么测？

正式写 CSDN 博客时，建议你可以加一个简单的 benchmark 小节（这里只给思路）：

准备两套路：
- A 组：直接 new Node / delete Node；
- B 组：用 ObjectPool<Node> 创建/销毁。
开 N 个线程，每个线程反复：
- 分配 M 次；
- 再释放 M 次；
用 std::chrono::high_resolution_clock 记录总耗时；
统计：
- 总时间；
- 每次操作平均耗时；
- 峰值 RSS（可以用外部工具看）。

一般来说，在小对象 + 高并发场景下，定长内存池能给你带来数量级上的稳定性提升，尤其是抖动会小很多。

六、几个常见坑，一定要提前踩平

对齐问题（alignment）
- 如果池里的对象有 double、std::max_align_t，一定要保证块对齐。
- 上面我们用 std::align_val_t(alignof(std::max_align_t)) 做了对齐。
越界写 = 池全剧终
- 任何越界访问都会破坏 free list，之后分配/释放会各种炸。
- 调试时建议配合 ASan / Valgrind。
释放非本池指针
- 要么协议上保证“谁分配谁释放”；
- 要么池里做额外保护，比如范围校验（但有额外开销）。
跨线程使用
- 如果用 thread-local 池，要保证对象生命周期 confined 在单线程。
- 否则需要更复杂的设计（中央缓存 + 跨线程释放策略）。
池的生命周期
- 如果你用 thread_local + static 对象，要注意程序退出顺序，防止先析构池再析构对象（或者反之）。
- 通常做法：
  - 池是进程级单例、早建后析构；
  - 或者彻底自己控制创建/销毁时机。