基于静态内存的循环队列的实现

码出名企路

我的产品《集合》

为什么有循环队列

基本的标准库容器比如vector，map，list，deque等不能满足复杂的日常需求时，我们就应该想一个更高效的容器去处理我们的业务场景，以此去提升我们程序性能。

什么场景下用循环队列

一个比较常见的业务需求是，当读写数据同时往一处填充时，写的速度和读取速度不一致，并且需要时刻清理历史数据，避免内存溢出，这些历史数据也被参与其他场景的计算中，就导致资源被大量消耗。

可以想到的标准容器

1，有序的插入std::map，底层是红黑树的实现，当写入和删除数据时，都会触发红黑树的调整，降低程序效率；

2，无序的插入std::unordered_map，底层实现是hash，插入和删除的效率不错，但是是动态内存申请，容器的容量时刻更改，频繁的内存申请降低程序的效率；

3，std::list, std::deque, std::queue不支持随机访问，并且访问的时间复杂度O(n),也是不合理的；

4，std::vector, std::set对查找不友好，无法支持按特定需求去进行查找。

循环队列登上历史舞台

固定的时间窗口，肯定是静态内存的使用，并且可以高效访问，插入容器。此时，只需要申请一块连续大小的内存，此内存通常大于实际使用的内存大小，不断在内存区域进行覆写，新数据覆盖老数据，无需重复申请新的内存，循环队列派上用场了。

什么是循环缓冲区队列

循环缓冲区(也称为环形缓冲区)是固定大小的缓冲区，工作原理就像内存是连续的且可循环的一样。在生成和使用内存时，不需将原来的数据全部重新清理掉，只要调整head/tail 指针即可。当添加数据时，head 指针前进。当使用数据时，tail 指针向前移动。当到达缓冲区的尾部时，指针又回到缓冲区的起始位置。

如何实现循环缓冲区队列

1，数据结构

基础数据缓冲区

缓冲区的最大范围

“head”指针的当前位置（添加元素时增加）

“tail”指针的当前位置（读取元素后增加）

一个标志位来指示缓冲区是否已满

// The hidden definition of our circular buffer structure
typedef struct circular_buf_t
{
    uint8_t *buffer;
    size_t head;
    size_t tail;
    size_t max; //of the buffer
    bool full;
} *cbuf_handle_t;

2，确认缓冲区是否已满

当head == tail，循环缓冲区的 “full” 和 “empty” 看起来是相同的：head 和 tail 指针是相等的。

有两种方法区分 full 和 empty：

使用缓冲区中的一个数据位：

Full：tail + 1 == head

Empty：head == tail

使用一个bool标志位和其他逻辑来区分：

Full：full

Empty：(head == tail) && (!full)

从上面的数据结构看到我们使用的是通过标志位来判断，那么就需要在get /set 操作方法中更新该标志位。

3，初始化和复位

/// Pass in a storage buffer and size 
/// Returns a circular buffer handle
cbuf_handle_t circular_buf_init(uint8_t* buffer, size_t size)
{
    assert(buffer && size);

    cbuf_handle_t cbuf = malloc(sizeof(circular_buf_t));
    assert(cbuf);

    cbuf->buffer = buffer;
    cbuf->max = size;
    circular_buf_reset(cbuf);

    assert(circular_buf_empty(cbuf));

    return cbuf;
}

/// Reset the circular buffer to empty, head == tail
void circular_buf_reset(cbuf_handle_t cbuf)
{
    assert(cbuf);

    cbuf->head = 0;
    cbuf->tail = 0;
    cbuf->full = false;
}

/// Free a circular buffer structure.
/// Does not free data buffer; owner is responsible for that
void circular_buf_free(cbuf_handle_t cbuf)
{
    assert(cbuf);
    free(cbuf);
}

4，状态检查

bool circular_buf_full(cbuf_handle_t cbuf)
{
    assert(cbuf);

    return cbuf->full;
}
bool circular_buf_empty(cbuf_handle_t cbuf)
{
    assert(cbuf);

    return (!cbuf->full && (cbuf->head == cbuf->tail));
}
size_t circular_buf_capacity(cbuf_handle_t cbuf)
{
    assert(cbuf);

    return cbuf->max;
}
size_t circular_buf_size(cbuf_handle_t cbuf)
{
    assert(cbuf);

    size_t size = cbuf->max;

    if(!cbuf->full)
    {
        if(cbuf->head >= cbuf->tail)
        {
            size = (cbuf->head - cbuf->tail);
        }
        else
        {
            size = (cbuf->max + cbuf->head - cbuf->tail);
        }
    }

    return size;
}

5，数据填充删除

从缓冲区读取数据后删除，即取出 tail 指针位置的值并更新 tail 指针。如果缓冲区是空的，不修改指针值并且返回 error=-1。

int circular_buf_get(cbuf_handle_t cbuf, uint8_t * data)
{
    assert(cbuf && data && cbuf->buffer);

    int r = -1;

    if(!circular_buf_empty(cbuf))
    {
        *data = cbuf->buffer[cbuf->tail];
        // 当删除数据时，full 标志置为 flase ，tail 指针向前移一位
        cbuf->full = false;
        cbuf->tail = (cbuf->tail + 1) % cbuf->max;
        r = 0;
    }

    return r;
}

当读跟不上写速度的时候有两种方法，一种是覆盖旧数据，一种是不做处理，阻塞或者丢弃新数据

1，阻塞或者丢弃新数据

以丢弃新数据为例，如果buff 已满了（circular_buf_full = true），则不做任何处理

int circular_buf_put_block(cbuf_handle_t cbuf, uint8_t data)
{
    int r = -1;

    assert(cbuf && cbuf->buffer);

    if(!circular_buf_full(cbuf))
    {
        cbuf->buffer[cbuf->head] = data;
        //  将写指针位置往前移动
        cbuf->head = (cbuf->head + 1) % cbuf->max;
        cbuf->full = (cbuf->head == cbuf->tail);
        r = 0;
    }

    return r;
}

以Demo为例

100ms 写一次，而400ms 读一次，buff 大小为16B，所以其阻塞住了，并且丢弃了新来写的数据，只有buff 有缓冲的时候才能写。

void *read_thread(void *arg)
{
    uint8_t data;;
    printf("%s %d\n", __FUNCTION__, __LINE__);
    while (1)
    {
        pthread_mutex_lock(&rw_mutex);
        circular_buf_get(arg, &data);
        pthread_mutex_unlock(&rw_mutex);
        printf("%s %d ===>data:0x%x\n", __FUNCTION__, __LINE__, data);

        usleep(400*1000);
    }

    return NULL;
}

int main()
{
    pthread_t ntid;
    uint8_t data;
    uint8_t buff[16] = {0};
    cbuf_handle_t handle = circular_buf_init(buff, sizeof(buff));
    int ret;
    pthread_create(&ntid, NULL, read_thread, handle);
    printf("%s %d data:0x%x\n", __FUNCTION__, __LINE__, data);
    while (1)
    {
        usleep(100*1000);
        data = rand()%0xff;
        pthread_mutex_lock(&rw_mutex);
        ret = circular_buf_put_block(handle, data);
        pthread_mutex_unlock(&rw_mutex);
        if(!ret)
            printf("%s %d write data:0x%x\n", __FUNCTION__, __LINE__, data);
    }
    free(buff);
    circular_buf_free(handle);
    return 0;
}

打印输出：

main 201 data:0x0
read_thread 179
read_thread 185 ===>data:0x0
main 210 write data:0x29
main 210 write data:0x6b
main 210 write data:0xd6
read_thread 185 ===>data:0x29
main 210 write data:0xeb
main 210 write data:0x2c
main 210 write data:0xa9
main 210 write data:0x3
read_thread 185 ===>data:0x6b
main 210 write data:0x21
main 210 write data:0xbb
main 210 write data:0xef
main 210 write data:0x5f
read_thread 185 ===>data:0xd6
main 210 write data:0x5f
main 210 write data:0x4c
main 210 write data:0xfc
main 210 write data:0x10
read_thread 185 ===>data:0xeb
main 210 write data:0xec
main 210 write data:0xbe
main 210 write data:0xd4
main 210 write data:0xed
read_thread 185 ===>data:0x2c
main 210 write data:0x51
main 210 write data:0x6
read_thread 185 ===>data:0xa9
main 210 write data:0x99
read_thread 185 ===>data:0x3
main 210 write data:0x65
read_thread 185 ===>data:0x21
main 210 write data:0x33

覆盖旧数据

int circular_buf_put_overwrite(cbuf_handle_t cbuf, uint8_t data)
{
    assert(cbuf && cbuf->buffer);

    cbuf->buffer[cbuf->head] = data;

    // 如果buff 满了需要复写读到的数据，那么需要将读位置往前移动
    if (cbuf->full)
    {
        cbuf->tail = (cbuf->tail + 1) % cbuf->max;
    }

    cbuf->head = (cbuf->head + 1) % cbuf->max;
    cbuf->full = (cbuf->head == cbuf->tail);
    return 0;
}

同样以该Demo为例

100ms 写一次，而400ms 读一次，buff 大小为16B，将写替换为 circular_buf_put_overwrite 来覆盖旧数据。

int main()
{
    pthread_t ntid;
    uint8_t data;
    uint8_t buff[16] = {0};
    cbuf_handle_t handle = circular_buf_init(buff, sizeof(buff));
    int ret;
    pthread_create(&ntid, NULL, read_thread, handle);
    printf("%s %d data:0x%x\n", __FUNCTION__, __LINE__, data);
    while (1)
    {
        usleep(100*1000);
        data = rand()%0xff;
        pthread_mutex_lock(&rw_mutex);
        ret = circular_buf_put_overwrite(handle, data);
        pthread_mutex_unlock(&rw_mutex);
        if(!ret)
            printf("%s %d write data:0x%x\n", __FUNCTION__, __LINE__, data);
    }
    free(buff);
    circular_buf_free(handle);
    return 0;
}

100ms 写一次，而400ms 读一次，buff 大小为16B，因为复写了，所以buffer满了之后，读到的数据总是16B写之前写的首次数据

输出

main 227 write data:0x33
main 227 write data:0xec
main 227 write data:0x3f
main 227 write data:0x54
read_thread 202 ===>data:0x51
main 227 write data:0x16
main 227 write data:0xa7
main 227 write data:0x22
main 227 write data:0xcd
read_thread 202 ===>data:0x99
main 227 write data:0xcc
main 227 write data:0x8f
main 227 write data:0x60
main 227 write data:0xd4
read_thread 202 ===>data:0x65
main 227 write data:0xf3
main 227 write data:0x4e
main 227 write data:0x4a
main 227 write data:0x60
read_thread 202 ===>data:0x33   // 读到这个数据位置与之前的写位置正好相差16
main 227 write data:0x3d

推荐阅读：

基于无锁循环队列的线程池的实现

认识一个人就是开了一扇窗户，就能看到不一样的东西，听到不一样的声音，能让你思考，觉悟，这已经够了。其他还有很多，比如机会，帮助，我不确定。这个在一般人看来可能不重要，但是我知道这个很重要。我是小阳哥，一个喜欢码字和交流的程序员：2023，我们依然在一起，希望用身边的人，身边的事，让我们少走一些弯路，一点点就好。欢迎扫码交流，多一点沟通，多一个朋友~

继续滑动看下一个