高并发内存池 · 基本认识

前言：

本文呢开始搞搞项目咯，于是准备从一个最经典的项目入手--tcmalloc，也就是从谷歌开源出来的一个高并发内存池项目，要说这个项目有多牛*呢，就这么说吧，GO语言是直接将其作为了自己的内存的回收机制。

但是因为该项目来源于谷歌，并且开源，也就代表了当时顶尖的C++高手是去写的，那么难度可想而知，所以我们这里呢，就实现一个min版本的，了解一下核心，就和当时我们学习stl一样。

以上摘自文心一言对于该项目的简单介绍：

TCMalloc是谷歌开源的高效内存分配器，专为C和C++设计，旨在提供比标准内存分配器更好的性能，特别是在多线程环境中。它通过线程缓存减少锁竞争，优化内存利用率，并支持高并发。TCMalloc可以作为标准内存分配器的替代品，在编译时链接到应用程序中，适用于需要高效内存管理的大规模并行应用。

项目基础认识

做一个项目嘛，我们肯定要知道为什么这么干，比如这个项目，为什么我们需要额外实现一个内存池？

首先，这个项目肯定是能在多线程调用的情况下提高内存分配效率的，那么为什么呢？是因为什么导致了内存分配效率的下降呢？

内存碎片

不知道大家有没有听说过内存碎片的问题：

在进程的地址空间中，中间有一块专门的区域用于实现内存分配，也就是我们平常说的操作系统这个科目中的堆区。那么因为计算机读取数据的时候都是一整个一整个读取的。

比如，在结构体的内存对齐中，我们会探究为什么需要内存对齐，因为计算机读取数据都是4kb或者8kb的读取，这就代表了如果我们没有内存对齐的话，大多数时候计算机需要多次读取一份数据才能拿到完整的数据。

这里也是一样的，当我们申请一块连续的内存的时候，即便堆区中的存储容量是够的，但是因为是碎片化的，那么OS也就没有办法一次性给我们，也就代表了这次内存分配是失败的。

这就是内存碎片问题。

效率问题

同样，不止还有内存碎片问题，还有的问题是效率问题，这非常好理解。

咱们拿大学生举例，大学生基本上都是每个月给固定的生活费，为什么呢~如果我要用的时候再给不行吗？当然是可以的，不过你每次申请生活费的时候，相当于每次都要和你的家人交互，那么交互的这个过程，你是空闲的，你的家人不一定，得等两个人都空暇了，才能成功申请。

所以这里的效率问题实际上就是为了减少交互的过程。

实际上OS对于malloc已经采取了这个措施，对于malloc来说本身就是一个内存池，和我们刚才所述的拿到一大块内存然后分配是一样的，对于这个过程来说有非常多的实现方式。那么这里我们可以脱离malloc实现一个内存池，但是毕竟咱们是新手，可以先用着，后面咱们也可以自己实现malloc再套接上来。也可以直接去堆上按页申请空间。

那么以上是对于内存池的基本认识——内存碎片，效率问题。

定长内存池

好了，到这里你该不会要以为咱们就要开始编写内存池了吧？

Nonono，当然不会，我们不妨拿一个开胃小菜试试水，比如定长内存池~对于该内存池我们在tcmalloc项目里面也会使用到，那么试试咯~

其实对于定长内存池来说，我们无非要考虑的点就是，怎么切内存，怎么回收内存，如何管理回收的内存，对于这三个点搞清楚这个定长内存池就算解决咯~

切内存

切内存我们要考虑一下几个点：

给谁切？怎么切？

1.对于这个点，我们不妨使用一个模板，某个对象来获取内存的时候，给这个对象一个大空间，后续该对象要使用的时候就从该内存池里获取，那么模板肯定跑不掉。

2.假定我们现在有一个空间，我们回想一下在最开始学习指针的时候，假如我们有一个int，但是我们使用char*去操控它，我们就会发现如果让char*移动的话，每次移动是一个字节，也就代表了指针的移动和指针指向的类型有关，所以如果让我们来操控一块空间，为了更好的操作，我们肯定希望能按单位字节操作，所以我们希望使用char*的指针来指向这块空间，那么一个成员变量是不是就确定了？需要一个char*的变量吧。

3.假定我们已经分配到最后了，如果T是int类型，此时空间还有5字节，那分配给了该int之后，还有一个字节，此时还能分配吗？当然不能，我们需要重新开辟空间，那么判断切内存的条件就不能是判断char*的指针怎么怎么样了，如果是通过char*的指针判断，判断空？判断越界？好像都没有太好操作。所以我们不妨引入一个成员变量判断是否需要扩容，该成员变量的作用就是用来记录剩余字节的个数的。

基本框架就有了：

template <class T>
class memoryPool
{
private:
    char* _memory;
    size_t _remain;
};

好了，好像我们就能写一个切内存的基本逻辑了，哦~char*的移动应该不需要多说吧！

	T* new_mem()
	{
		T* obj = nullptr;
		// 剩余空间不够对象大小额外开空间
		if (_remain < sizeof(T))
		{
			_remain = MEMORY_SIZE;
			_memory = (char*)malloc(MEMORY_SIZE);
			if (_memory == nullptr)
			{
				throw std::bad_alloc();
			}
		}
		obj = reinterpret_cast<T*>(_memory);
		// 判断是否满足指向操作
		size_t _size = sizeof(T) < sizeof(void*) ? sizeof(void*) : sizeof(T);
		_memory += sizeof(_size);
		_remain -= _size;
	}
	// 定位new
	new(obj)T;
	return obj;

当C++中的new，底层是抛异常 + malloc + 初始化，所以我们不妨也调用一下对应的构造，如果申请失败了也抛异常，虽然我们这里没有捕获。

到这里肯定是有人发现那个三目操作符是没有讲解的，这因为涉及到了内存管理。下一阶段~

怎么管理回收内存？

你看奥，如果我们把内存分配出去了，那么直接free？还是还给我们刚才申请的一大块内存?

第一个，直接free，对于malloc的回收的机制是按照当前字节回收后面所有的连续的malloc回收的机制，那你这，，似乎非常不合理吧？

第二个，哪里借的就还到哪里去，直接还给申请的内存？那我问你，一个100，按顺序切出去了23份，但是还回来的时候七零八落的，我的char*指针，该何去何从呢？所以第二个方案显然不行。

所以我们引入了第三个成员变量，_freelist，用来管理回收内存的。

template <class T>
class memoryPool
{
private:
    char* _memory;
    size_t _remain;
    void* _freelist;
};

那么对于还回来的内存块我们是建议使用链表的方式，一个一个的串起来，那么怎么串呢？我们难道还需要写什么节点，整一个指针？显然不用。

如果该内存块的前4个或者8个字节能够存放指针，那么我们不就能获取到地址，就能够直接管理了吗？可是万一我们的T是一个int怎么办，如果是32位环境还好，64位环境的话，指针是8个字节。

所以我们现在需要想一个办法，解决两个问题，第一个是如何让分配出去的内存块一定可以存放指针，第二个是如何判断当前环境是32位还是64位环境。

第一个问题我们实际上已经解决了，在上面子问题的时候，如果分配的对象小于指针的大小，那么就分配指针大小的内存块，所以这个问题已经解决了，第二个问题如何解决呢~

我们有一个非常非常简单粗暴的解决方案，直接通过sizeof(void*)判断指针的大小是4还是8然后做出相应的判断即可。

但是这样是不是太挫了一点？我们需要在分配的对象中找一块空间足以存放指针，那么对于obj指向的内存块，我们”升级“为二级指针再解引用，就完美的得到了一个指针空间：

	void del_mem(T* obj)
	{
		obj->~T();
		// 判断32位系统还是64位系统
		*(reinterpret_cast<void**>(obj)) = _freelist;
		_freelist = obj;
	}

然后这里是一个链表的头插，这里不明显，在这里会更明显，即：

	T* new_mem()
	{
		T* obj = nullptr;
		// 优先使用还回来的内存
		if (_freelist)
		{
			void* next = *(static_cast<void**>(_freelist));
			obj = static_cast<T*>(_freelist);
			_freelist = next;
		}
		else
		{
			// 剩余空间不够对象大小额外开空间
			if (_remain < sizeof(T))
			{
				_remain = MEMORY_SIZE;
				_memory = (char*)malloc(MEMORY_SIZE);
				if (_memory == nullptr)
				{
					throw std::bad_alloc();
				}
			}
			//T* obj = reinterpret_cast<T*>(_memory);
			obj = reinterpret_cast<T*>(_memory);
			// 判断是否满足指向操作
			size_t _size = sizeof(T) < sizeof(void*) ? sizeof(void*) : sizeof(T);
			_memory += sizeof(_size);
			_remain -= _size;
		}
		// 定位new
		new(obj)T;
		return obj;
	}

有现成的内存为什么不使用呢~对吧！

来一个测试用例，大家下来自己测试：

struct TreeNode
{
	int _val;
	TreeNode* _left;
	TreeNode* _right;

	TreeNode()
		:_val(0)
		, _left(nullptr)
		, _right(nullptr)
	{
	}
};

void TestMemoryPool()
{
	// 申请释放的轮次
	const size_t Rounds = 5;

	// 每轮申请释放多少次
	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);
	memoryPool<TreeNode> mmPool;

	size_t begin2 = clock();
	for (size_t j = 0; j < Rounds; ++j)
	{
		for (int i = 0; i < N; ++i)
		{
			//if (i == N - 3) // 第一个Bug -> 定位new的时候出现
			//	std::cout << "test";

			v2.push_back(mmPool.new_mem());
		}
		for (int i = 0; i < N; ++i)
		{
			mmPool.del_mem(v2[i]);
		}
		v2.clear();
	}
	size_t end2 = clock();

	std::cout << "new cost time:" << end1 - begin1 << std::endl;
	std::cout << "object pool cost time:" << end2 - begin2 << std::endl;
}

感谢阅读！