功能说明
环形缓冲区适合只关心最近一段数据的场景,例如最近的日志行、行情快照、性能采样窗口。容量在创建时固定,写入超过容量后自动覆盖最旧元素,因此内存占用恒定,不会像列表那样无限增长。push 会返回被淘汰的旧值,调用方可以据此做落盘或告警。
内部用数组加头指针和元素计数实现,读写均为常数时间;迭代时按从旧到新的顺序输出,latest 则取最后若干个元素,便于展示最近事件。
关键代码
push 分两种情况:未满时把新元素写到 head 加 count 的位置;已满时先记下 head 位置的旧值,把 head 前进一格,再写到环形末尾对应的槽位。取模保证下标回绕。items 从 head 开始按顺序访问 count 个元素;latest 在参数为零时直接返回空列表,负数则抛出错误,避免切片语义造成误解。
使用注意
容量必须为正整数。元素被覆盖后原值不再被引用,依赖析构清理的资源需要调用方自行处理。缓冲区不是线程安全的,多线程写入需要加锁或改用队列。迭代过程中修改缓冲区会导致当前迭代器看到混合状态,需要快照时先转成列表。由于用 None 表示空槽,存入 None 会被跳过,若业务需要保存空值请改用哨兵对象。
验证方式
容量为三时依次写入 a、b、c,三次 push 都返回 None,items 输出 a、b、c。再写入 d 时返回被淘汰的 a,缓冲区内容变为 b、c、d;latest 取两个元素得到 c、d,长度为三。可以继续写入更多值,确认返回值依次是被覆盖的旧元素。
代码示例
from collections.abc import Iterator
from typing import Generic, TypeVar
T = TypeVar("T")
class RingBuffer(Generic[T]):
def __init__(self, capacity: int) -> None:
if capacity < 1:
raise ValueError("capacity must be positive")
self._slots: list[T | None] = [None] * capacity
self._head = 0
self._count = 0
@property
def capacity(self) -> int:
return len(self._slots)
def __len__(self) -> int:
return self._count
def push(self, item: T) -> T | None:
evicted: T | None = None
if self._count == self.capacity:
evicted = self._slots[self._head]
self._head = (self._head + 1) % self.capacity
index = (self._head + self._count - 1) % self.capacity
else:
index = (self._head + self._count) % self.capacity
self._slots[index] = item
if self._count < self.capacity:
self._count += 1
return evicted
def items(self) -> Iterator[T]:
for offset in range(self._count):
value = self._slots[(self._head + offset) % self.capacity]
if value is not None:
yield value
def latest(self, count: int) -> list[T]:
if count < 0:
raise ValueError("count must not be negative")
if count == 0:
return []
return list(self.items())[-count:]
def main() -> None:
buffer: RingBuffer[str] = RingBuffer(3)
for event in ["a", "b", "c"]:
print("push", event, "evicted", buffer.push(event))
print("full:", list(buffer.items()))
print("push d evicted:", buffer.push("d"))
print("after:", list(buffer.items()))
print("latest 2:", buffer.latest(2))
print("length:", len(buffer))
if __name__ == "__main__":
main()