RingBuffer，或者说Circular Buffer,是一个长度固定的缓冲区，当从一端插入元素超过指定的最大长度时，缓冲区另一端的元素会溢出。当用户仅对近期一定长度的历史记录感兴趣时，这种数据结构就特别有用。

根据[wiki](# Circular buffer - Wikipedia ),它有三个要素：

* 缓冲区的长度固定

* 先进先出

* 当往缓冲区中增加或者删除元素时，其它元素的位置保持不变。

本文介绍了三种实现方法及各自的特点。

<!--more-->

# collections.dequeue实现

RingBuffer的一个直观实现是使用Pyhton collections的dequeue(发音deck)：

``Python

import collections

import time

 

d = collections.deque(maxlen=1000)

tm1 = time.time()

for i in range(int(1e6)):

    d.append(i)

 

print(list(d)[:10])

d.clear()

tm2 = time.time()

print("{:.2f}seconds".format(tm2 - tm1))

 

执行结果如下：

[999000, 999001, 999002, 999003, 999004, 999005, 999006, 999007, 999008, 999009]

0.20seconds

支持的操作有pop, append;以及从左端操作的popleft, appendleft。但没有缓冲区需要的read操作。缓冲区read操作包括两个动作，一是从缓冲区中读取数据；二是在读取后清除掉数据。我们可以通过list(d:dequeue)和clear的组合来实现这一功能。

# 一个有趣的自定义实现

David Ascher在[这篇文章](# Implementing a Ring Buffer [Book] )里给出了一个自定义的实现：

``python

class RingBuffer:

    """ class that implements a not-yet-full buffer """

    def __init__(self,size_max):

        self.max = size_max

        self.data = []

 

    class __Full:

        """ class that implements a full buffer """

        def append(self, x):

            """ Append an element overwriting the oldest one. """

            self.data[self.cur] = x

            self.cur = (self.cur+1) % self.max

        def get(self):

            """ return list of elements in correct order """

            ret = self.data[self.cur:]+self.data[:self.cur]

            self.data.clear()

            return ret

 

    def append(self,x):

        """append an element at the end of the buffer"""

        self.data.append(x)

        if len(self.data) == self.max: #line 21

            self.cur = 0

            # Permanently change self's class from non-full to full

            self.__class__ = self.__Full

 

    def get(self):

        """ Return a list of elements from the oldest to the newest. """

        return self.data

 

x = RingBuffer(1000)

 

import time

tm1 = time.time()

for i in range(int(1e6)):

    x.append(i)

 

print(x.get()[:10])

tm2 = time.time()

print("{:.2f}seconds".format(tm2 - tm1))

 

执行结果如下：

[999000, 999001, 999002, 999003, 999004, 999005, 999006, 999007, 999008, 999009]

0.64seconds

这个实现的特点是，使用了数组，而不是双端队列来作为基础存储数据结构。当然由于Ring Buffer的访问特性，我们也基本上只对队头、队尾元素进行操作，所以无论是使用数组还是双端队列，操作复杂度都是O(1)。

另一个值得一提的点是，它使用了动态改变对象_class_实现的方式，避免了不必要的判断语句操作，即在创建之初，缓冲区未满时，每插入一个元素，都要在第19行处进行一次判断；而一旦缓冲区被充满后，RingBuffer对象升级为__Full类，从而行为也从此改变--今后再向缓冲区添加新的元素时，不再进行条件语句判断。

尽管代码做了足够的优化，python内建dequeue的实现性能更高一些。

# 使用C语言的实现-pyringbuf

[sirlark](# sirlark )用C语言实现了一个开源的[RingBuffer](# pyringbuf 0.1b2 : Python Package Index )，可以通过pip来安装使用。

        pip install pyringbuf

这个模块提供了push, pop, write, read等函数，使用示例如下：

>>> from ringbuf import RingBuffer

>>> R = RingBuffer(5)    #choose your buffer size

>>> R.push("a")          #push a single character into the buffer

>>> R.pop()              #pop a single character

'a'

>>> R.write("bcdef")     #fill buffer with many characters at once

>>> R.read(4)            #read many characters at once

'bcde'

>>> R.read(1)

'f'

>>> R.read(1)            #returns an empty string if the buffer is empty

这个模块以c语言源码提供，使用前需要编译。