【手把手带你做项目】C++高并发内存池

我和码今天只能有一个

920人浏览 · 2025-10-14 21:15:00

我和码今天只能有一个 · 2025-10-14 21:15:00 发布

前言

这个项目是为了实现一个高并发的内存池，模拟一下高效的多线程内存管理，用于替代malloc、free。

此代码目前仅可在windows环境下运行，同时也解决了32位和64位的问题。Linux版本不行主要是没了解向系统申请内存的端口。

我们可能会用到C/C++的知识，以及数据结构，操作系统的线程，锁等知识。

在开始项目之前我们要先明确几个概念：
1.什么是并发

简单来说，并发的核心是任务切换—— 系统可以在多个任务之间快速切换执行，让外界看起来这些任务是在同时进行的。

并发与并行（Parallelism）的区别：

并发是 "看起来同时"（通过快速切换）并行是 "真正同时"（需要多核 CPU 支持）

2.什么是内存池

（1）池化技术

所谓“池化技术”，就是程序先向系统申请过量的资源，然后自己管理，以备不时之需。之所以要申请过量的资源，是因为每次申请该资源都有较大的开销，不如提前申请好了，这样使用时就会变得非常快捷，大大提高程序运行效率。

（2）内存池

我们每次要申请空间时，都需要进行malloc向堆请求一块空间，如果我们请求的次数特别的多，那么效率就会降低，内存池就是，我先申请一大段空间（即使你现在需要的只是一小部分），等下次我们向申请内存时，就不用去向堆申请了，直接在内存池拿就可以了。空间释放也是一样，相当于还给内存池。

3.内存池能够解决的问题

当然就是效率问题，其次也可以解决内存碎片的问题，。外部碎片是⼀些空闲的连续内存区域太小，这些内存空间不连续，以至于合计的内存足够，但是不能满足⼀些的内存分配申请需求。内部碎片是由于⼀些对齐的需求，导致分配出去的空间中⼀些内存无法被利用。内碎片问题，我们后面就会看到，那会再进行更准确的理解。

接下来我们就正式进行项目学习。

小试牛刀——创建一个定长的内存池

我们选择char*变量来创建我们的内存池（char一个字节比较好统计大小）然后用这个指针来接受malloc出来的空间。

template<class T>
class ObjectPool
{
   public:
   T*New()
    {
       T*obj;
       if(_memory=nullptr)
        {
          _memory=(char*)malloc(128*1024);//先开个128kb
          if(_memory=nullptr)
          {
            cout<<"error"<<end;
            exit(1);
          }
        }
          obj=(T*)_memory;
          return obj;
    }
  


  private:
      char*_memory=nullptr;
};

上面这个思路还有欠缺，比如我们的obj拿到内存池，它不可能拿着一整大块去使用，要从其中分出来一块使用，大小就是sizeof(T)，同时，我们也要使_memory把这段空间腾出来，供下次再向其申请时能成功分割.

第二，虽然我们申请的内存池够大，但总会有申请完的那一刻，因为我们申请后的空间使用完不会重新交给内存池，因此我们要解决的问题是：记录池内剩余空间的大小，解决用完不要的空间放在哪里的问题。

记录剩余大小，我们用一个int即可，每次申请空间后改成我们申请的大小，每次被拿走一块就减去对应的大小，不够了再去申请。

接受剩余空间我们用一个指针模拟链表即可。（细节一会说）

public:
   T*New()
    {
       T*obj;
       if(_memory=nullptr)
        {
           _remainSize=128*1024;
          _memory=(char*)malloc(_remainSize);//先开个128kb
          if(_memory=nullptr)
          {
            cout<<"error"<<end;
            exit(1);
          }
        }
          obj=(T*)_memory;
          _memory+=sizeof(T);
          _remainSize-=sizeof(T);
            return obj;
    }
  


  private:
      char*_memory=nullptr;
      int _remainSize=0;
      void*_freelist=nullptr; //回收空间的链表

这样一看我们又出现了新的问题，难道我们每次new都需要去内存池要空间吗？当然不是，如果freelist中有用完的空间，我们直接拿过来用即可，这样也可以减少malloc的次数。

现在问题就在于：我们怎么通过一个指针实现链表的功能？

原理其实不难，每一个块存放下一个块的地址就可以连起来，但是一整个块都用来存是不是有点太冗余了。我们可以用前四个字节来存下一个地址。

public:

   T*New()
     {
        T*obj;
         if(_freelist)
        {
          int*next=*(int*)_freelist;//先记录下一个的地址
          obj=(T*)_freelist;
          _freelist=next;
        }
        else {....}
       return obj;
     }

   void Delete(T*obj)
   {
      *(int*)obj=_freelist;
      _freelist=obj;
   }



private:
      char*_memory=nullptr;
      int _remainSize=0;
      void*_freelist=nullptr; //回收空间的链表

思路就是我们先把obj换成int*类型再解引用就可以拿到一个int大小前四个字节，进而就能获取到地址。但这里不太合理，因为你不知道机器是32位还是64位，32位下这个可以，但64位下地址就是八字节了，你只取四个是行不通的。

方法：把上面的*(int*)换成*(void**)用二级指针即可也不必是void，int，T也可以，这样的话解引用还是指针，void*在32位下就是4字节64位就是八字节，完美解决了这个问题。

还有一些其他问题，如果T是char或int等类型，在64位下是4，但是我们要用八字节去存内存块的地址，那么就会不够用，因此我们在申请内存大小时也要做修改。

以下是定长内存池完整的代码：

#include <iostream>
using std::cout;
using std::cin;


template<class T>
class ObjectPool
{
   public:
   T*New()
   {
     T*obj=nullptr;
      //如果freelist不为空则不需要再切或申请数据库
      //直接从freelist中拿
      if(_freelist)
      {
           void*next=*(void**)_freelist;//*先拿到再强转，拿到第二个块的地址
           obj=(T*)_freelist;   //*
           _freelist=next;
      }
      else//freelist==nullptr;
      {
        if(_remainSize<sizeof(T))//剩余字节数不足以开空间，只能舍弃
        {
            _remainSize=128*1024;//模拟开个128kb
            _memory=(char*)malloc(_remainSize);
            if(_memory==nullptr)
            {
                cout<<"malloc error"<<endl;
                exit(1);
            }
            
            
        }
        
            obj=(T*)_memory;//*每次获取空间后必须强转成模板类型
             //64位下就给一个void*的大小（如果是int等类型）确保够用
            size_t objsize=sizeof(T)<sizeof(void*)?sizeof(void*):sizeof(T);//*
            _memory+=objsize;
            _remainSize-=objsize;
      }
      //定位new初始化
      new(obj)T;
      return obj;
   }

   void Delete(T*obj)
   {
     obj->~T();
      //头插进freelist，无论freelist是否为空都可以这么写
      *(void**)obj=_freelist;
      _freelist=obj;
   }

   private:
     char*_memory=nullptr;//申请大块内存空间的指针
     int _remainSize=0;//大块内存中剩余的空间（字节数）
     void* _freelist=nullptr;//空间释放时回收空间的链表
};

高并发内存池整体架构

很多的开发环境都是多核多线程，在申请内存的场景下，必然存在激烈的锁竞争问题。但我们的项目原型tcmalloc在多线程高并发的场景下更高效，因此我们要解决的问题是：性能问题，多线程下锁的竞争问题以及内存碎片问题。

整体由三个部分构成：

threadcache：线程缓存是每个线程独有的，用于小于256KB的内存的分配，线程从这里申请内存不需要加锁，每个线程独享⼀个threadcache，这也就是这个并发线程池高效的地方。

central cache：中心缓存是所有线程所共享，threadcache是按需从centralcache中获取的对象。centralcache合适的时机回收threadcache中的对象，避免⼀个线程占用了太多的内存，而其他线程的内存紧张，达到内存分配在多个线程中更均衡的按需调度的目的。centralcache是存在竞争的，所以从这里取内存对象是需要加锁，首先这里用的是桶锁，其次只有threadcache的没有内存对象时才会找centralcache，所以这里竞争不会很激烈。

pagecache：页缓存是在centralcache缓存上面的⼀层缓存，存储的内存是以页为单位存储及分配的，centralcache没有内存对象时，从pagecache分配出⼀定数量的page，并切割成定长大小的小块内存，分配给centralcache。当⼀个span的几个跨度页的对象都回收以后，pagecache会回收centralcache满足条件的span对象，并且合并相邻的页，组成更大的页，缓解内存碎片的问题。

Threadcache的封装

这个就相当于每一个线程自己的内存池，因此我们要封装对应的freelist和memory。但是我们在上面只能搞定4和8字节的申请情况，如果是不规则的字节数可能就会产生浪费问题，比如内碎片（假设我可以搞定8字节，但是你的大小只有3字节，但我还是要给你8字节，剩下的就是内碎片用不到）（外碎片是我们申请的一大块连续空间由于不规则的分块释放导致内部释放的空间不连续的碎片），因此我们要对应不同的大小建立对应的字节数的freelist，但也不可能1一节1链表吧，那也太多了，我们可以以8字节为一个单位弄一个链表，然后把他们放在hash桶中形成对应就可以了。

我们先搞定一下基础的Freelist类（在公共部分common.hpp中）

关于回收链表无非就两个接口，插入和删除数据块。

//Common.hpp
#include <iostream>
#include <assert.h>

class Freelist
{
    void Push(void*obj)
    {
        *(void**)obj=_freelist;//获取obj前4/8字节来保存下一个的地址
        _freelist=obj;
    }

    void*Pop()
    {
        assert(_freelist);
        void*obj=_freelist;
        _freelist=*(void**)obj;
    }
   
    void*_freelist;
};

建立哈希桶的对应关系

我们说过，为了使链表的数量足够少，我们规定了8字节为一个单位对齐，但如果是这样也需要256*1024/8=32768个桶也是很多。因此我们换一种规则：

//整体控制在最多10%左右的内碎⽚浪费
// 字节数               该范围的对齐规则    对应桶的数量下标
// [1,128]              8byte对⻬      freelist[0,16)
// [128+1,1024]         16byte对⻬     freelist[16,72)
// [1024+1,81024]       128byte对⻬    freelist[72,128) 
// [8*1024+1,641024]    1024byte对⻬   freelist[128,184) 
// [64*1024+1,256*1024] 8*1024byte对⻬ freelist[184,208)

我们解释以下，假如说我们要的字节数在第一个范围内（假设是7），那么根据8字节对齐规则就应该申请8个字节如果是申请14个字节就会给2*8=16个字节，后面桶的下标就是字节范围数/对齐规则。

因此，我们还要在Common.h中封装一下映射桶的类。

这个对齐规则我们思路还是不难的，接下来看看映射的思路

对于子函数，我们的对于1-8，发现1-7/8都是可以对应0的但是8也在范围内却对应1，因此我们要对边界进行判断一下。

对于主函数，我们定义了数组，元素记录着每一个区间的桶的数量，这样在后面的映射可以对应正确的区间（相当于/用于标记你在第几组%相当于你在组开始向后走了多少步）

有了桶的映射，我们就可以根据申请大小去对应的链表拿内存了。

else的情况是当前链表没有内存了，那只能老实的再向上申请了。

Threadcache的TLS无锁访问

现在我们的链表哈希桶就基本实现了，如果桶中有链表空间，那么效率是非常高的。每一个线程独有自己的空间桶。但是，这个线程如何获取到自己的thread cache？以及如何知道这个thread cache对应哪一个线程？也不能搞成全局的，那就变成多抢一的锁问题了。因此我们需要借助一个东西——TLS线程本地存储

是一种变量的存储方法，这个变量在它所在的线程内是全局可访问的，但是不能被其他线程访问到，这样就保持了数据的线程独立性。而熟知的全局变量，是所有线程都可以访问的，这样就不可避免需要锁来控制，增加了控制成本和代码复杂度。

我们这里用windows下的静态TLS

有了这个变量，我们就可以拿着变量去申请对应的Threadcache了，（调用<thread>然后创建线程调用下面的函数）

对应的Threadcache的malloc和free

除此之外，Threadcache的Deallocate我们还没有定义

暂总结一下流程

从下而上的分析，现在我们是有多个Threadcache*的线程的变量，然后当想要空间时，传一个大小的参数size，然后就会调用ConcurrentAlloc函数（在ConcurrentAlloc.hpp）去申请空间，而该函数的底层就是new一个Threadcache的对象去调用它的Allocate函数，然后Allocate函数就会进行对齐和映射找到对应的桶里链表去获取，如果链表是空的，那么只能再向上去Centralcache拿块空间了。

释放也一样，Threadcache*线程变量->ConcurrentFree->Deallocate->放入桶。

Central cache的整体结构和设计

Central cache也是一个哈希桶的结构，其结构与threadcache有很多相似之处，当threadcache的内存申请光了后，所有的threadcache都会到centralcache申请内存，因此，centralcache的设计是需要带锁的，但为了保证高效，我们并不是说每次只让一个threadcache访问centralcache，而是设立一个桶锁，也就是说，我们的Cc哈希的每一个桶中都设立一个锁，如果两个或多个threadcache访问的不是一个桶，那么就不会有锁的干扰，体现了高性能，如果是多个threadcache访问一个桶就要锁的竞争了。

那么Cc哈希桶中存的是什么呢？其实就是更大块的内存，但每一个桶中存放的是存span的链表(但两个哈希的桶的数量和区分规则是相同的），这个span是以页为单位的大块的内存（一个span可能有多页大小）。从这个span中的内存切成对应的块给threadcache。一个span中会有多个单位大的内存块，（我们这里规定一页大小是8k）（比如我想申请一个8bytes，Cs直接把Span中的10块8bytes给我的Tc，这样下次申请就不需要向Cc要了）

每一个桶中都存放的是一个span的链表，只不过不同桶中的span里span中的页被切成块的块大小不同，span的大小也不尽相同。

但某单位的Span链表用完了就会再向上要内存(Page cache）

接下来我们来看一下Cc的结构

首先，Cc是一个哈希桶，桶里面是记录一个个Span的链表

解释一下名词，页号就相当于把一定大小内存按单位大小（这里是8k）分成若干个，每一个称为页。对应的下标就是页号。

此外，我们桶中的链表是一个带头双向循环链表，为了查找链表内是否还有剩余内存以及使用情况。还有一个回收用过的内存链表。

下面就是双向链表的思路

每一个链表就代表一个桶，因此内部成员变量要有锁。

这样Cc的结构初步就出来了

Pagecache（以下简称Pc）的整体结构

首先，Pc的底层也是一个哈希桶，其内部也是一个存放Span的链表，但和Cc不同的是，它的桶的映射结构并不像Tc和Cc一样按字节数映射，而是按页数映射，规定1-128个页，每个桶中存放对应页大小的Span（虽然也是span，但其内部是一整个大块内存，不会像Cc的span被分成若干小块）。

注意的是：当Tc和Cc都没有对应的空间申请，那么Cc就会向Pc中申请，根据申请的页数去对应的桶拿。但如果这个桶也没有，它首先会向上查找，（假如我要2页的span但是没有了，就会继续向上找有没有更大页的span）假设我找到了120页的Span，那么120页的Span就会分裂成2和118页的Span供使用，只有当128页的Span也没有时，才会去堆中申请一个128页内存继续分裂。

释放的话，如果Tc回收的多了就会还给Cc如果Cc多了就会还给Pc，而从Pc拿走的Span都是有页号的，如果回收后发现相邻页号的Span均是未使用就会进行合并，解决了内存碎片的问题。

Pc的类也是采用单例模式，我们接下来要解决的问题是如何实现Pc把一个对应页的Span给Cc进而再向下交付。（getonespan函数）

此外，Pc是不用桶锁的，而是要一个大锁控制整个Pc。（如果1页和2页Span都没有就会向上找3页的Span，如果是桶锁就会出现问题）

我们先简单看一下Pc的初步结构

首先我们要弄清楚，每一个Span下都有一个回收链表

Getonespan函数实现

我们再来分析一下运行流程，每一个线程获取它的threadcache然后Tc空间不够调用FetchfromCc函数，在这个函数中我们要根据给的大小以及映射关系给予相应的空间（利用Fenchrangeobj函数把span切割然后返回）但span切割前我们要得到一个span也就是getonespan函数。

我们先来看一下整体函数细节。

这个newspan我们稍后实现（获取一个span,span的页由传参决定）它的参数我们用了一个NumMovePage函数，传入单位大小，就会按规则分配

我们默认一页是8k，反正原则是size越小页越少，size越大页越大。

下面我们需要获取页的起始地址，公式：页号<<页大小（8k）（第0号第0字节地址是0，且内存连续）

bytes就是span的大小，页个数*8k。这样end就可以记录到大块内存的末地址了。

接下来我们就要把span的start-end切下来放到Cc的桶链上（尾插到链表上保持地址取走时一致）。

这里加了注释能更方便一些，总的来说就是我们先新建一个Span对象然后把大块内存切分成若干块挂在对象上。然后再把这个有空间的span给spanlist

Newspan函数的实现

首先newspan函数的参数是你要一个多少页的span（因为Pc是以页数分成不同的桶的，1-128个桶）

我们根据Pc的结构首先要知道大体流程：去对应桶获取，如果桶空了，向上找更大的，如果找到就切分，找不到就向堆申请最大的页。

我们说一下if以下的语句，nspan是从大桶里准备拿出来切分的span然后我们把它的页号直接给新的kspan，因为kspan有固定需要的页我们直接赋值即可。同时我们也要把nspan的页数-k，因为把nspan的k页给了kspan，因此nspan下一个起始页号就要+k。这样k页的span就切好了，然后就是把剩下被切的重新放桶里。

如果128桶里都没有就需要申请了，图中的system函数是windows向堆申请内存特定的函数，在此就不展示了（virtualalloc），它被申请完以后，这下就可以找到k页的span了，也就是我们需要再来一次上面的for循环，但为了代码简洁，我们采用递归写法（这种程度的递归与重新遍历同比差别并不大可忽略）。

关于Cc和Pc加锁的问题

我们知道Tc获取内存是不用加锁的，但当Tc为空或不足，就需要从Cc获取Span切块内存，这时就需要加锁了。这里我们谈一下加锁的细节和时机。

首先，我们申请前肯定是要加锁的，但当我们从Cc中获取到了一个span就可以解锁了，因为我们此时获取Span别的线程是拿不走的，但如果不解锁其他线程空间的释放就会有问题。因此，当Getonespan函数结束后先解锁。

此外，我们向Pc申请span也是需要加锁的，至于解锁时机也是当我们结束了newspan函数就可以解锁。但我们不建议写在for循环里，毕竟一次循环加锁解锁太繁琐了。我们可以写在getonespan函数里。

关于高并发内存池申请内存流程的总结

1.Tc、Cc、Pc的结构回顾

Tc：桶中的每一个位置挂着的是一块块待使用的、大小适配的内存块。

Cc：桶中的每一个位置是一个Spanlist的类型（双向链表），而这个链表里的每一个节点是Span类型，每一个Span类型都会有一个用于回收的自由链表freelist。

Pc：同Cc只不过Pc的Span是未切的大块内存。而Cc的是已经根据页数切好并放在桶里待使用的，而Cc的一个Span又会切成若干份给对应的Tc。

2.最开始均为空的内存申请流程

t1首先利用Concurrentalloc，new一个Tc的指针对象，然后根据传入的size参数告诉函数我们要申请的字节数，然后Concurrentalloc调用Tc的Alloc函数并把size传给他，Alloc根据size算出我们的对齐字节数已经对应的桶，发现桶里没有多余的内存块可用，调用FetchFromCc去向Cc要若干块，在FFCc函数中，我们根据size算出其应给的块数（应与链表的maxsize比，谁小谁为主，慢增长算法）然后调用FenchRangeObj函数把Cc对应的桶的内存切对应数量的块给Tc，因此我们要在对应桶里拿到一个Span再切。结果发现Cc也是空，没办法，只能去newspan再向上要，如果Pc的该桶没有了（Cc要多少页的Span就会去Pc几号桶去找），就会向上找知道找到并切成两份向下传递。

3.再次强调Cc和Pc中Span的区别

Cc的Span，是对应的桶中切成了对应的大小（比如8byte的桶下挂的Span都是被切成了8byte大小的若干块，256byte就是256大小的若干块）

Pc的Span，是以页为单位分别映射在对应的桶中，不需要切分，需要的时候再切。

以上就是我们内存池大致的申请内存流程，除此之外我们还说过，我们的高并发内存池还可以解决内存碎片问题，这就涉及到内存的释放问题。下面我们来谈谈三个结构的内存释放的代码编写。

Tc内存释放函数

我们知道，线程申请内存就会去Tc释放，当用完时就会放入对应桶的freelist中，但是如果我们一直push，甚至freelist的长度超过我们之前一次性批量给的数量，那么就会导致其他线程可能拿不到内存的问题，因此，当回收链表的内存数量够多，我们应适当进行回收给Cc，然后其他线程就可以使用这些内存，提高利用率。

注释下面就是我们对于回收函数的编写，我们再看一下listtoolong函数

poprange就是我们把链表的若干块拿出链表，下面的RLTS函数就是将这些拿下来的内存块交付给Cc的过程，当然这就属于Cc的回收范畴了一会我们再说。

因此，我们要在Freelist类型再添加一个成员变量_size用于记录当前回收链表的长度，它的push，pop，poprange，pushrange都要改动_size。

（注意，Poprange的start和end是输出型参数）

总结一下，当freelist数量大于等于其maxsize，进行回收，先把链表的目标数量块拿出链表然后调用RTLS函数上交给Cc。

Cc回收内存

我们说过，我们的高并发内存池还要解决内存碎片问题，这就意味着我们拿走的内存在还回来时要按原来的位置进行存放，也就是给原来的span，但是，我们如何才能找到对应的span呢？

只需要找到这个span对应哪个页即可，因为我们并不关心回来的时候内存的顺序，只需要确保他们是连续的即可，因此，当Pc把它的span给到Cc时，我们都要记录一下其对应哪一（些）页（因为一个span可能包含多个页），我们在Pc结构中再加一个unordered_map结构。

同时我们也要写一个span和页的对应关系

这里我们是拿走了k页，因此这几个页号都对应着kspan（kspan包含这些页）。

把传进来的内存地址强转然后右移13位获得当前的页号（因为在此页里所有的span经过pageshift都会对应此页号），然后根据此页号找到之前映射的span。

接下来我们就可以完成刚才的RTLS函数了

传size的原因就是我们要根据映射规则进行加锁再操作。剩下的注释已经有解释。这里的usecount指的是Cc中的span。

总结一下，我们建立映射的目的就是为了查找当时Cc给Tc时是哪个（些）span，然后找到后把内存挂在这些span的freelist中。

Pc的内存回收

当Cc的span的usecount为0时，证明此span里的所有块内存都回来了，因此可以向上交给Pc，

把此Span交付给Pc时，为了解决外碎片问题，我们要将相邻页号的span尝试进行合并成一个大span，这就涉及到几个问题，如果前/后的span如果还有Cc正在使用的我们是无法进行合并的，以及我们已知的span的页号如何去找到之前页号对应的span，为了解决这些问题，我们在把大span切分成小span时，可以把span 的首尾页号放进unordered_map中存储便于合并时span的查找（之前的Cc映射不用改，只是这一步为了合并）