C++ 笔记 —— 实现一个环形阻塞队列

Posted 2023-02-24 湖广午王

实现原理

环形阻塞队列，顾名思义，首先，它是一个队列，然后，它应当是一个环形，并且它是会进行阻塞的。但是根据我们的常识，内存地址是用一个long long int来存储的，我们存储的数据的地址无法绕成一个环，所以我们想要成环的话，需要我们自己去处理。

如上图，相比环状实现的来说，数据在内存中的存储更接近线性实现那样。线性实现和环形实现中，我们都记录着队头、队尾。如果是一个内存中，数据可以存储为环形，那么我们只需要在写入数据的时候，注意写在队尾，读取数据时，从堆头开始读取就行了。但是很明显，数据在内存中无法存储位环形，所以我们还需要做以下处理：

• 写入数据时，当需要写入的数据长度大于末尾上的可用存储空间时，需要把待写入的数据拆分成两部分，一部分写在末尾，一部分写在最前。
• 读取数据时，当需要读取的数据长度大于末尾上的有效数据长度，且最前面还有有效数据时，需要读取末尾的数据，在读出最前面的一部分数据
• 写入数据和读取数据后，需要及时的更新队头和队尾的游标，以便于下一次读取。
• 为保证线程安全，避免同时写入和读取

另外我们需要实现的环形队列还需要有阻塞的功能。所以在，上面的基础上，我们还要做更多的处理：

• 写入数据，如果需要写入的数据长度，大于剩余的可用的存储空间时，需要将可用的存储空间填满，然后等待有可用存储空间时，再写入还没有写入的数据到队列中。
• 读取数据后，通知有可用的存储空间，以方便数据写入。

实现代码

根据实现原来，我们来用C++11进行实现，首先我们来定义头文件，为了这个队列可以存储各类数据，我们将队列定义为一个模板类，内容如下：

#pragma once

#include <cstdint>
#include <mutex>
#include <condition_variable>

template <typename T>
class CircleBlockQueue 
private:
	//用来存储数据
    T * data;
    //总容量
    uint64_t size;

	//用来记录读取位置的游标
    uint64_t readPos0;
    //用来记录写入位置的游标
    uint64_t writePos0;
    //当前队列中有效数据的长度
    uint64_t currentDataLength0;

	//用来实现阻塞的互斥锁及信号量
    std::mutex mutex;
    std::condition_variable capacityLock;

    bool clearFlagfalse;
	
    bool innerPush(T * data,uint64_t length);
public:
    explicit CircleBlockQueue(uint64_t size);
    ~CircleBlockQueue();
    //向队列中添加数据
    bool push(T * data,uint64_t length);
    //从队列中读出数据
    uint64_t pop(T * data,uint64_t length);
    //清除队列中的数据
    void clear();
    //队列数据长度
    uint64_t length();
;

#include "CircleBlockQueue.inl"

CircleBlockQueue.inl 具体的实现如下，相关细节见注释：

#include <cstdlib>
#include <cstring>

template <typename T>
CircleBlockQueue<T>::CircleBlockQueue(uint64_t size):size(size) 
    data = (T *)malloc(size * sizeof(T));


template <typename T>
CircleBlockQueue<T>::~CircleBlockQueue() 
    free(data);


template <typename T> bool CircleBlockQueue<T>::innerPush(T *t, uint64_t length) 
    if(length <= size - currentDataLength)
        //尾巴上的容量不足以存储期望push的数据时，需要将数据分成两段，一段在队尾，一段在队头
        //足够存储时，直接存储即可
        if(writePos + length > size)
            uint64_t remainLength = size - writePos;
            memcpy((void *)(this->data + writePos),(void *)t,remainLength* sizeof(T));
            memcpy((void *)(this->data),(void *)(t+remainLength),(length-remainLength) * sizeof(T));
            writePos = length + writePos - size;
            currentDataLength += length;
        else
            memcpy((void *)(this->data + writePos),(void *)t,length * sizeof(T));
            writePos += length;
            currentDataLength += length;
        
        return true;
    
    return false;


template <typename T>
bool CircleBlockQueue<T>::push(T *t, uint64_t length) 
    std::unique_lock<std::mutex> lck(mutex);
    T * dataPos = t;
    uint64_t currentLength = length;
    do
        if(!innerPush(dataPos,currentLength))
            //容量不足以够存储期望push的数据时，按照能力进行存储
            //然后等待有新的存储空间时，再次尝试存储
            uint64_t remainSize = size - currentDataLength;
            if(remainSize > 0)
                innerPush(dataPos,remainSize);
                currentLength -= remainSize;
                dataPos += remainSize;
            
            capacityLock.wait(lck);
        else
            break;
        
    while(!clearFlag);
    clearFlag = false;


template <typename T>
uint64_t CircleBlockQueue<T>::pop(T *t, uint64_t length) 
    std::unique_lock<std::mutex> lck(mutex);
    if(currentDataLength)
        //当前数据大于期望读取的数据长度
        uint64_t realReadLength = length;
        if(length > currentDataLength)
            realReadLength = currentDataLength;
        
        //被读取的数据不是在首尾两段
        if(size - readPos >= realReadLength)
            memcpy((void *)t,(void *)(this->data+readPos),realReadLength* sizeof(T));
            readPos += realReadLength;
            currentDataLength -= realReadLength;
        else
            uint64_t readLength = size - readPos;
            memcpy((void *)t,(void*)(this->data + readPos),readLength * sizeof(T));
            memcpy((void *)(t+readLength),(void *)(this->data),(realReadLength - readLength) * sizeof(T));
            readPos = readPos + realReadLength - size;
            currentDataLength -= realReadLength;
        
        capacityLock.notify_all();
        return realReadLength;
    
    return 0;


template <typename T>
void CircleBlockQueue<T>::clear() 
    std::unique_lock<std::mutex> lck(mutex);
    clearFlag = true;
    readPos = 0;
    writePos = 0;
    currentDataLength = 0;
    memset(data,0, size * sizeof(T));
    capacityLock.notify_all();


template <typename T>
uint64_t CircleBlockQueue<T>::length() 
    std::unique_lock<std::mutex> lck(mutex);
    return currentDataLength;

简单测试

实现了一个环形阻塞队列后，我们需要对实现进行测试，看看实现是否符合我们的预期：

void writeThread(CircleBlockQueue<char> * queue)
    while(threadFlag)
        queue->push(const_cast<char *>("hello"), 5);
    


void queueTest()
    CircleBlockQueue<char> testQueue(8);
    std::thread thread(writeThread,&testQueue);
    thread.detach();

    std::string cmd;
    while(cmd != "exit")
        std::cin>>cmd;
        if(cmd == "clear")
            testQueue.clear();
        else if(cmd == "exit")
            testQueue.clear();
            threadFlag = false;
        else
            char temp[3];
            testQueue.pop(temp, 3);
            std::cout <<"pop:"<< temp <<std::endl;
        
    
    testQueue.clear();


void main()
	queueTest();
	return 0;

其输入输出如下：

结合代码，从结果可以看到，push、pop、clear方法的结果都符合我们的预期，这样一个环形阻塞队列就实现完成了。

---

欢迎转载，转载请保留文章出处。湖广午王的博客[http://blog.csdn.net/junzia/article/details/88926431]

---