C++ 环形缓冲区（队列）简单实现

1. 说明

在实际工作中，如果数据流量过大，可以先把数据接收到数据缓冲区中，处理之后再取出。我们定义的包协议可以采用定长包，可以采用不定长度的包，环形缓冲区都能处理。

2. 使用场景

2.1 生产消费的场所

在 “生产者--消费者”模式中，往往会用到这种环形缓冲区的功能。

2.2 线程间高速通信的场所

3. 使用的局限性

主要适用于单向传输。不太适用于双向传输的情形。

4. 代码展示

CircleBuffer.h

#pragma once
# include	<stdint.h>
# include	<mutex>
/**
 * 数据包头，每次存往环形队列的数据长度可能不一致，所以每次先存一个EleHeader用于指示长度，然后再存实际的数据。
 */
struct EleHeader 
{
	int payloadLen;//实际消息的长度
};

/**
 * 此环形队列不考虑多线程问题，默认一进(putData)一出(getData)
 */

class CircleBuffer
{
public:
	CircleBuffer(uint32_t bufSize);
	~CircleBuffer();

private:
	char	*m_buf;           //指向环形缓冲区的指针
	int		m_readPos;        //缓冲区读位置索引
	int		m_writePos;       //缓冲区写位置索引
	int     m_readableSize;   //缓存区中可读的字节数（已经写但是还未读的字节数,取值范围为[0,m_maxBufSize]）
	int		m_maxBufSize;     //缓冲区的最大值(根据业务不同，设置合适的值)
	int		m_packHeadLength; //数据包头长度(如果是定长包，那么这里可以是整个数据包。如果是不
							  //定长的，那么这里就是包头的长度)
	std::mutex   m_mutex;
public:
	/**
	 * 把接收到的数据拷贝到缓冲区中
	 * 返回值: -1:出错。 >=0:实际写入的字节数。
	 */
	int putData(char *data, int len);

	/**
	* 从缓冲区中获取指定长度的数据，存到buf中
	* 返回值: -1:出错。 >=0:实际获取到的字节数。
	*/
	int getData(char* buf, int len);

	//返回缓存区中的可读字节数量
	int getBufReadSize();

	//重置缓冲区    
	void	resetBuffer();   

	//缓冲区是否为空
	bool Empty();
};

CircleBuffer.cpp

#include "CircleBuffer.h"
#include <assert.h>

//源文件实现
CircleBuffer::CircleBuffer(uint32_t bufSize)
{
	m_writePos = 0;
	m_readPos = 0;
	m_readableSize = 0;
	m_maxBufSize = bufSize;
	m_buf = new char[m_maxBufSize];
	m_packHeadLength = sizeof(EleHeader);//EleHeader为"你自己定义的数据包头"。
}

CircleBuffer::~CircleBuffer()
{
	if (m_buf)
	{
		delete[] m_buf;
		m_buf = NULL;
	}
}


bool CircleBuffer::Empty()
{
	return m_readableSize == 0;
}

int CircleBuffer::putData(char *data, int len)
{
	if (len <= 0)
	{
		return 0;
	}
	//判断队列的空闲区域大小
	assert(m_readableSize >= 0 && m_readableSize <= m_maxBufSize);
	m_mutex.lock();
	int resLen = m_maxBufSize - m_readableSize;//可写的总字节数
	
	if (len > resLen)
	{
		len = resLen;
	}

	//按字节拷贝
	for (int i = 0; i < len; i++)
	{
		m_buf[m_writePos] = data[i];
		//注意这里不能用rear++,既然是环形缓冲区，如果填充到底部之后，底部的索引要回到头部
		m_writePos = (m_writePos + 1) % m_maxBufSize;
	}
	m_readableSize += len;	//可读字节数 + len
	m_mutex.unlock();
	return len;
}

int CircleBuffer::getData(char* buf, int len)
{
	if (len <= 0)
	{
		return 0;
	}
	assert(m_readableSize >= 0);
	m_mutex.lock();
	if (m_readableSize < len)
	{
		len = m_readableSize;
	}
	
	int temp = m_readPos;//获取数据的时候不能改变头部索引的值，全部处理完成之后才改变头部索引的位置
	for (int i = 0; i < len; i++)
	{
		buf[i] = m_buf[temp];
		//这里和上面一样，头部的索引也有可能跑到底部，得让它能跑回来
		temp = (temp + 1) % m_maxBufSize;
	}
	m_readPos = (m_readPos + len) % m_maxBufSize;
	m_readableSize -= len;	//可读字节数 - len
	m_mutex.unlock();
	return len;
}


//返回缓存区中的可读字节数量
int CircleBuffer::getBufReadSize()
{
	return m_readableSize;
}

//提供给使用者的接口
//重置缓冲区
void CircleBuffer::resetBuffer()
{
	m_mutex.lock();
	m_writePos = 0;
	m_readPos = 0;
	m_readableSize = 0;
	m_mutex.unlock();
}

使用测试：

#include "CircleBuffer.h"

void main()
{
	CircleBuffer container(1024*11);
	char	buf[512] = { 0 };
	EleHeader eHdr; 
	int ret = 0;
	for (int i = 0; i < 20; i++)
	{
		eHdr.payloadLen = sizeof(buf) - i;
		ret = container.putData((char*)&eHdr, sizeof(EleHeader));
		printf("%d, write size:%d\n", i+1, ret);
		ret = container.putData(buf, eHdr.payloadLen);
		printf("%d, write size:%d\n", i + 1, ret);
	}

	char readBuf[1024] = {};
	int payloadLen = 0;
	int count = 0;
	while (true)
	{
		int size = container.getBufReadSize();
		printf("can read size:%d\n", size);
		if (size <= 0)
		{
			break;
		}
		ret = container.getData(readBuf, sizeof(EleHeader));
		payloadLen = ((EleHeader*)readBuf)->payloadLen;
		printf("read EleHeader ele size:%d\n", payloadLen);

		ret = container.getData(readBuf, payloadLen);
		if (ret != payloadLen)
		{
			printf("read EleHeader payloadLen failed\n");
			break;
		}
		else
			printf("read EleHeader payloadLen ok,count:%d\n", (count++) + 1);

	}

	getchar();
	return ;
}