用 SQLite 替代 Redis？轻量级缓存方案实战指南

前言

提到缓存，大家第一反应就是 Redis。但在很多中小项目中，引入 Redis 意味着额外的运维成本、内存开销和架构复杂度。其实对于日请求量在百万以下的服务，SQLite 完全可以胜任缓存角色。

本文带你用 SQLite 搭建一个轻量级缓存系统，适合个人项目、内部工具、嵌入式场景。

为什么考虑 SQLite 做缓存？

对比维度	Redis	SQLite
部署成本	需要独立进程	零部署，嵌入应用
持久化	可选（RDB/AOF）	天然持久化
并发性能	极高	WAL 模式下读并发优秀
内存占用	全内存	磁盘为主，内存可控
适用场景	高并发分布式	单机/嵌入式/中小项目

核心实现

1. 表结构设计

CREATE TABLE IF NOT EXISTS cache (
    key TEXT PRIMARY KEY,
    value BLOB,
    expires_at INTEGER,  -- Unix 时间戳
    created_at INTEGER DEFAULT (strftime('%s', 'now')),
    hit_count INTEGER DEFAULT 0
);

CREATE INDEX idx_cache_expires ON cache(expires_at);

关键点：

• key 作为主键，天然唯一
• expires_at 支持 TTL 过期机制
• hit_count 可选，用于统计热点数据

2. Python 封装示例

import sqlite3
import time
import json

class SQLiteCache:
    def __init__(self, db_path="cache.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("PRAGMA journal_mode=WAL")  # 关键！
        self.conn.execute("PRAGMA synchronous=NORMAL")
        self._create_table()
    
    def _create_table(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS cache (
                key TEXT PRIMARY KEY,
                value TEXT,
                expires_at INTEGER,
                created_at INTEGER DEFAULT (strftime('%s', 'now'))
            )
        """)
        self.conn.commit()
    
    def set(self, key, value, ttl=3600):
        expires_at = int(time.time()) + ttl
        self.conn.execute(
            "INSERT OR REPLACE INTO cache (key, value, expires_at) VALUES (?, ?, ?)",
            (key, json.dumps(value), expires_at)
        )
        self.conn.commit()
    
    def get(self, key):
        self._cleanup()  # 懒清理
        row = self.conn.execute(
            "SELECT value FROM cache WHERE key=? AND expires_at > ?",
            (key, int(time.time()))
        ).fetchone()
        return json.loads(row[0]) if row else None
    
    def delete(self, key):
        self.conn.execute("DELETE FROM cache WHERE key=?", (key,))
        self.conn.commit()
    
    def _cleanup(self):
        """每 100 次调用清理一次过期数据"""
        if not hasattr(self, "_call_count"):
            self._call_count = 0
        self._call_count += 1
        if self._call_count % 100 == 0:
            self.conn.execute(
                "DELETE FROM cache WHERE expires_at < ?",
                (int(time.time()),)
            )
            self.conn.commit()

3. WAL 模式是关键

PRAGMA journal_mode=WAL;

开启 WAL（Write-Ahead Logging）后：

• 读操作不阻塞写操作
• 多个读可以并发执行
• 写性能提升 5-10 倍

这是 SQLite 做缓存的前提条件，没有 WAL 就别想了。

进阶技巧

批量过期清理

不要每次 get 都清理，用定时任务：

import threading

def periodic_cleanup(cache, interval=60):
    def _run():
        while True:
            cache.conn.execute(
                "DELETE FROM cache WHERE expires_at < ?",
                (int(time.time()),)
            )
            cache.conn.commit()
            time.sleep(interval)
    t = threading.Thread(target=_run, daemon=True)
    t.start()

内存映射加速

对于高频读取场景，可以用 PRAGMA mmap_size 把热数据映射到内存：

PRAGMA mmap_size = 268435456;  -- 256MB

LRU 淘汰策略

加一个 last_access 字段，定期清理最久未访问的数据：

DELETE FROM cache WHERE key IN (
    SELECT key FROM cache ORDER BY last_access ASC LIMIT 1000
) AND (SELECT COUNT(*) FROM cache) > 100000;

性能实测

在 M1 MacBook 上测试（WAL 模式）：

• 写入: ~50,000 ops/sec（单条 INSERT OR REPLACE）
• 读取: ~200,000 ops/sec
• 批量写入: ~500,000 ops/sec（事务包裹）

对比 Redis 单机约 100,000 ops/sec（网络开销），SQLite 在本地场景下性能甚至更优。

什么时候不该用 SQLite 缓存？

• 分布式场景（多机共享缓存）
• 需要发布/订阅功能
• 数据量超过 10GB
• 需要复杂数据结构（Sorted Set、HyperLogLog 等）

总结

SQLite 缓存 = 零部署 + 持久化 + 够用的性能。

下次启动新项目时，先问自己：真的需要 Redis 吗？也许一个 SQLite 文件就够了。

工程的艺术不是用最强的工具，而是用最合适的工具。