高并发内存池实战：用C++构建高性能服务器（上）-阿里云开发者社区

前言：(Memory Pool)是一种内存分配方式，又被称为固定大小区块规划（fixed-size-blocks allocation）。通常我们习惯直接使用new、malloc等API申请分配内存，这样做的缺点在于：由于所申请内存块的大小不定，当频繁使用时会造成大量的内存碎片并进而降低性能。

秋招可以写进简历的6个实战项目：

当前项目是实现一个高并发的内存池，他的原型是google的一个开源项目tcmalloc，tcmalloc全称Thread-Caching Malloc，即线程缓存的malloc，实现了高效的多线程内存管理，用于替代系统的内存分配相关的函数（malloc、free）。

这个项目的要求的知识储备？

这个项目会用到C/C++、数据结构（链表、哈希桶）、操作系统内存管理、单例模式、多线程、互斥锁等等方面的知识。

一、基础概念

1.1什么是内存池

所谓池化技术，就是程序先向系统申请过量的资源，然后自己管理，当程序中需要申请内存时，不是直接向操作系统申请，而是直接从内存池中获取，释放内存时也不是将内存返回给操作系统，而是返回内存池中。

因为每次申请该资源都有较大的开销，这样提前申请好了，使用时就会非常快捷，能够大大提高程序运行效率。在计算机中有很多使用这种池技术的地方，例如线程池、连接池等。

动态内存申请malloc

C++中动态申请内存都是通过malloc去申请的，但实际上我们并不是直接去堆中获取内存的，而malloc就是一个内存池。

malloc() 相当于向系统 “批发” 了一块较大的内存空间，然后“零售” 给程序使用，当全部使用完或者程序有大量内存需求时，再根据需求向操作系统申请内存。

1.2内存池

内存池是指程序预先从操作系统申请一块足够大内存，此后，当程序中需要申请内存的时候，不是直接向操作系统申请，而是直接从内存池中获取；同理，当程序释放内存的时候，并不真正将内存返回给操作系统，而是返回内存池。当程序退出(或者特定时间)时，内存池才将之前申请的内存真正释放。

内存池主要解决的问题

内存池解决

效率（主要）
内存碎片。（系统的内存分配器的角度）

那么什么是内存碎片呢？

外部碎片：一些空闲的连续内存区域太小，这些内存空间不连续，以至于合计的内存足够，但是不能满足一些的内存分配申请需求

内部碎片：由于一些对齐的需求，导致分配出去的空间中一些内存无法被利用。

1.3malloc

C/C++中我们要动态申请内存都是通过malloc去申请内存，但是我们要知道，实际我们不是直接去堆获取内存的。而malloc就是一个内存池。malloc() 相当于向操作系统“批发”了一块较大的内存空间，然后“零售”给程序用。当全部“售完”或程序有大量的内存需求时，再根据实际需求向操作系统“进货”。

malloc的实现方式有很多种，一般不同编译器平台用的都是不同的。比如windows的vs系列用的微软自己写的一套，linux gcc用的glibc中的ptmalloc。

原理图

二、定长内存池设计

2.1概述

作为程序员(C/C++)我们知道申请内存使用的是malloc，malloc其实就是一个通用的大众货，什么场景下都可以用，但是什么场景下都可以用就意味着什么场景下都不会有很高的性能，下面我们就先来设计一个定长内存池。

2.2设计思路

开辟内存：

使用malloc开辟一大块内存，让_memory指针指向这个大块内存
_memory 设置为char* 类型，是为了方便切割时_memory向后移动多少字节数。

申请内存：

将_memory强转为对应类型，然后赋值给对方，_memory指针向后移动对应字节数即可。
如果有存在已经切割好的小块内存，则优先使用小块内存。

释放内存：

用类型链表的结构来进行存储。
用当前小块内存的头4字节存储下一个小块内存的地址，最后用_freeList指针指向第一个小块内存的地址（并不是将内存释放给操作系统）
所以开辟内存时，开辟的内存大小必须大于或等于一个指针类型的大小。

代码位置：高并发内存池项目中的ObjectPool.h 文件。

2.3内容讲解

注意：核心三变量

char* _memory = nullptr; // 指向大块内存的指针
size_t _remainBytes = 0; // 大块内存分割过程中剩余的字节数
void* _freeList = nullptr; // 还回来的内存块用自由链接表的头指针链接起来

剩余内存字节数必须大于我们要申请的类型内存空间，回收内存块，我们只需要借助内存块头四个或八个字节（平台不同，指针字节数不同）储存下个内存块的地址。

链表链接内存块方式：数据结构头插思想

存储方式

*(void**)obj = _freeList;

程序结束申请的内存自动回收（内存池，我们不需要关心；回收问题）

实现思路：先从freeList（回收链表）中查找是否还有内存块

有就返回内存块地址，没有— 接下来从 _memory（这个大的内存块）中取；如果内存不足，再申请一大块内存。

2.4代码实现

.h文件

#pragma once
#include <iostream>
#include <vector>
#include <time.h>
using std::cout;
using std::endl;
#ifdef _WIN32
  #include<windows.h>
#else
#endif
// 直接去堆上按页申请空间
inline static void* SystemAlloc(size_t kpage)
{
#ifdef _WIN32
  void* ptr = VirtualAlloc(0, kpage << 13, MEM_COMMIT | MEM_RESERVE, PAGE_READWRITE);
#else
  // linux下brk mmap等
#endif
  if (ptr == nullptr)
    throw std::bad_alloc();
  return ptr;
}
template <class T>
class myObject
{
public: 
  T* New()
  {
    // 优先把还回来内存块对象，再次重复利用
    T* obj = nullptr;
    if (_freeList)
    {
      //  头删
      void* next= *(void**)_freeList;
      obj = (T*)_freeList;
      _freeList = next;
    }
    else
    {
      // 剩余内存不够一个对象大小时，则重新开大块空间
      if (_remainBytes < sizeof(T))
      {
        _remainBytes = 1024 * 128;
        //_memory = (char*)malloc(_remainBytes);
        _memory = (char*)SystemAlloc(_remainBytes >> 13); // 按页申请内存
        if (_memory == nullptr)
        {
          throw std::bad_alloc();
        }
      }
      obj = (T*)_memory;
      size_t objSize = sizeof(T) < sizeof(void*) ? sizeof(void*) : sizeof(T); //申请内存空间必须大于或等于一个指针的空间
      _memory += objSize;
      _remainBytes -= objSize;
    }
    // 定位new，显示调用T的构造函数初始化
    new(obj)T;
    return obj;
  }
  void Delete(T* obj)
  {
    // 显示调用析构函数清理对象
    obj->~T();
    //  头插
    *(void**)obj = _freeList; //  取前四或八个字节
    _freeList = obj;
  }
private:
  char* _memory = nullptr;  //  指向大块内存的指针
  size_t _remainBytes = 0;  //  大块内存分割过程中剩余的字节数
  void* _freeList = nullptr;  //  还回来的内存块用自由链接表的头指针链接起来
};

【文章福利】小编推荐自己的Linux内核技术交流群:【 865977150】整理了一些个人觉得比较好的学习书籍、视频资料共享在群文件里面，有需要的可以自行添加哦！！！

2.5效率（malloc与定长内存池）

测试代码：.cpp文件

struct TreeNode
{
  int _val;
  TreeNode* _left;
  TreeNode* _right;
  TreeNode()
    :_val(0)
    , _left(nullptr)
    , _right(nullptr)
  {}
};
void TestObjectPool()
{
  // 申请释放的轮次
  const size_t Rounds = 100;
  // 每轮申请释放多少次
  const size_t N = 1000000;
  std::vector<TreeNode*> v1;
  v1.reserve(N);
  size_t begin1 = clock();
  for (size_t j = 0; j < Rounds; ++j)
  {
    for (int i = 0; i < N; ++i)
    {
      v1.push_back(new TreeNode);
    }
    for (int i = 0; i < N; ++i)
    {
      delete v1[i];
    }
    v1.clear();
  }
  size_t end1 = clock();
  std::vector<TreeNode*> v2;
  v2.reserve(N);
  myObject<TreeNode> TNPool;
  size_t begin2 = clock();
  for (size_t j = 0; j < Rounds; ++j)
  {
    for (int i = 0; i < N; ++i)
    {
      v2.push_back(TNPool.New());
    }
    for (int i = 0; i < N; ++i)
    {
      TNPool.Delete(v2[i]);
    }
    v2.clear();
  }
  size_t end2 = clock();
  cout << "new cost time:" << end1 - begin1 << endl;
  cout << "object pool cost time:" << end2 - begin2 << endl;
}

测试结果图：

结论：明显内存池效率更快。

三、高并发内存池整体设计框架

现代很多的开发环境都是多核多线程，在申请内存的场景下，必然存在激烈的锁竞争问题。malloc本身其实已经很优秀，那么我们项目的原型tcmalloc就是在多线程高并发的场景下更胜一筹，所以这次我们实现的内存池需要解决以下几方面的问题。

1.性能问题。
2.多线程环境下，锁竞争问题。
3.内存碎片问题。

3.1三层设计思路

第一层：thread cache（线程缓存）：

每个线程独享一个thread cache，用于小于256k的内存分配情况，线程从这里申请内存不需要加锁（因为其他线程无法访问当前线程的 thread cache，没有竞争关系）

第二层：central cache（中心缓存）：

所有线程共享一个central cache，thread cache 是按需从central cache中获取对象，central cache在合适的时候会收回thread cache中的对象，避免一个线程占用太多资源。

central cache 是所有线程共享的，所以存在竞争关系，需要加锁；这里使用的锁是桶锁，并且因为只有thread cache没有内存时才会申请内存，所以这里竞争不会太激烈。

第三层：page cache（页缓存）：

页缓存存储的内存是以页为单位进行存储的及分配的。central cache没有内存时，则会从page cache中申请一定数量的page，并切割成定长大小的小内存块。

当一个span的几个跨度页的对象都回收后，page cache会回收central cache满足条件的span对象，并且合并相邻的页，组成更大的页，缓解内存碎片的问题。

3.2thread cache 设计思路（线程缓存）

thread cache是哈希桶结构，每个桶是一个按桶位置映射大小的内存块对象的自由链表。每个线程都会有一个thread cache对象，这样每个线程在这里获取对象和释放对象时是无锁的。

线程申请内存：

线程申请内存时，会有不同规格的内存申请（4字节、5字节等），根据范围划定不同的自由链表，设计多个自由链表管理不同规格的内存小块。实质就相当于使用多个定长内存池的自由链表，每个内存小块采用向上对齐原则（可能会出现多申请内存的情况，这就是内碎片）

例如：

需要9字节，则开辟一个大小为2个8字节的空间的节点
需要100字节，则开辟一个大小为13个8字节的空间的节点。

线程对齐规则：

整体控制在最多10%左右的内碎片浪费，总计设计208个桶，[0,15]个桶，每个桶的对齐数相差8字节（最高128字节对齐），[16,71]个桶，每个桶的对齐数相差16字节（最高1024字节对齐）以此类推。

注意：_freeLists是一个数组，每个元素都是自由链表类型（即存储自由链表的头结点）

线程释放内存

释放内存后：采用自由链表的结构来管理切好的小块内存（每一个切分好的小块内存就是一个节点）
具体方法是：用切分好的小块内存的前4字节或8字节来存储下一个小块内存的地址。
插入节点时，采用头插的方式。

thread cache代码框架

本质：就相当于一个数组，每个数组挂个桶；而每个桶就类似于上面那个定长内存池一样。

3.3线程缓存无锁设计（TLS）

（ConcurrentAlloc.h文件）

#pragma once
#include "Common.h"
#include "ThreadCache.h"
static void* ConcurrentAlloc(size_t size)
{
  // 通过TLS 每个线程无锁的获取自己的专属的ThreadCache对象
  if (pTLSThreadCache == nullptr)
  {
    pTLSThreadCache = new ThreadCache;
  }
  cout << std::this_thread::get_id() << ":" << pTLSThreadCache << endl;
  return pTLSThreadCache->Allocate(size);
}
static void ConcurrentFree(void* ptr, size_t size)
{
  assert(pTLSThreadCache);
  pTLSThreadCache->Deallocate(ptr, size);
}

TLS：thread local storage 线程本地存储（linux和Windows下有各自的TLS）
TLS是一种变量的存储方式，这个变量在它所在的线程内是全局可访问的，但是不能被其他线程访问，这样就保证了线程的独立性。
静态TLS使用方法：
_declspec(thread) DWORD data=0;
声明了一个 _declspec(thread) 类型的变量，会为每一个线程创建一个单独的拷贝。

原理：

在x86 CPU上，将为每次引用的静态TLS变量生成3个辅助机器指令
如果在进程中创建子线程，那么系统将会捕获它并且自动分配一另一个内存块，以便存放新线程的静态TLS变量。

//每个线程刚创建时，就会获得这个指针，每个线程的指针是不同的。
static _declspec(thread) ThreadCache* pTLSThreadCache = nullptr;
//static 修饰，可以保证变量只在当前文件可见。

简单测试一下：

void Alloc1()
{
  for (size_t i = 0; i < 5; ++i)
  {
    void* ptr = ConcurrentAlloc(16);
  }
}
void Alloc2()
{
  for (size_t i = 0; i < 5; ++i)
  {
    void* ptr = ConcurrentAlloc(17);
  }
}
void TLSTest()
{
  std::thread t1(Alloc1);
  t1.join();
  std::thread t2(Alloc2);
  t2.join();
}
int main()
{
  //TestObjectPool();
  TLSTest();
  return 0;
}

3.4central cache设计思路（中心缓存）

central cache也是一个哈希桶结构，他的哈希桶的映射关系跟thread cache是一样的。不同的是他的每个哈希桶位置挂是SpanList链表结构，不过每个映射桶下面的span中的大内存块被按映射关系切成了一个个小内存块对象挂在span的自由链表中。

注意：

span是双向链表，而span下挂载的小块内存对象是单链表。
中心缓存需要加桶锁。
_spanLists 是一个数组，数组中每个元素都是一个span自由链表的_head头指针。
每个span又是一个单向自由链表。

内存申请：

当thread cache中没有内存时，就会批量向central cache申请一些内存对象，从对应的span中取出小块内存对象给thread cache，这个过程是需要加锁的（加桶锁）
这里批量获取对象的数量使用了类似网络tcp协议拥塞控制的慢开始算法。
如果所有的span都为空了（即central cache使用完了），则将空的span链在一起，向page cache申请一个span对象，span对象中是一些以页为单位的内存，需要切成对应的小块内存对象，并链接起来，挂到span中。
central cache中每一个span都有一个use_count，分配一个对象给thread cache，就加加。

内存释放：

当thread_cache将内存释放回central cache中的时，释放回来就减减use_count。
当use_count减到0时则表示所有对象都回到了span，则将span释放回page cache，page cache中会对前后相邻的空闲页进行合并。

CentralCache 代码框架：

注意：
CentrealCache 是多个线程都可以进行访问的。我们设计时需要考虑下面两点
单例模式 2. 临界资源加锁

（CentralCache.h文件）

#pragma once
#include "common.h"
//  单例模式---饿汉模式
class CentralCache
{
public:
  static CentralCache* GetInstance()
  {
    return &_sInst;
  }
  // 获取一个非空的span
  Span* GetOneSpan(SpanList& list, size_t byte_size);   //  list: 表示_spanLists中span的_freeList连续挂接空间的数量
  // 从中心缓存获取一定数量的对象给thread cache
  size_t FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size); //  size：内存空间大小；batchNum：内存空间个数
  //  start 和 end 表示内存空间地址起始到终末范围
private:
  SpanList _spanLists[NFREELIST];
private:
  CentralCache() {};
  CentralCache(const CentralCache& t) = delete;
  static CentralCache _sInst; //  声明
};

GetOneSpan函数：需要到PageCache模块去讲解（CentralCache.cpp文件）

CentralCache CentralCache::_sInst;  //定义
// 获取一个非空的span
Span* CentralCache::GetOneSpan(SpanList& list, size_t byte_size)
{
  //  ...
  return nullptr;
}
// 从中心缓存获取一定数量的对象给thread cache
size_t CentralCache::FetchRangeObj(void*& start, void*& end, size_t batchNum, size_t size)
{
  //  每个桶加锁
  size_t index = SizeClass::Index(size);
  _spanLists[index]._mtx.lock();
  Span* span = GetOneSpan(_spanLists[index], size);
  assert(span);
  assert(span->_freeList);
  // 从span中获取batchNum个对象
  // 如果不够batchNum个，有多少拿多少
  start = span->_freeList;
  end = start;
  size_t i = 0, actualNum = 1;
  while (i < batchNum - 1 && NextObj(end) != nullptr)   //  取连续内存块，当batchNum超过原有时，按拥有的数量给
  {
    end = NextObj(end);
    ++i;
    ++actualNum;
  }
  span->_freeList = NextObj(end);   // 剩余空间仍挂在Span中的_freeList去
  NextObj(end) = nullptr;       //  最后end记录下一个空间地址置空
  span->_useCount += actualNum;
  _spanLists[index]._mtx.unlock();  //  解锁
  return actualNum;
}

以页为单位的大内存管理span的定义及spanlist定义（common.h文件）

// Span管理一个跨度的大块内存
// 管理以页为单位的大块内存
struct Span
{
  PAGE_ID _pageId = 0;  //  页号
  size_t n = 0;     //  页数
  //  双向链表的结构
  Span* next = nullptr;
  Span* prev = nullptr;
  size_t _useCount = 0;   //  使用计数，切好的小块内存分配给ThreadCache
  void* _freeList = nullptr;  // 大块内存切小链接起来，这样回收回来的内存也方便链接
};
// 带头双向循环链表
class SpanList
{
public:
  SpanList()
  {
    _head = new Span;
    _head->next = _head;
    _head->prev = _head;
  }
  Span* Begin()
  {
    return _head->next;
  }
  Span* End()
  {
    return _head;
  }
  void Insert(Span* pos, Span* newSpan)
  {
    //  头插
    Span* prev = pos->prev;
    prev->next = newSpan;
    newSpan->prev = prev;
    newSpan->next = pos;
    pos->prev = newSpan;
  }
  void Erase(Span* pos)
  {
    assert(pos != _head);
    Span* prev = pos->prev;
    Span* next = pos->next;
    prev->next = next;
    next->prev = prev;
  }
private:
  Span* _head;
public:
  std::mutex _mtx;  //  桶锁: 数组_spanLists中每一个位置的桶都有一把锁
};

慢开始反馈调节算法

补全hreadCache.h中FetchFromCentralCache函数只声明未实现

完善common.h文件

1）FreeList中成员变量和函数：

void PushRange(void* start, void* end)
  {
    //  一连串空间存储
    NextObj(end) = _freeList;
    _freeList = start;
  }

MaxSize函数调节页数：

2）类SizeClass的静态成员函数

// 一次从中心缓存获取多少个
  static size_t NumMoveSize(size_t size)
  {
    assert(size > 0);
    // [2, 512]，一次批量移动多少个对象的(慢启动)上限值
    // 小对象一次批量上限高
    // 小对象一次批量上限低
    int num = MAX_BYTES / size;
    if (num < 2)
      num = 2;
    if (num > 512)
      num = 512;
    return num;
  }

（ThreadCache.cpp文件）

// 慢开始反馈调节算法
// 1、最开始不会一次向central cache一次批量要太多，因为要太多了可能用不完
// 2、如果你不要这个size大小内存需求，那么batchNum就会不断增长，直到上限
// 3、size越大，一次向central cache要的batchNum就越小
// 4、size越小，一次向central cache要的batchNum就越大
void* ThreadCache::FetchFromCentralCache(size_t index, size_t size)
{
  //  慢开始反馈调节法    不会开始要太多内存
  size_t  batchNum = std::min(_freeLists[index].MaxSize(), SizeClass::NumMoveSize(size));
  if (batchNum == _freeLists[index].MaxSize())
  {
    ++_freeLists[index].MaxSize();
  }
  void* start = nullptr;
  void* end = nullptr;
  size_t actualNum = CentralCache::GetInstance()->FetchRangeObj(start, end, batchNum, size);  //  获取连续实际内存的数目
  assert(actualNum >= 1); //  至少一个
  if (actualNum == 1)
  {
    assert(start == end);
  }
  else
  {
    _freeLists[index].PushRange(NextObj(start), end);
  }
  return start;
}

3.5page cache设计思路（页缓存）

page cache中也是哈希桶结构，但是每个节点存储的都是span，因为页缓存是所有线程共享的，只需要定义一个对象，所以这里将 page cache 设计为单例模式（这里采用饿汉模式的设计方法）。

注意：page cache需要加整体锁（因为是所有线程共享的）

申请内存：

当central cache向page cache申请内存时，page cache先检查对应位置有没有span，如果没有则向更大页寻找一个span，如果找到则分裂成两个。
例如：申请的是4page，4page后面没有挂span，则向后面寻找更大的span，假设在10page位置找到一个span，则将10page span分裂为一个4page span和一个6page span，把4page的span返回给central cache，把6page的span挂载到对应的6号桶中去。
如果找到128 page都没有合适的span，则向系统使用mmap、brk或者是VirtualAlloc等方式申请一个128page 的span挂在自由链表中，再重复1中的过程。
第一次申请内存时，page cache是空的，经过上面的过程后会向堆申请一个128page的span，然后重复上面的过程，将128page的span进行分裂和重新挂载。

释放内存：

如果central cache释放回一个span，则依次寻找span的前后page id的span，看是否可以合并，如果合并继续向前寻找。这样就可以将切小的内存合并收缩成大的span，减少内存碎片。

PageCache 代码框架：

page cache是一个以页为单位的span自由链表
为了保证全局只有唯一的page cache，这个类被设计成了单例模式。
注意线程安全，需要锁（此处我们采用整体锁）

高并发内存池实战：用C++构建高性能服务器（上）