很多项目在早期只依赖SQLite做本地持久化，单机读写完全够用。但当业务扩展到多设备、多节点时，本地库之间的数据割裂就会暴露出来。一个常见的做法是把共享数据搬到远程数据库，但全部请求都走远程又会增加延迟和网络开销。Tencent Cloud Tendis作为一个兼容Redis协议的云上缓存产品，非常适合充当SQLite前面的共享缓存层。这样既保留了SQLite的本地可靠性，又能通过Tendis实现跨设备的数据交换和热点数据加速。
本文会围绕一个实际项目展开：一个门店收银终端，每个终端本地有一个SQLite文件存储订单明细，同时需要把库存和会员信息同步到云端。如果每次查询都直接访问云数据库，网络抖动会明显拖慢收银流程。于是我们引入Tencent Cloud Tendis作为远程缓存，本地SQLite作为一级缓存，Tendis作为二级缓存，形成读多写少的加速结构。
为什么要把SQLite和Tendis放在一起用
SQLite的优势在于零配置、嵌入式、事务完整，非常适合单机场景。它的缺陷也很明显：只能被本地进程访问，无法直接被其他机器共享。而Tendis兼容Redis协议，支持丰富的数据结构，部署在腾讯云上可以跨可用区访问。把两者结合，本质上是在构建一个分层存储：SQLite负责最终落盘和离线可用，Tendis负责在线共享和高速读写。
从读写路径看，读请求先查SQLite，如果本地有数据就直接返回；如果本地没有或者数据已过期，再查Tendis，拿到结果后写回SQLite。这样大部分重复查询都能命中本地库，只有冷数据或需要共享的数据才会走到远程。写请求则相反，先写SQLite保证本地持久化，再异步写入Tendis，让其他终端能尽快看到更新。
这种组合并不适合所有项目。如果数据量极大且需要复杂SQL关联查询，Tendis作为纯缓存无法替代关系型数据库。它的定位是加速热数据访问和跨节点共享，不是取代SQLite的存储职责。理解这一点后，再决定哪些数据放进Tendis，哪些只留在本地。
环境准备与基础连接代码
先在Python环境里安装依赖。SQLite使用标准库sqlite3，不需要额外安装。Tendis推荐使用redis-py客户端，因为它完全兼容Redis协议。安装命令如下：
pip install redis
接下来初始化SQLite数据库。为了提升并发读写性能，连接时开启WAL模式，并设置合理的busy_timeout，避免多个线程同时写入时直接报错。
import sqlite3
import redis
import json
import time
def init_sqlite(db_path):
conn = sqlite3.connect(db_path, check_same_thread=False)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=5000")
conn.execute("""
CREATE TABLE IF NOT EXISTS product_cache (
product_id TEXT PRIMARY KEY,
name TEXT,
stock INTEGER,
price REAL,
updated_at REAL
)
""")
conn.commit()
return conn连接Tencent Cloud Tendis时，需要用到实例的内网地址、端口和密码。如果项目和Tendis实例在同一个VPC内，使用内网地址可以避免公网流量费用和额外延迟。
def init_tendis(host, port, password):
pool = redis.ConnectionPool(
host=host,
port=port,
password=password,
db=0,
decode_responses=True,
socket_connect_timeout=5,
socket_timeout=5,
max_connections=20
)
return redis.Redis(connection_pool=pool)这里把decode_responses设为True，让返回数据自动从字节转为字符串，方便后续处理。连接池限制为20，防止突发流量打满Tendis的连接数。如果终端数量较多，可以适当调大，但要参考实例规格。
读写路径与数据同步策略
先看读路径的实现。核心逻辑是本地优先，远程兜底。为了避免频繁穿透到Tendis，可以在SQLite里记录一个过期时间戳，超过该时间就认为本地数据可能不是最新，需要去Tendis确认。
CACHE_TTL = 60
def get_product(conn, tendis, product_id):
row = conn.execute(
"SELECT name, stock, price, updated_at FROM product_cache WHERE product_id = ?",
(product_id,)
).fetchone()
now = time.time()
if row and (now - row[3]) < CACHE_TTL:
return {
"product_id": product_id,
"name": row[0],
"stock": row[1],
"price": row[2],
"source": "sqlite"
}
data = tendis.get(f"product:{product_id}")
if data:
product = json.loads(data)
conn.execute(
"""
INSERT OR REPLACE INTO product_cache
(product_id, name, stock, price, updated_at)
VALUES (?, ?, ?, ?, ?)
""",
(product_id, product["name"], product["stock"], product["price"], now)
)
conn.commit()
product["source"] = "tendis"
return product
return None这个函数先用本地时间戳判断SQLite数据是否新鲜。如果本地数据在60秒内更新过，直接返回。否则查询Tendis，命中后写回SQLite，这样下一次查询就能走本地。若两边都没有，返回None，由上层业务处理。
写路径需要保证本地持久化优先，同时尽量把变更传播到Tendis。如果直接同步写入Tendis，写延迟会增加；如果异步写入，又可能丢失更新。实际项目里可以采用先写SQLite，再通过后台队列异步同步到Tendis的方式。
import threading
def update_product(conn, tendis, product_id, name, stock, price):
now = time.time()
conn.execute(
"""
INSERT OR REPLACE INTO product_cache
(product_id, name, stock, price, updated_at)
VALUES (?, ?, ?, ?, ?)
""",
(product_id, name, stock, price, now)
)
conn.commit()
payload = json.dumps({
"name": name,
"stock": stock,
"price": price,
"updated_at": now
})
def sync_to_tendis():
tendis.set(f"product:{product_id}", payload)
tendis.expire(f"product:{product_id}", 3600)
threading.Thread(target=sync_to_tendis, daemon=True).start()这里把Tendis的写入放到子线程里执行，主线程只负责SQLite写入和返回。这样收银终端的响应时间不会因为远程缓存抖动而变长。Tendis里设置1小时过期，避免长期不活跃的数据占用内存。
有一点需要特别注意：SQLite的commit操作在WAL模式下比较快，但如果频繁单条写入，仍然会产生大量小事务。对于批量导入或批量更新，建议使用事务包裹，减少磁盘同步次数。
性能优化与实战避坑
SQLite侧的性能优化主要集中在WAL模式、事务批量和索引设计上。WAL模式允许读操作不阻塞写操作，适合读多写少的缓存场景。批量写入时，把多条INSERT放在同一个事务里提交，性能可以提升数倍。缓存表的主键是product_id，如果还要按名称查询，需要给name字段加索引。
conn.execute("CREATE INDEX IF NOT EXISTS idx_product_name ON product_cache(name)")Tendis侧的优化则要关注连接池和管道。连接池避免频繁建立TCP连接，管道可以把多个命令打包发送，减少网络往返。对于需要批量读取的场景，用pipeline一次性获取多个key。
def get_products_batch(tendis, product_ids):
pipe = tendis.pipeline()
for pid in product_ids:
pipe.get(f"product:{pid}")
return pipe.execute()另外，Tendis和SQLite的数据一致性不可能做到强一致。我们的方案偏向最终一致：写SQLite成功后，异步同步到Tendis可能在几十毫秒后才完成。如果业务要求读到的库存必须实时准确，就不能依赖这种缓存架构，需要增加版本号或直接查源数据库。在收银场景里，库存短暂不一致影响可控，换取的是整体响应速度。
还要注意Tendis的key设计。为了避免不同业务数据冲突，key加上前缀，比如product:、member:、order:。如果数据体积大，避免把整个对象存成一个大字符串，尽量拆成字段哈希，方便局部更新。
最后，监控和降级不能少。当Tendis不可用时，本地SQLite仍然可以独立工作，只是拿不到最新共享数据。代码里要捕获redis连接异常，打印日志，但不能让异常影响主流程。可以设置一个开关，检测到Tendis连续失败后自动切换为纯本地模式，等连接恢复后再切回来。
def get_product_safe(conn, tendis, product_id):
try:
return get_product(conn, tendis, product_id)
except redis.RedisError:
row = conn.execute(
"SELECT name, stock, price FROM product_cache WHERE product_id = ?",
(product_id,)
).fetchone()
if row:
return {
"product_id": product_id,
"name": row[0],
"stock": row[1],
"price": row[2],
"source": "sqlite_fallback"
}
return None这个降级函数在Tendis异常时直接读取本地SQLite，不抛出错误。对收银终端来说，哪怕云端缓存挂了，本地历史数据仍然可用，只是库存可能不是最新的。配合后台重连机制，可以在Tendis恢复后自动同步。
综合来看，SQLite与Tencent Cloud Tendis的结合非常适合边缘计算、移动端、门店终端等场景。它不追求强一致，而是用本地可靠存储和远程共享缓存的组合，换取低延迟和高可用。只要理清读写路径、做好过期与降级策略，这套方案能在实际项目中稳定运行。
SQLiteTencent Cloud Tendis二级缓存修改时间：2026-09-17 21:19:37