Qt实现环形缓冲区的两种方法

一个环形buffer，在尾部追加数据，从头部读取数据，适合用作IO的缓冲区。

详细介绍可参考：https://en.wikipedia.org/wiki/Circular_buffer

一.使用QList和QByteArray

这个方法参考的是Qt源码中的QRingBuffer类，这个类不是Qt API的一部分，所以Qt助手里是查不到的，它的存在只是为了服务其他的源码。

QRingBuffer的源文件在D:\Qt\Qt5.7.0\5.7\Src\qtbase\src\corelib\tools目录中，由qringbuffer_p.h和qringbuffer.cpp实现。

QRingBuffer实现的环形缓冲区大概如下图所示。

Qt实现环形缓冲区的两种方法

qringbuffer.h

#ifndef QRINGBUFFER_P_H
#define QRINGBUFFER_P_H
#include <QByteArray>
#include <QList>
#ifndef QRINGBUFFER_CHUNKSIZE
#define QRINGBUFFER_CHUNKSIZE 4096
#endif
enum
{
//1G-1字节
MaxAllocSize = (1 << (std::numeric_limits<int>::digits - 1)) - 1
};
enum
{
//1G-1-16字节
MaxByteArraySize = MaxAllocSize - sizeof(QtPrivate::remove_pointer<QByteArray::DataPtr>::type)
};
class QRingBuffer
{
public:
//默认分配QRINGBUFFER_CHUNKSIZE大小的buffer
QRingBuffer(int growth = QRINGBUFFER_CHUNKSIZE) :
head(0), tail(0), tailBuffer(0), basicBlockSize(growth), bufferSize(0) { }
~QRingBuffer(){}
//获取环形缓冲区指定位置的指针
//length，输出这个指定位置到缓冲区结尾的长度
char *readPointerAtPosition(qint64 pos, qint64 &length);
//申请空间：从尾开始，返回新空间的指针
char *reserve(qint64 bytes);
//申请空间：从头开始，返回新空间的指针
char *reserveFront(qint64 bytes);
//缩短空间
void truncate(qint64 pos)
{
if (pos < bufferSize)
chop(bufferSize - pos);
}
//判断buffers数据是否为空
bool isEmpty()
{
return bufferSize == 0;
}
//从头读取一个字符，并转换为int返回
int getChar()
{
if (isEmpty())
return -1;
char c = *readPointer();
free(1);
return int(uchar(c));
}
//在缓冲区尾部添加字符
void putChar(char c)
{
char *ptr = reserve(1);
*ptr = c;
}
//在缓冲区头部添加字符
void ungetChar(char c)
{
if (head > 0) {
--head;
buffers.first()[head] = c;
++bufferSize;
} else {
char *ptr = reserveFront(1);
*ptr = c;
}
}
//清空缓冲区
void clear();
//读取maxLength长度数据到data中，如果buffers中的数据少于maxLength，则读取所有数据，
//返回读取数据的长度
qint64 read(char *data, qint64 maxLength);
//读取buffers中的第一个buffer
QByteArray read();
//从指定位置pos拷贝maxLength长度的数据到data中
//返回实际截取的数据长度
qint64 peek(char *data, qint64 maxLength, qint64 pos = 0);
//扩展最后一个buffer
void append(const char *data, qint64 size);
//在最后添加一个新buffer
void append(const QByteArray &qba);
//从头释放lenght长度空间，一般需要配合reserve使用
qint64 skip(qint64 length)
{
qint64 bytesToSkip = qMin(length, bufferSize);
free(bytesToSkip);
return bytesToSkip;
}
//从尾释放length长度空间，一般需要配合reserve使用
void chop(qint64 length);
//读取一行,包括该行的结束标志'\n'
qint64 readLine(char *data, qint64 maxLength);
bool canReadLine()
{
return indexOf('\n', bufferSize) >= 0;
}
private:
//获取下一个数据块的大小
//如果只剩一个buffer，返回最后一个buffer所含数据的大小；否则返回第一个buffer所含数据的大小。
qint64 nextDataBlockSize()
{
return (tailBuffer == 0 ? tail : buffers.first().size()) - head;
}
//获取缓冲区第一个有效数据的指针
char *readPointer()
{
return bufferSize == 0 ? Q_NULLPTR : (buffers.first().data() + head);
}
qint64 indexOf(char c, qint64 maxLength, qint64 pos = 0);
//释放空间
void free(qint64 bytes);
private:
QList<QByteArray> buffers;
//标识第一个buffer数据起始位置和最后一个buffer数据的结尾位置
int head, tail;
//大小为buffers.size()-1，如果为0，说明只剩一个buffer
int tailBuffer;
//初始分配空间的大小
int basicBlockSize;
//buffers数据总大小
qint64 bufferSize;
};
#endif // QRINGBUFFER_P_H

qringbuffer.cpp

#include "qringbuffer.h"
#include <string.h>
char *QRingBuffer::readPointerAtPosition(qint64 pos, qint64 &length)
{
if (pos >= 0)
{
pos += head;
for (int i = 0; i < buffers.size(); ++i)
{
length = (i == tailBuffer ? tail : buffers[i].size());
if (length > pos)
{
length -= pos;
return buffers[i].data() + pos;
}
pos -= length;
}
}
length = 0;
return 0;
}
void QRingBuffer::free(qint64 bytes)
{
Q_ASSERT(bytes <= bufferSize);
while (bytes > 0)
{
const qint64 blockSize = buffers.first().size() - head;
if (tailBuffer == 0 || blockSize > bytes)
{
if (bufferSize <= bytes)
{
if (buffers.first().size() <= basicBlockSize)
{
bufferSize = 0;
head = tail = 0;
} else
{
clear();
}
}
else
{
Q_ASSERT(bytes < MaxByteArraySize);
head += int(bytes);
bufferSize -= bytes;
}
return;
}
bufferSize -= blockSize;
bytes -= blockSize;
buffers.removeFirst();
--tailBuffer;
head = 0;
}
}
char *QRingBuffer::reserve(qint64 bytes)
{
if (bytes <= 0 || bytes >= MaxByteArraySize)
return 0;
if (buffers.isEmpty())
{
buffers.append(QByteArray());
buffers.first().resize(qMax(basicBlockSize, int(bytes)));
}
else
{
const qint64 newSize = bytes + tail;
//如果超过最后一个buffer所含数据的大小，则最后一个buffer需要从新分配
if (newSize > buffers.last().size())
{
//满足以下条件时，将最后一个buffer的容积缩小到其当前所含数据的大小，
//然后新开辟一个buffer，并将该buffer数据的结尾位置tail设置为0
if (newSize > buffers.last().capacity() && (tail >= basicBlockSize
|| newSize >= MaxByteArraySize))
{
buffers.last().resize(tail);
buffers.append(QByteArray());
++tailBuffer;
tail = 0;
}
//将最后一个buffer进行扩容
buffers.last().resize(qMax(basicBlockSize, tail + int(bytes)));
}
}
char *writePtr = buffers.last().data() + tail;
bufferSize += bytes;
Q_ASSERT(bytes < MaxByteArraySize);
tail += int(bytes);
return writePtr;
}
char *QRingBuffer::reserveFront(qint64 bytes)
{
if (bytes <= 0 || bytes >= MaxByteArraySize)
return 0;
if (head < bytes)
{
if (buffers.isEmpty())
{
buffers.append(QByteArray());
}
else
{
buffers.first().remove(0, head);
if (tailBuffer == 0)
tail -= head;
}
head = qMax(basicBlockSize, int(bytes));
if (bufferSize == 0)
{
tail = head;
}
else
{
buffers.prepend(QByteArray());
++tailBuffer;
}
buffers.first().resize(head);
}
head -= int(bytes);
bufferSize += bytes;
return buffers.first().data() + head;
}
void QRingBuffer::chop(qint64 length)
{
Q_ASSERT(length <= bufferSize);
while (length > 0)
{
if (tailBuffer == 0 || tail > length)
{
if (bufferSize <= length)
{
if (buffers.first().size() <= basicBlockSize)
{
bufferSize = 0;
head = tail = 0;
}
else
{
clear();
}
}
else
{
Q_ASSERT(length < MaxByteArraySize);
tail -= int(length);
bufferSize -= length;
}
return;
}
bufferSize -= tail;
length -= tail;
buffers.removeLast();
--tailBuffer;
tail = buffers.last().size();
}
}
void QRingBuffer::clear()
{
if (buffers.isEmpty())
return;
buffers.erase(buffers.begin() + 1, buffers.end());
buffers.first().clear();
head = tail = 0;
tailBuffer = 0;
bufferSize = 0;
}
qint64 QRingBuffer::indexOf(char c, qint64 maxLength, qint64 pos)
{
if (maxLength <= 0 || pos < 0)
return -1;
qint64 index = -(pos + head);
for (int i = 0; i < buffers.size(); ++i)
{
qint64 nextBlockIndex = qMin(index + (i == tailBuffer ? tail : buffers[i].size()),
maxLength);
if (nextBlockIndex > 0)
{
const char *ptr = buffers[i].data();
if (index < 0)
{
ptr -= index;
index = 0;
}
const char *findPtr = reinterpret_cast<const char *>(memchr(ptr, c,
nextBlockIndex - index));
if (findPtr)
return qint64(findPtr - ptr) + index + pos;
if (nextBlockIndex == maxLength)
return -1;
}
index = nextBlockIndex;
}
return -1;
}
qint64 QRingBuffer::read(char *data, qint64 maxLength)
{
const qint64 bytesToRead = qMin(bufferSize, maxLength);
qint64 readSoFar = 0;
while (readSoFar < bytesToRead)
{
const qint64 bytesToReadFromThisBlock = qMin(bytesToRead - readSoFar,
nextDataBlockSize());
if (data)
memcpy(data + readSoFar, readPointer(), bytesToReadFromThisBlock);
readSoFar += bytesToReadFromThisBlock;
free(bytesToReadFromThisBlock);
}
return readSoFar;
}
QByteArray QRingBuffer::read()
{
if (bufferSize == 0)
return QByteArray();
QByteArray qba(buffers.takeFirst());
//避免调整大小时不必要的内存分配，使QByteArray更高效
qba.reserve(0);
if (tailBuffer == 0)
{
qba.resize(tail);
tail = 0;
} else
{
--tailBuffer;
}
qba.remove(0, head);
head = 0;
bufferSize -= qba.size();
return qba;
}
qint64 QRingBuffer::peek(char *data, qint64 maxLength, qint64 pos)
{
qint64 readSoFar = 0;
if (pos >= 0)
{
pos += head;
for (int i = 0; readSoFar < maxLength && i < buffers.size(); ++i)
{
qint64 blockLength = (i == tailBuffer ? tail : buffers[i].size());
if (pos < blockLength)
{
blockLength = qMin(blockLength - pos, maxLength - readSoFar);
memcpy(data + readSoFar, buffers[i].data() + pos, blockLength);
readSoFar += blockLength;
pos = 0;
}
else
{
pos -= blockLength;
}
}
}
return readSoFar;
}
void QRingBuffer::append(const char *data, qint64 size)
{
char *writePointer = reserve(size);
if (size == 1)
*writePointer = *data;
else if (size)
::memcpy(writePointer, data, size);
}
void QRingBuffer::append(const QByteArray &qba)
{
if (tail == 0)
{
if (buffers.isEmpty())
buffers.append(qba);
else
buffers.last() = qba;
}
else
{
buffers.last().resize(tail);
buffers.append(qba);
++tailBuffer;
}
tail = qba.size();
bufferSize += tail;
}
qint64 QRingBuffer::readLine(char *data, qint64 maxLength)
{
if (!data || --maxLength <= 0)
return -1;
qint64 i = indexOf('\n', maxLength);
i = read(data, i >= 0 ? (i+1) : maxLength);
data[i] = '\0';
return i;
}

main.cpp

#include <qringbuffer.h>
#include <QDebug>
int main()
{
//测试环形缓冲区的写入和读取+++++++++++++++++++++++++++++++
qDebug()<<QStringLiteral("测试环形缓冲区的写入和读取+++++++++++++++++++++++++++++++");
//方法1
QRingBuffer ringBuffer;
ringBuffer.append("CSDN ",5);
ringBuffer.append("blog ",5);
QByteArray qba("http://blog.csdn.net/caoshangpa");
ringBuffer.append(qba);
QByteArray head=ringBuffer.read();
QByteArray tail=ringBuffer.read();
qDebug()<<head<<tail;
//方法2
ringBuffer.clear();
ringBuffer.append("CSDN ",5);
ringBuffer.append("blog ",5);
ringBuffer.append(qba);
char str[100]={'\0'};
ringBuffer.read(str,100);
qDebug()<<str;
//测试在缓冲区头和尾添加字符+++++++++++++++++++++++++++++++
qDebug()<<QStringLiteral("测试在缓冲区头和尾添加字符+++++++++++++++++++++++++++++++");
ringBuffer.clear();
ringBuffer.append("CSDN ",5);
ringBuffer.append("blog ",5);
ringBuffer.append(qba);
//头
ringBuffer.ungetChar('{');
//尾
ringBuffer.putChar('}');
memset(str,0,100);
ringBuffer.read(str,100);
qDebug()<<str;
//测试读取一行+++++++++++++++++++++++++++++++++++++++++++
qDebug()<<QStringLiteral("测试读取一行+++++++++++++++++++++++++++++++++++++++++++");
ringBuffer.clear();
ringBuffer.append("CSDN ",5);
ringBuffer.append("blog\n",5);
ringBuffer.append(qba);
memset(str,0,100);
if(ringBuffer.canReadLine())
ringBuffer.readLine(str,100);
qDebug()<<str;
//测试拷贝数据+++++++++++++++++++++++++++++++++++++++++++
qDebug()<<QStringLiteral("测试拷贝数据+++++++++++++++++++++++++++++++++++++++++++");
ringBuffer.clear();
ringBuffer.append("CSDN ",5);
ringBuffer.append("blog ",5);
ringBuffer.append(qba);
memset(str,0,100);
ringBuffer.peek(str,10),
qDebug()<<str;
//证明peek只是拷贝数据
memset(str,0,100);
ringBuffer.read(str,100);
qDebug()<<str;
//测试释放指定长度缓冲区+++++++++++++++++++++++++++++++++++++++++++
qDebug()<<QStringLiteral("测试释放指定长度缓冲区+++++++++++++++++++++++++++++++++++++++++++");
ringBuffer.clear();
ringBuffer.append("CSDN ",5);
ringBuffer.append("blog ",5);
ringBuffer.append(qba);
memset(str,0,100);
ringBuffer.skip(10),
ringBuffer.read(str,100);
qDebug()<<str;
//测试申请指定长度缓冲区+++++++++++++++++++++++++++++++++++++++++++
qDebug()<<QStringLiteral("测试申请指定长度缓冲区+++++++++++++++++++++++++++++++++++++++++++");
ringBuffer.clear();
ringBuffer.reserve(100);
qint64 lenToTail=0;
char * pos=ringBuffer.readPointerAtPosition(10,lenToTail);
char * test="CSDN blog http://blog.csdn.net/caoshangpa";
memcpy(pos,test,strlen(test));
ringBuffer.skip(10);
ringBuffer.chop(100-10-strlen(test));
qDebug()<<ringBuffer.read();
qDebug()<<lenToTail;
system("pause");
}

测试结果

Qt实现环形缓冲区的两种方法

二.使用QSemaphore

用定时器将当前时间存储到QStringList对象中，然后通过现场去QStringList对象中取出并打印，通过两个QSemaphore对象进行同步，形成环形缓冲区。

thread.h

#ifndef THREAD_H
#define THREAD_H
#include <QThread>
#include <QSemaphore>
#include <QStringList>
class Thread : public QThread
{
Q_OBJECT
public:
Thread();
~Thread();
protected:
void run();
private slots:
void slotTimeout();
};
#endif // THREAD_H

thread.cpp

#include "thread.h"
#include <QDebug>
#include <QTimer>
#include <QTime>
//1024是这个list的缓冲区大小，当然可以随意设定；80的意思是初始空闲的缓冲区大小是1024
QSemaphore freeBytes(1024);
//初始被使用的缓冲区大小是0
QSemaphore usedBytes(0);
QStringList timeList;
Thread::Thread()
{
QTimer *timer=new QTimer(this);
connect(timer,SIGNAL(timeout()),this,SLOT(slotTimeout()));
timer->start(50);
}
Thread::~Thread()
{
}
void Thread::run()
{
//获取并打印list中的时间
while(true)
{
usedBytes.acquire();
qDebug()<<timeList.takeFirst();
freeBytes.release();
}
}
//通过定时器给list存入当前时间
void Thread::slotTimeout()
{
freeBytes.acquire();//先申请一个位置，相当于空闲的缓冲区变为1023
timeList.append(QTime::currentTime().toString("now : hh:mm:ss:zzz"));
usedBytes.release();//给使用的大小+1
}

main.cpp

#include <QCoreApplication>
#include "thread.h"
int main(int argc, char *argv[])
{
QCoreApplication a(argc, argv);
Thread *thread=new Thread;
thread->start();
return a.exec();
}

测试结果

Qt实现环形缓冲区的两种方法