【高并发内存池】从零到一的项目:项目介绍、内存池及定长内存池的设计

个人主页 ： zxctscl
如有转载请先通知

这个项目是我在看了网上tcmalloc源代码，并且在网上查找了一些相关资料后，在学习后实现的一个缩小版本。

1. 项目介绍

1.1 这个项目做的是什么？

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

这个项目是把tcmalloc最核心的框架简化后拿出来，模拟实现出一个自己的高并发内存池，目的就是学习tcamlloc的精华，这种方式有点类似之前学习STL容器的方式。但是相比STL容器部分，tcmalloc的代码量和复杂度上升了很多，大家要有心理准备。当前另一方面，难度的上升，我们的收获和成长也是在这个过程中同步上升。

另一方面tcmalloc是全球大厂google开源的，可以认为当时顶尖的C++高手写出来的，他的知名度也是非常高的，不少公司都在用它，Go语言直接用它做了自己内存分配器。所以很多程序员是熟悉这个项目的，那么有好处，也有坏处。好处就是把这个项目理解扎实了，会很受面试官的认可。坏处就是面试官可能也比较熟悉项目，对项目会问得比较深，比较细。如果你对项目掌握得不扎实，那么就容易碰钉子。
tcmalloc源代码

1.2 这个项目的要求的知识储备和难度

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

2. 内存池

2.1 池化技术

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

在计算机中，有很多使用“池”这种技术的地方，除了内存池，还有连接池、线程池、对象池等。以服务器上的线程池为例，它的主要思想是：先启动若干数量的线程，让它们处于睡眠状态，当接收到客户端的请求时，唤醒池中某个睡眠的线程，让它来处理客户端的请求，当处理完这个请求，线程又进入睡眠状态。

2.2 内存池

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

2.3 内存池主要解决的问题

内存池主要解决的当然是效率的问题，其次如果作为系统的内存分配器的角度，还需要解决一下内存碎片的问题。那么什么是内存碎片呢？
最开始list申请空间，然后mysql、map、vector，就使用了这块空间。但是释放的时候并没有规定谁先释放，是比较自由的，谁不用了，谁就先释放。list对象销毁了，它占那块空间就是放了；vector也释放了，他们现在就总的空出384byte空间。假设申请300byte空间，却申请不出来。虽然是有足够的空间，但是不连续，是碎片化的，就导致申请不了这300byte的空间。这个也就是内存碎片问题。

再需要补充说明的是内存碎片分为外碎片和内碎片，上面我们讲的外碎片问题。**外部碎片是一些空闲的连续内存区域太小，这些内存空间不连续，以至于合计的内存足够，但是不能满足一些的内存分配申请需求。内部碎片是由于一些对齐的需求，导致分配出去的空间中一些内存无法被利用。**内碎片问题，我们后面项目就会看到，那会再进行更准确的理解。

2.4 malloc

C/C++中我们要动态申请内存都是通过malloc去申请内存，但是我们要知道，实际我们不是直接去堆获取内存的，而malloc就是一个内存池。**malloc() 相当于向操作系统“批发”了一块较大的内存空间，然后“零售”给程序用。**当全部“售完”或程序有大量的内存需求时，再根据实际需求向操作系统“进货”。malloc的实现方式有很多种，一般不同编译器平台用的都是不同的。比如windows的vs系列用的微软自己写的一套，linux gcc用的glibc中的ptmalloc。下面有几篇关于这块的文章，大概可以去简单看看了解一下，关于ptmalloc，学完这个的项目以后，有兴趣大家可以去看看他的实现细节。

一文了解，Linux内存管理，malloc、free 实现原理
malloc()背后的实现原理——内存池
malloc的底层实现（ptmalloc）

3. 设计一个定长的内存池

作为程序员(C/C++)我们知道申请内存使用的是malloc，malloc其实就是一个通用的大众货，什么场景下都可以用，**但是什么场景下都可以用就意味着什么场景下都不会有很高的性能，**下面我们就先来设计一个定长内存池做个开胃菜，当然这个定长内存池在我们后面的高并发内存池中也是有价值的，所以学习他目的有两层，先熟悉一下简单内存池是如何控制的，第二他会作为我们后面内存池的一个基础组件。

第一需要申请一的块内存，第二内存不仅仅有申请还有释放，释放出来内存池怎么办呢，怎么管理？

这个定长内存池，它解决固定大小的内存申请释放需求
特点：

1. 性能达到极致
2. 不考虑内碎片问题

3.1 设计

为什么成员变量要用char*，不用void*？

指针就是一个地址，它前面的类型表示这个指针指向的类型，而void*是不能解引用，不能加加的。
这里在申请的一大块内存中加10不好加，得强制类型转换，所以直接用char*，方便去切内存块，像切3字节、2字节这些。

那么切了的内存块释放回来后怎么管理呢？
这是就用到了自由链表，用链式结构把它们管理起来。此时他们还回来时候顺序是混乱的，也不需要加加减减，就定义为 void*_freeList，它去存内存块的地址。这些内存块就相当于一个个节点，在32位平台下，只需要把下一个释放的内存地址存入到上一个释放的内存中就行。

它成员变量就是

	char* _memory=nullptr;// 指向大块内存的指针
	void* _freeList=nullptr;// 还回来过程中链接的自由链表的头指针

如果没有大块内存，那就先申请。_memory往后走多少，就是那么切一个大小T的空间出来。切出来空间就用T* obj表示。实际上它地址还是连续的，这只想象上的说法。

最基本一个操作：

	T* New()
	{
		if (_memory == nullptr)
		{
			_memory = (char*)malloc(128 * 1024);
			if (_memory == nullptr)
			{
				throw bad_alloc();
			}
		}
		T* obj = (T*)_memory;
		_memory += sizeof(T);
		return obj;
	}

如果_memory开的128字节空间开完了，obj就越界。
此时就得加一个成员来记录大块空间剩下的字节数：size_t _remainBytes 。
成员变量就是：

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

如果不能完整切除_memory，不够一个T对象用，就会_memory导致剩下空间。
此时如果_remainBytes < sizeof(T)，就重新申请。

T* New()
	{
	// 剩余内存不够一个对象大小时，则重新开大块空间
		if(_remainBytes < sizeof(T))
		{
		   _remainBytes = 128 * 1024;
		   _memory = (char*)malloc( _remainBytes);
			if (_memory == nullptr)
			{
				throw bad_alloc();
			}
		}
		T* obj = (T*)_memory;
		_memory += sizeof(T);
		_remainBytes -= sizeof(T);
		
		return obj;	
	}

还回来的对象怎么处理？
可能是空

如果给了对象：

如果想让头上4字节或者8字节，那么怎么处理？
虚函数表的指针就是4字节或者8字节

把obj类型强转成int*，现在对它进行解引用，一个int解引用就是4个字节。解引用一个int可以读这里的值，还可以对这个值进行写。

		if (_freeList == nullptr)
		{
			_freeList = obj;
			*(int*)obj = nullptr;
		}

但是在32位下，上面这个程序是没有问题的。
可64位下，上面程序是跑不动的，此时指针是8个字节。
怎么判断现在执行程序的平台是32还是64呢？
把obj强转为void**，再解引用：*(void**)obj =nullptr;

void**，解引用时候void*在32位平台下是4字节，在64位平台下是8字节。

单链表找尾比较麻烦，所以就直接头插。

			//头插
			*(void**)obj = _freeList;
			_freeList= obj;

	void Delete(T* obj)
	{
		if (_freeList == nullptr)
		{
			_freeList = obj;
		 *(void**)obj =nullptr;
			/**(int*)obj = nullptr;*/
		}
		else
		{
			//头插
			*(void**)obj = _freeList;
			_freeList= obj;
		}	
	}

没有节点和有节点，此时头插是完全一样的，并不需要区分，就直接是：

	void Delete(T* obj)
	{
		// 头插
		*(void**)obj = _freeList;
		_freeList = obj;
	}

当需要再切内存块的时候，一定要从_memory切吗？
并不是，那些还回来的内存块，还能重复利用。
所以在new的时候，还考虑_freeList，用了，就是_freeList的头删。

		// 优先把还回来内存块对象，再次重复利用
		if (_freeList)
		{
			void* next = *((void**)_freeList);
			obj = (T*)_freeList;
			_freeList = next;
			return obj;
		}

如果T大小小于一个指针，objSize大小就就是sizeof(void*)，否则就是sizeof(T)，这样就至少能存下下一个释放内存块的地址。

size_t objSize = sizeof(T) < sizeof(void*) ? sizeof(void*) : sizeof(T);

可以对已经有的空间进行初始化，调用构造函数初始化：

		// 定位new，显示调用T的构造函数初始化
		new(obj)T;

同样在Delete中，调用析构函数：

	void Delete(T* obj)
	{
		// 显示调用析构函数清理对象
		obj->~T();

		// 头插
		*(void**)obj = _freeList;
		_freeList = obj;
	}

windows和Linux下如何直接向堆申请页为单位的大块内存：
VirtualAlloc

brk和mmap

我们代码在windows下：

#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;
}

3.2 代码实现

#include <iostream>
#include <vector>
#include <time.h>
using std::cout;
using std::endl;

#ifdef _WIN32
#include<windows.h>
#else
// 
#endif

// 定长内存池
//template<size_t N>
//class ObjectPool
//{};

// 直接去堆上按页申请空间
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 ObjectPool
{
public:
	T* New()
	{
		T* obj = nullptr;

		// 优先把还回来内存块对象，再次重复利用
		if (_freeList)
		{
			void* next = *((void**)_freeList);
			obj = (T*)_freeList;
			_freeList = next;
		}
		else
		{
			// 剩余内存不够一个对象大小时，则重新开大块空间
			if (_remainBytes < sizeof(T))
			{
				_remainBytes = 128 * 1024;
				//_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; // 还回来过程中链接的自由链表的头指针
};

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

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

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

	// 每轮申请释放多少次
	const size_t N = 100000;

	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);

	ObjectPool<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;
}

有问题请指出，大家一起进步！！！