← 中文资源

简体中文版本

覆盖式环形缓冲区

用固定容量保存最近的若干条记录,溢出时覆盖最旧元素并返回被淘汰的值,支持读取最新切片。

阅读英文原文

功能说明

环形缓冲区适合只关心最近一段数据的场景,例如最近的日志行、行情快照、性能采样窗口。容量在创建时固定,写入超过容量后自动覆盖最旧元素,因此内存占用恒定,不会像列表那样无限增长。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()