缓冲池

工作原理

缓冲池是一组帧，每帧保存一个页和元数据：页 ID、脏标志、钉住计数和访问信息。读取时对页 ID 做哈希，找到帧后直接返回，否则从磁盘加载。写入修改内存中的页，并在页刷回磁盘前记录预写日志。脏页由后台刷写或检查点活动写回。

替换通常采用 LRU 或时钟算法变体。朴素 LRU 在全表扫描下表现很差，因为一次大的顺序读取会驱逐整个热集合。引擎通过中点插入、频率草图或按页类型分池来增强抗扫描能力，有些允许钉住小型热表。

何时运用

在调整内存、调查 I/O 尖峰或解释重启后的延迟时使用这些知识。工作集能放入缓冲池时，读取由内存服务；放不下时会出现重复磁盘读取。写密集负载也依赖缓冲池，因为脏页必须在提交不阻塞的情况下刷写。

权衡

更大的池提高命中率，但会与每查询内存、连接和操作系统竞争。更小的池给操作系统页缓存留出空间，而页缓存可能缓存同样的文件，造成双重缓冲。激进刷写平滑检查点，却增加 I/O；懒惰刷写则在池被脏页填满时引发尖峰。如果未命中都是廉价的顺序读取，单看命中率会产生误导。

示例与误解

一个分析查询在 16 GB 缓冲池上扫描 200 GB 的表，驱逐了热索引页，导致下一次事务查询变慢。为分析增加只读副本，或限制扫描的缓冲使用量，可以保护运营负载。

误解是 99% 的命中率就说明缓冲池健康。如果剩下的 1% 是关键路径上的随机读，延迟仍然可能很糟。要同时衡量命中率和未命中的代价。

代码示例

import os
from typing import BinaryIO, Iterator, Tuple


class ChunkedReadError(ValueError):
    """Raised when chunk parameters are invalid."""


def iter_chunks(path: str, chunk_size: int = 65536, offset: int = 0) -> Iterator[bytes]:
    """Yield the file at path in fixed-size byte chunks starting at offset."""
    if chunk_size < 1:
        raise ChunkedReadError("chunk_size must be >= 1")
    if offset < 0:
        raise ChunkedReadError("offset must be >= 0")
    if not os.path.isfile(path):
        raise FileNotFoundError(f"file not found: {path}")
    if offset > os.path.getsize(path):
        raise ChunkedReadError("offset is beyond the end of the file")

    with open(path, "rb") as handle:
        handle.seek(offset)
        while True:
            chunk = handle.read(chunk_size)
            if not chunk:
                return
            yield chunk


def read_all_chunks(path: str, chunk_size: int = 65536) -> Tuple[bytes, int]:
    """Return the whole file content and the total number of bytes read."""
    parts = []
    total = 0
    for chunk in iter_chunks(path, chunk_size=chunk_size):
        parts.append(chunk)
        total += len(chunk)
    return b"".join(parts), total

import xml.etree.ElementTree as ET
from collections.abc import Iterator
from pathlib import Path


def iter_records(path: Path, tag: str = "record") -> Iterator[dict[str, str]]:
    context = ET.iterparse(path, events=("end",))
    for _, element in context:
        if element.tag != tag:
            continue
        record = {
            child.tag: (child.text or "").strip() for child in element
        }
        yield record
        element.clear()


def write_fixture(path: Path) -> None:
    path.write_text(
        "<catalog>"
        "<record><id>1</id><name>Ada</name></record>"
        "<record><id>2</id><name>Grace</name></record>"
        "</catalog>",
        encoding="utf-8",
    )


if __name__ == "__main__":
    fixture = Path("catalog.xml")
    write_fixture(fixture)
    try:
        print(list(iter_records(fixture)))
    finally:
        fixture.unlink()