嵌入式

SQLite 是链接进应用的数据库引擎，而不是单独运行的数据库服务器。主数据库文件以可移植格式保存模式、表和索引，因此适合本地应用状态、嵌入式设备，以及写入负载符合其并发模型的服务。这个文件依然受事务与锁管理，不能只凭普通文件复制的直觉来操作它，否则容易把方便的存储形式误解成没有数据库运行规则。 坏掉的样子 主文件并不总是当前数据库状态的全部。日志模式可以让已经提交的变更仍位于独立 WAL 文件中，而主文件停留在更早的检查点。连接之间的共享内存协调通常还会使用另一个旁路文件。因此，连接保持活动时只复制主文件，可能漏掉已经提交的数据；依次复制几个正在变化的文件，也不会自动组成一致快照。 日志模式改善读写之间的重叠执行，但仍然只允许一个写者。一个长写事务可能让另一个写者收到数据库忙错误。等待更久只是增加耐心，不会增加同时写入者的数量。应缩短事务，并在实际测得的写入并发或远程访问要求超出这种架构的适用范围时，选择有独立服务端的数据库，而不是一味调大超时。 读者也有快照边界。另一个连接持有未提交更新时，读者仍能看见较旧的已提交值。这体现的是事务隔离，并不证明 SQLite 丢失了那次修改。错误之后，应用应检查事务状态，只有操作语义允许时才重试适当范围的事务。不能把所有忙错误、旧值和提交失败都归为同一种可以无限重试的现象。 快速验证 把下面内容保存为 sqlite_lab.py，使用普通 python3 运行并保持断言开启。它只使用标准库和本地临时文件。本次环境是 Python 3.14.4 搭配 SQLite 3.53.0；可以读取模块的引擎版本属性确认，不能用 Python 版本替代实际链接的数据库版本。所有文件都由实验自己创建，不需要连接任何已有数据库。 import shutil import sqlite3 import tempfile from contextlib import closing from pathlib import Path with tempfile.TemporaryDirectory(prefix="xbcatm-sqlite-") as directory: root = Path(directory) database = root / "source.db" with closing(sqlite3.connect(database, isolation_level=None, timeout=0.05)) as first, \ closing(sqlite3.connect(database, isolation_level=None, timeout=0.05)) as second: assert first.execute("PRAGMA journal_mode=WAL").fetchone() == ("wal",) first.execute("PRAGMA wal_autocheckpoint=0") second.execute("PRAGMA wal_autocheckpoint=0") first.execute("CREATE TABLE marker (id INTEGER PRIMARY KEY, value INTEGER)") first.execute("INSERT INTO marker VALUES (1,10)") assert first.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone() == (0, 0, 0) first.execute("BEGIN IMMEDIATE") first.execute("UPDATE marker SET value=99 WHERE id=1") assert second.execute("SELECT value FROM marker").fetchone() == (10,) try: second.execute("BEGIN IMMEDIATE") except sqlite3.OperationalError as error: assert error.sqlite_errorcode == sqlite3.SQLITE_BUSY assert not second.in_transaction print("second_writer", error.sqlite_errorname) else: raise AssertionError("second writer unexpectedly acquired the lock") first.execute("ROLLBACK") second.execute("BEGIN IMMEDIATE") second.execute("UPDATE marker SET value=20 WHERE id=1") second.execute("COMMIT") assert first.execute("SELECT value FROM marker").fetchone() == (20,) # Deliberately unsafe copy, only as a negative control in this quiet fixture. shutil.copyfile(database, root / "main-only.db") with closing(sqlite3.connect(root / "main-only.db")) as incomplete: assert incomplete.execute("PRAGMA integrity_check").fetchone() == ("ok",) assert incomplete.execute("SELECT value FROM marker").fetchone() == (10,) print("main_only_copy", "integrity ok, stale value 10") with closing(sqlite3.connect(root / "backup.db")) as backup: first.backup(backup) assert backup.execute("PRAGMA integrity_check").fetchone() == ("ok",) assert backup.execute("SELECT value FROM marker").fetchone() == (20,) first.execute("UPDATE marker SET value=30 WHERE id=1") assert backup.execute("SELECT value FROM marker").fetchone() == (20,) print("online_backup", "value 20, source now 30") 夹具先建立已经完成检查点的值十，然后开启写事务，把它改成九十九但不提交。另一个连接仍然读到十；它试图成为写者时，必须得到数据库忙错误，并且仍然处于事务之外。脚本随后回滚第一个写者，让第二个连接取得写入位置，再成功提交二十。这同时观察到写入竞争与回滚后的正常推进。 ...