FlaxKV2

✨ 特性亮点

FlaxKV2 是一个提供 类字典接口 的持久化键值存储库，将 LMDB 的高性能与 Python 的易用性完美结合。

核心特性

• 🚀 极致性能：基于 LMDB（mmap + MVCC），缓存命中时读可达 2M+ ops/s
• 🎯 简单易用：Python dict 风格的 API，上手即用
• 🔒 线程安全 / 多进程安全：LMDB 原生支持多进程读写，无需额外中转
• 📦 丰富类型：原生支持字符串、数字、列表、字典、NumPy 数组、Pandas DataFrame
• 🔐 原子操作与事务：pop/setdefault 单事务原子；cross_db_txn/atomic_txn 支持跨 sub_db 的读-判断-写条件更新
• 🗂️ Named Sub-DB：同一 env 内多个独立 B+ 树，物理隔离且可跨库原子读写
• 🌐 远程访问：基于 ZeroMQ 的客户端/服务器架构，支持 CurveZMQ 加密与 LZ4 压缩
• ⏰ TTL 支持：键自动过期功能，后台线程自动清理
• 🪆 嵌套存储：高效的嵌套字典/列表存储，避免频繁序列化
• 📋 持久化数据结构：FlaxList（完整 list API + sorted 模式 + 流式统计）、 FlaxQueue（O(1) FIFO）、FlaxTimeSeries（O(log n) 范围查询 + 降采样）、 RotatingFlaxList（时间分片日志 + 自动清理）
• 🔍 范围迭代：keys_iter/items_iter/values_iter 支持 prefix/start/stop/reverse
• 💾 持久性可调：sync/metasync 可配置，配合写缓冲把 N 次写压成 1 次 fsync
• 🛡️ 自动管理：自动关闭、上下文管理器、实例引用计数
• 📊 可视化工具：内置 Inspector CLI；Web UI 为独立包 flaxkv2-web

📦 安装

基础安装

pip install flaxkv2

完整安装（推荐）

# 包含 Pandas 和向量存储等所有可选依赖
pip install flaxkv2[full]

# 或按需安装特定功能
pip install flaxkv2[pandas]     # Pandas DataFrame 序列化
pip install flaxkv2[vector]     # 向量存储（hnswlib）
pip install flaxkv2[migration]  # LevelDB 数据迁移工具

# Web UI 是独立包（依赖 Flask，不随主包安装）
pip install flaxkv2-web

从源码安装

git clone https://github.com/KenyonY/flaxkv2.git
cd flaxkv2
pip install -e .

🚀 快速开始

基础用法（30 秒上手）

from flaxkv2 import FlaxKV

# 创建/打开数据库
db = FlaxKV("my_database", "./data")

# 像使用字典一样使用
db["username"] = "alice"
db["user_data"] = {"age": 30, "city": "Beijing"}
db["scores"] = [95, 87, 92, 88]

# 读取数据
print(db["username"])        # "alice"
print(db["user_data"]["age"]) # 30

# 检查键是否存在
if "username" in db:
    print("User exists!")

# 遍历所有数据
for key, value in db.items():
    print(f"{key}: {value}")

# 删除数据
del db["username"]

# 关闭数据库
db.close()

使用上下文管理器（推荐）

from flaxkv2 import FlaxKV

# 自动管理数据库生命周期
with FlaxKV("my_database", "./data") as db:
    db["key"] = "value"
    print(db["key"])
# 离开 with 块时自动关闭

🔥 核心功能

1. 性能配置文件

FlaxKV2 提供 6 种预设配置，每种只设定两件事：LMDB 的 map_size（初始 mmap 大小，写满会自动翻倍）和 max_readers（最大并发读者数）。

from flaxkv2 import FlaxKV

db = FlaxKV("mydb", "./data", performance_profile='read_optimized')

⚠️ profile 不配置缓存。读缓存和写缓冲由 read_cache_size / write_buffer_size 单独控制，且单位是条目数不是字节。只传 performance_profile 得到的是无缓存的 RawLmdbDict。

自定义配置：

# 覆盖 profile 的 map_size / max_readers
db = FlaxKV("mydb", "./data",
            performance_profile='large_database',
            map_size=8 * 1024**3,   # 8 GB
            max_readers=512)

# 想要缓存，传缓存参数（自动切到 CachedLmdbDict）
db = FlaxKV("mydb", "./data",
            read_cache_size=50_000,   # 缓存 5 万个条目
            write_buffer_size=1_000)  # 攒够 1000 条再落盘

2. TTL（键过期）

from flaxkv2 import FlaxKV

db = FlaxKV("mydb", "./data")

# 设置键值
db["session_token"] = "abc123"

# 设置 10 秒后过期
db.set_ttl("session_token", 10)

# 获取剩余时间
ttl = db.get_ttl("session_token")
print(f"剩余 {ttl} 秒")

# 10 秒后，键自动删除
# db["session_token"]  # KeyError

# 设置默认 TTL（所有新键都会应用）
db = FlaxKV("cache_db", "./data", default_ttl=300)  # 5分钟
db["key"] = "value"  # 自动 5 分钟后过期

3. 嵌套存储（高性能字典/列表）

from flaxkv2 import FlaxKV

# 启用自动嵌套模式
db = FlaxKV("mydb", "./data", auto_nested=True)

# 嵌套字典
db["config"] = {
    "database": {
        "host": "localhost",
        "port": 5432,
        "credentials": {
            "user": "admin",
            "password": "secret"
        }
    }
}

# 直接访问嵌套键，避免整体反序列化
config = db["config"]
print(config["database"]["host"])  # "localhost"

# 修改嵌套值（高效，不需要读取整个对象）
config["database"]["port"] = 3306

# 嵌套列表
db["users"] = ["alice", "bob", "charlie"]
users = db["users"]
users.append("david")  # 直接操作，高效持久化
print(len(users))  # 4

4. FlaxList — 持久化列表

持久化列表，支持完整 Python list API、sorted 模式、函数式查询和流式统计分析。底层是 LMDB，原生支持多进程并发读。

from flaxkv2 import FlaxList

lst = FlaxList("events", "./data")

# 追加
lst.append({"event": "click", "ts": 12345})
lst.extend([{"event": "scroll"}, {"event": "submit"}])

# 按索引读取
lst[0]          # 第一条
lst[-1]         # 最后一条
lst[10:20]      # 切片（利用 range iterator）

# 迭代
for item in lst:
    print(item)

# 范围迭代（内存友好，适合大数据量）
for item in lst.iter_range(1000, 2000):
    print(item)

# 长度 O(1)
len(lst)

lst.close()

函数式查询（惰性求值，分批流式读取）：

lst = FlaxList("numbers", "./data")
lst.extend(range(10000))

# 过滤
evens = list(lst.where(lambda x: x % 2 == 0))

# 查找第一个/最后一个满足条件的元素
lst.first(lambda x: x > 100)
lst.last(lambda x: x < 500)

# 映射
doubled = list(lst.map(lambda x: x * 2))

# 分批处理（适合大数据集）
for batch in lst.batch(1000):
    process(batch)

# 条件检查（短路求值）
lst.any_match(lambda x: x > 9998)   # True（range(10000) 最大值是 9999）
lst.all_match(lambda x: x >= 0)     # True

流式统计分析（O(1) 内存，单次遍历）：

# 纯数值列表
lst = FlaxList("scores", "./data")
lst.extend([85, 92, 78, 95, 88, 76, 91])

s = lst.stats()
print(s.count, s.mean, s.std)  # 7, 86.43, 6.65
print(s.min, s.max, s.sum)     # 76.0, 95.0, 605.0

# 字典列表，按字段统计
lst = FlaxList("products", "./data")
lst.extend([
    {"name": "A", "price": 10.5},
    {"name": "B", "price": 25.0},
    {"name": "C", "price": 18.8},
])
s = lst.stats(key=lambda x: x["price"])
print(s.mean)  # 18.1

# sorted 模式：精确中位数和分位数
lst = FlaxList("sorted_data", "./data", sorted=True)
for v in [5, 1, 3, 2, 4]:
    lst.add(v)
s = lst.stats(percentiles=[25, 50, 75])
print(s.median)       # 3.0
print(s.percentiles)  # {'p25': 2.0, 'p50': 3.0, 'p75': 4.0}

# 自定义聚合（reduce）
result = lst.reduce(
    lambda acc, x: {"sum": acc["sum"] + x, "count": acc["count"] + 1},
    {"sum": 0, "count": 0},
)

5. 批量操作与范围迭代

db = FlaxKV("mydb", "./data")

# 批量写入（内部使用 LMDB write transaction，原子且高性能）
db.batch_set({"k1": "v1", "k2": "v2", "k3": "v3"})

# 批量读取
values = db.batch_get(["k1", "k2", "k3"])

# 范围迭代（利用 LMDB 有序遍历）
for key in db.keys_iter(start="b", stop="d"):
    print(key)  # 只返回 [b, d) 范围内的键

# 反向迭代
for key in db.keys_iter(reverse=True):
    print(key)

# 前缀过滤
for key in db.keys_iter(prefix="user:"):
    print(key)

# 键值对迭代 / 值迭代
for key, value in db.items_iter(start="a", stop="z"):
    print(key, value)

for value in db.values_iter(prefix="config:"):
    print(value)

6. 原子操作与事务

单个 pop / setdefault 在一个 LMDB 事务内完成读-改-写，可直接用来做任务认领：

# 多个 worker 并发认领任务，同一个 key 只会被一个 worker 抢到
task = db.pop("pending:job1", None)
if task is not None:
    process(task)

# 不存在才写入，返回最终值
cfg = db.setdefault("config", {"retries": 3})

跨 sub_db 事务：cross_db_txn 把多个库的读写放进同一个写事务， txn.get 是事务内读，看得到本事务尚未提交的写，因此能做条件更新：

from flaxkv2 import FlaxKV, cross_db_txn, atomic_txn

counters = FlaxKV("app", "./data", sub_db="counters")
grants   = FlaxKV("app", "./data", sub_db="grants")   # 同 path → 共享一个 env

with cross_db_txn(counters, grants) as txn:
    used = txn.get(counters, "used", 0)
    if used < 10:
        txn.put(counters, "used", used + 1)
        txn.put(grants, user_id, used)

批量写用 atomic_txn：它接受一个函数而非 with 块，因此写满 map_size 时 能自动扩容并重放，适合大批量写入：

def load(txn):
    for k, v in big_dict.items():
        txn.put(db, k, v)

atomic_txn(db, fn=load)     # N 次写合并成 1 次提交 = 1 次 fsync

⚠️ fn 必须幂等（重试会从头再跑），且事务不能跨 await 持有。 详见 事务与并发原语。

7. Named Sub-Database

同一个 LMDB environment 内开多个独立 B+ 树，物理隔离：

meta = FlaxKV("store", "./data", sub_db="meta")
body = FlaxKV("store", "./data", sub_db="body")

# 扫描 meta 不会触碰 body 的 overflow pages
for k in meta.keys_iter():
    ...

同一 path 下的 sub_db 共享一个 env，因此可以用 cross_db_txn 跨库原子读写。

8. 持久化数据结构

除 FlaxList 外还有三种，接口风格统一：

from flaxkv2 import FlaxQueue, FlaxTimeSeries, RotatingFlaxList

# FIFO 队列，O(1) 入队/出队
q = FlaxQueue("tasks", "./data", max_size=10_000)
q.put("job_1"); q.put_batch(["job_2", "job_3"])
job = q.get()                    # 'job_1'
q.peek(); q.qsize(); q.empty()
q.compact()                      # 回收已出队空间

# 时间序列，sorted 模式，O(log n) 范围查询
ts = FlaxTimeSeries("metrics", "./data", retention=7*86400, min_size=1000)
ts.append(0.8)                          # 注意 value 在前，timestamp 默认取当前时间
ts.append(0.9, timestamp=t0)            # 显式指定时间戳
ts.extend([(0.7, t1), (0.6, t2)])       # 批量，元素同样是 (value, timestamp)

ts.query(start=t0, end=t1)              # 范围查询，返回 [(ts, value), ...]
ts.query_values(start=t0, end=t1)       # 只要值
ts.stats(start=t0, end=t1)              # 聚合：count/min/max/mean/std
ts.downsample(interval=3600, agg="mean")    # 降采样到每小时
ts.iter_windows(window_size=60, step=30)    # 滑动窗口
ts.latest(n=10); ts.oldest(); ts.count(start=t0)
ts.time_range(); ts.delete_range(t0, t1); ts.cleanup()

# 时间分片日志，自动清理过期分片
log = RotatingFlaxList("app_log", "./data", partition_by="day", retention=7)
log.append({"level": "INFO", "msg": "started"})
log.partitions()                 # 列出分片
log.cleanup()                    # 删除超出 retention 的分片

9. 持久性（fsync）

LMDB 默认每次写事务提交都 fsync，这是写吞吐的物理上限（NVMe 上约 0.4ms/次）。

# 首选提速方式：写缓冲合并提交，N 次写 → 1 次 fsync
# 落盘失败不丢数据：flush()/close() 抛 FlushError，数据留在缓冲里，修好再 close 一次
db = FlaxKV("mydb", "./data", write_buffer_size=500)

# 跳过 meta page 的 fsync。完整性不受影响，崩溃最多丢最后一个事务
db = FlaxKV("mydb", "./data", metasync=False)

# 关掉数据 fsync。丢 durability 保留 ACI：崩溃不损坏库，只回退到较早的一致状态
db = FlaxKV("cache", "./data", sync=False)
db.sync()   # 自己的安全点上补落盘

实测（ext4/NVMe，tests/benchmarks/benchmark_durability.py）：

⚠️ sync / metasync 是 env 级属性：同一 path 下所有 sub_db 共享，且在首次 打开时固定。提速优先级：write_buffer_size → metasync=False → sync=False。

10. 数据库维护

from flaxkv2 import FlaxKV

# 范围压缩 — 大量删除后回收磁盘空间
db = FlaxKV("mydb", "./data")
db.compact()                              # 全量压缩
db.compact(start="old_", stop="old_~")    # 范围压缩
db.close()

# 彻底删除数据库（不可恢复）
FlaxKV.destroy("mydb", "./data")

# 重建：删掉已有数据重新开一个
db = FlaxKV("mydb", "./data", rebuild=True)

# 强制 fsync 落盘（sync=False 打开时的安全点）
db.flush()   # 先把写缓冲推进 LMDB
db.sync()    # 再 fsync；db.sync() 内部已包含 flush

LMDB 是 B+ 树 + MVCC，写事务要么整体提交要么整体回滚，不存在 LevelDB 那种 需要 repair() 的半损坏状态，因此没有提供修复接口。

11. 丰富的数据类型支持

from flaxkv2 import FlaxKV
import numpy as np
import pandas as pd

db = FlaxKV("mydb", "./data")

# 基本类型
db["string"] = "hello"
db["integer"] = 42
db["float"] = 3.14
db["boolean"] = True

# 容器类型
db["list"] = [1, 2, 3, 4, 5]
db["dict"] = {"name": "Alice", "age": 30}
db["tuple"] = (1, 2, 3)
db["set"] = {1, 2, 3}

# NumPy 数组
db["array"] = np.array([[1, 2], [3, 4]])
db["matrix"] = np.random.randn(100, 100)

# Pandas DataFrame（需要安装 pandas）
db["dataframe"] = pd.DataFrame({
    "name": ["Alice", "Bob"],
    "age": [30, 25]
})

# 自定义对象（通过 pickle）
class MyClass:
    def __init__(self, value):
        self.value = value

db["custom"] = MyClass(42)

12. 远程数据库（ZeroMQ）

FlaxKV2 支持通过网络访问数据库，采用客户端/服务器架构。

启动服务器

# 命令行启动
flaxkv2 run --host 0.0.0.0 --port 5555 --data-dir ./data

# 或通过 Python 启动
python -m flaxkv2 run --host 0.0.0.0 --port 5555 --data-dir ./data

服务器选项（默认值可被配置文件的 [server] 段覆盖）：

• --host: 监听地址（默认 127.0.0.1，仅本地可访问）
• --port: 监听端口（默认 5555）
• --data-dir: 数据存储目录（默认 .，即当前目录）
• --workers: 工作线程数（默认 4）
• --log-level: 日志级别（默认 INFO）
• --enable-encryption / --password: CurveZMQ 加密（默认关闭）
• --enable-compression: LZ4 压缩（默认关闭）
• --profile: 使用配置文件中的 [server.profiles.<name>]

客户端连接

from flaxkv2 import FlaxKV

# 方式1：显式指定 backend='remote'（推荐）
db = FlaxKV("remote_db", "127.0.0.1:5555", backend='remote')

# 方式2：使用 tcp:// 前缀自动识别
db = FlaxKV("remote_db", "tcp://127.0.0.1:5555")

# 使用方式与本地数据库完全相同
db["key"] = "value"
print(db["key"])  # "value"

# 配置超时和重试
db = FlaxKV("remote_db", "127.0.0.1:5555",
            backend='remote',
            timeout=5000,      # 5秒超时
            max_retries=3,     # 最多重试3次
            retry_delay=0.1)   # 重试间隔100ms

使用场景：

• 🔹 多进程共享数据库
• 🔹 微服务架构中的中央缓存
• 🔹 分布式机器学习参数存储

13. Inspector 可视化工具

FlaxKV2 内置强大的数据可视化和管理工具，提供 CLI 和 Web UI 两种方式。

CLI 工具

# 查看所有键
flaxkv2 inspect keys mydb --path /data

# 查看键详情
flaxkv2 inspect get mydb user123 --path /data

# 统计分析
flaxkv2 inspect stats mydb --path /data

# 搜索键（支持正则表达式）
flaxkv2 inspect search mydb "user_.*" --path /data

# 删除键
flaxkv2 inspect delete mydb temp_key --path /data

# 设置键值
flaxkv2 inspect set mydb name "John" --path /data

Web UI

启动 Web 界面进行可视化管理：

Web UI 是独立包 flaxkv2-web（见仓库内 flaxkv2-web/），不随主包安装：

pip install flaxkv2-web

flaxkv2-web mydb --path /data --port 8080

# 然后访问 http://127.0.0.1:8080

Web UI 提供：

• 📂 数据浏览: 分页显示所有键值，搜索过滤
• 📊 统计分析: 类型分布、大小分布、TTL 状态可视化
• 🛠️ 数据管理: 在线增删改查，支持 TTL 设置

详见 Inspector 文档

14. 日志配置（作为基础库使用）

FlaxKV2 作为基础库，默认不输出任何日志，不会污染应用程序的终端。

from flaxkv2 import FlaxKV

# 默认完全静默
db = FlaxKV("mydb", "./data")
db["key"] = "value"  # 没有任何日志输出

# 需要调试时，手动启用日志
from flaxkv2.utils.log import enable_logging
enable_logging(level="INFO")  # 或 "DEBUG", "WARNING", "ERROR"

# 使用完后可以禁用
from flaxkv2.utils.log import disable_logging
disable_logging()

环境变量方式：

# 启用日志
export FLAXKV_ENABLE_LOGGING=1
export FLAXKV_LOG_LEVEL=DEBUG

python your_script.py

详见 日志配置文档

📚 API 参考

FlaxKV 类

FlaxKV(
    db_name: str,                          # 数据库名称
    root_path_or_url: str = ".",           # 本地路径或远程 URL（tcp://host:port）
    backend: str = None,                   # 'local' / 'remote'，None 自动检测
    auto_nested: bool = False,             # 启用嵌套存储
    rebuild: bool = False,                 # 重建数据库（删掉已有数据）
    raw: bool = False,                     # 原始模式（不序列化）
    default_ttl: int = None,               # 默认 TTL（秒）

    # LMDB 参数
    performance_profile: str = 'balanced', # 预设 map_size / max_readers
    map_size: int = None,                  # 覆盖 profile；写满自动翻倍
    max_readers: int = None,               # 覆盖 profile
    sub_db: str = None,                    # named sub-database（同 env 内物理隔离）

    # 缓存参数（传任一个即切到 CachedLmdbDict）—— 单位是条目数
    read_cache_size: int = None,           # 读缓存条目数
    write_buffer_size: int = None,         # 写缓冲条目数
    write_buffer_flush_interval: int = 30, # 定时刷新（秒）
    async_flush: bool = True,              # 异步 flush（更快，崩溃风险更大）

    # 持久性（仅本地后端）
    sync: bool = True,                     # 提交时 fsync 数据文件
    metasync: bool = True,                 # 提交时 fsync meta page

    # TTL 清理
    enable_ttl_cleanup: bool = True,
    cleanup_interval: int = 60,
    cleanup_batch_size: int = 1000,

    # 远程连接参数（仅 remote 后端）
    timeout: int = 5000,                   # 超时（毫秒）
    max_retries: int = 3,                  # 最大重试次数
    retry_delay: float = 0.1,              # 重试延迟（秒）
    enable_encryption: bool = False,       # CurveZMQ 加密
    password: str = None,                  # 服务器密码
    enable_compression: bool = False,      # LZ4 压缩
)

未知关键字参数直接 TypeError，不会被静默忽略。

字典操作

# 读写
db[key] = value              # 写入
value = db[key]              # 读取
value = db.get(key, default) # 安全读取
del db[key]                  # 删除

# 批量操作
db.batch_set({k1: v1, k2: v2})        # 批量写入（write transaction 原子操作）
values = db.batch_get([k1, k2, k3])   # 批量读取
db.update({k1: v1, k2: v2})           # 批量更新（write transaction）

# 查询
key in db                    # 检查存在
len(db)                      # 键数量（走 keys_count，不物化键列表）
db.keys_count()              # 同上，语义更明确

# ⚠️ 下面三个返回完整 list，大库会吃满内存，优先用 *_iter
db.keys()                    # 所有键（列表）
db.values()                  # 所有值（列表）
db.items()                   # 所有键值对（列表）
db.to_dict()                 # 转普通 dict（必然全量物化）

# 原子操作（单事务内完成读-改-写，可直接做队列认领）
db.pop(key, default)         # 取出并删除
db.setdefault(key, default)  # 不存在才写入

# 流式迭代（适合大数据库）
db.keys_iter(prefix=, start=, stop=, reverse=)
db.items_iter(prefix=, start=, stop=, reverse=)
db.values_iter(prefix=, start=, stop=, reverse=)

TTL 操作

db.set_ttl(key, ttl_seconds)      # 设置 TTL
remaining = db.get_ttl(key)       # 获取剩余时间
db.remove_ttl(key)                # 移除 TTL
db.cleanup_expired()              # 手动清理过期键

数据库维护

db.compact(start=, stop=)    # 范围压缩，回收磁盘空间
db.flush()                   # 把写缓冲推进 LMDB（无缓冲后端是空操作）
db.sync(force=True)          # 强制 fsync 落盘（内含 flush）
db.stat()                    # 统计信息
db.close()                   # 关闭数据库
FlaxKV.destroy(name, path)   # 彻底删除数据库（静态方法）

# 上下文管理器
with FlaxKV("mydb", "./data") as db:
    db["key"] = "value"

FlaxList

from flaxkv2 import FlaxList

lst = FlaxList(name, path)             # 普通模式（LMDB 后端）
lst = FlaxList(name, path, sorted=True) # sorted 模式（自动维护有序）

# 基本操作
lst.append(value)            # 追加，返回索引
lst.extend(values)           # 批量追加（write_batch），返回起始索引
lst[i]                       # 按索引读取（支持负索引）
lst[start:stop]              # 切片读取
lst.insert(i, value)         # 在索引 i 处插入
lst.pop(i=-1)                # 移除并返回元素
del lst[i]                   # 删除元素
len(lst)                     # 长度 O(1)

# 迭代
list(lst)                    # 转为列表
lst.iter_range(start, stop)  # 范围迭代（内存友好）

# 函数式查询（惰性求值，分批流式读取）
lst.where(predicate)         # 过滤，返回生成器
lst.first(predicate)         # 第一个匹配（短路）
lst.last(predicate)          # 最后一个匹配（反向短路）
lst.map(func)                # 映射变换，返回生成器
lst.batch(size)              # 按批次迭代
lst.any_match(predicate)     # 存在匹配（短路）
lst.all_match(predicate)     # 全部匹配（短路）

# 统计分析（O(1) 内存，单次遍历 Welford 算法）
s = lst.stats()              # 返回 ListStats(count, min, max, sum, mean, variance, std)
s = lst.stats(key=fn)        # 字典列表按字段统计
s = lst.stats(percentiles=[25,50,75])  # sorted 模式支持精确分位数

# 自定义聚合
lst.reduce(func, initial)    # 流式归约，O(1) 内存

# 随机操作
lst.shuffle()                # 原地 Fisher-Yates 洗牌
lst.sample(k)                # 随机抽样 k 个不重复元素，O(k) 内存
lst.choice()                 # 随机返回一个元素

# sorted 模式专用
lst.add(value)               # 二分查找插入（自动维护有序）

lst.close()

⚠️ 安全注意事项

1. Pickle 序列化风险

FlaxKV2 对复杂对象使用 pickle 序列化。pickle 存在安全风险：

# ⚠️ 危险：不要从不可信来源加载数据
# pickle 可以执行任意代码！

# ✅ 安全：只在可信环境中使用
db["trusted_data"] = my_custom_object

# ✅ 推荐：生产环境只存储简单类型
db["config"] = {"host": "localhost", "port": 8080}
db["users"] = ["alice", "bob", "charlie"]

最佳实践：

• ✅ 仅在可信环境中使用
• ✅ 生产环境优先使用 JSON、msgpack 等安全格式
• ✅ 对不可信数据进行验证
• ✅ 远程客户端默认 allow_pickle=False：需要 pickle 的值读写都抛 TypeError， 只有确认所有写入方可信时才传 allow_pickle=True

2. 远程连接安全

使用远程数据库时的安全考虑：

库名（db_name）在服务端只允许 [A-Za-z0-9_][A-Za-z0-9_.-]*，不能含路径 分隔符或以 . 开头。

生产环境推荐配置：

# 默认就只监听回环接口；需要对外暴露时才显式传 --host 0.0.0.0
flaxkv2 run --port 5555 --data-dir ./data

# 通过 SSH 隧道安全访问
ssh -L 5555:localhost:5555 user@remote-server

# 或使用防火墙规则限制访问
sudo ufw allow from 192.168.1.0/24 to any port 5555

📊 性能提示

1. 先开缓存，而不是调 profile

performance_profile 只设定 map_size 和 max_readers，不配置任何缓存。 真正决定读写性能的是缓存参数：

# 读多写少：开读缓存，热数据提升 ~3x（小 value；value 越大越明显）。
# 注意 FlaxKV 只传 read_cache_size 目前仍会附带异步写缓冲（0.3.5 的行为，已弃用，
# 会打 FutureWarning）；要读缓存 + 写入立即落盘，直接构造：
from flaxkv2 import CachedLmdbDict
db = CachedLmdbDict("cache", "./data", read_cache_size=50_000, enable_write_buffer=False)

# 写多读少：开写缓冲，N 次写合并成 1 次提交（也就是 1 次 fsync）
db = FlaxKV("logs", "./data", write_buffer_size=1_000)

# 大数据库（>100GB）：profile 负责的是 map_size 上限
db = FlaxKV("bigdata", "./data", performance_profile='large_database')

⚠️ 开了写缓冲必须正常 close()（或用 with），否则缓冲区数据会丢。

⚠️ read_cache_size 的单位是条目数，不是字节，缓存里存的还是反序列化后的 Python 对象。常驻内存 ≈ 条目数 × 单条对象大小，请按内存预算反推条数：单条 600KB 的库开 read_cache_size=1000 就是 600MB 常驻。不提供字节上限是因为 Python 对象的真实占用测不准（sys.getsizeof 只看顶层容器），给一个测不准的 上限比让你自己按数据大小算更糟。

ℹ️ 读缓存跨进程是安全的：每次读前校验一次 LMDB 的 env txnid（约 0.2µs）， 别的进程写过就整体失效重读。反过来说，写得很频繁的库开读缓存收益有限—— 别人每提交一次，本进程的读缓存就清一次。

2. 批量操作

# ❌ 慢：逐个写入
for i in range(10000):
    db[f"key{i}"] = f"value{i}"

# ✅ 快：批量写入
db.update({f"key{i}": f"value{i}" for i in range(10000)})

3. 使用嵌套存储

# ❌ 慢：每次修改都需要完整序列化
config = db["config"]  # 完整反序列化
config["port"] = 8080
db["config"] = config  # 完整序列化

# ✅ 快：只修改变化的部分
db = FlaxKV("mydb", "./data", auto_nested=True)
db["config"]["port"] = 8080  # 只序列化修改的值

4. TTL 自动清理

# TTL 清理是异步的，默认 60 秒一次
# 可以手动触发立即清理
db.cleanup_expired()

🔗 文档

索引：文档目录树

上手示例见 example.ipynb，性能测试脚本见 benchmarks/ 与 tests/benchmarks/。

🛠️ 依赖

核心依赖

• Python >= 3.10
• lmdb - LMDB Python 绑定
• msgpack - 高效序列化
• numpy - 数组支持
• pyzmq - ZeroMQ 网络通信
• cryptography / lz4 - 加密和压缩

可选依赖

• pandas - DataFrame 支持（pip install flaxkv2[pandas]）

🤝 贡献

欢迎贡献代码、报告问题或提出建议！

# Fork 项目
git clone https://github.com/KenyonY/flaxkv2.git
cd flaxkv2

# 安装开发依赖（含测试）
pip install -e .[test]

# 运行测试
pytest tests/

# 提交 Pull Request

📄 许可证

MIT License - 详见 LICENSE 文件

🌟 致谢

• 基于强大的 LMDB
• 使用 lmdb Python 绑定
• 网络通信基于 ZeroMQ