Electric 实时同步实战将 Postgres 变更自动同步到 Redis 缓存examples/redis 示例全解析【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric本篇技术指南以仓库examples/redis示例为主线讲解如何借助 Electric 的增量同步能力将 Postgres 中items表的每一次插入、更新、删除实时同步到本地 Redis Hash 缓存中。你将掌握 Electric 与 TypeScript 客户端ShapeStream的完整接入流程、Redis 侧合并部分更新的 Lua 脚本技巧、基于 pipeline 的事务化写入模式以及用redis-cli与psql验证端到端同步链路的具体方法可直接套用到任意Postgres 为源、Redis 为缓存的真实业务场景。核心思想由 Electric 自动托管缓存数据传统本地缓存的痛点在于失效管理数据变更后需要手动清理缓存、设置 TTL 过期时间或者引入额外的失效通知机制稍有遗漏就会产生脏数据。而本示例展示了 Electric 的核心理念——Electric 自动为你管理本地缓存中的数据当 Postgres 中的数据发生变更时变更会通过 Electric sync service 持续推送下来本地缓存这里的 Redis Hash随之自动更新你无需单独管理缓存失效cache invalidation也无需为缓存中的记录设置过期时间或 TTL——这些都由 Electric 在同步层替你完成。换句话说Redis 在这里并不是需要手工刷新的旁路缓存而是由ShapeStream客户端持续订阅、由 Electric 驱动的实时镜像缓存始终与 Postgres 中的 shape 数据保持一致。示例整体架构从源码结构看本示例的数据流是一条完整闭环Postgres (items 表) │ 逻辑复制 / WAL 捕获 ▼ Electric sync service (localhost:3000 /v1/shape) │ HTTP SSE / 增量日志 ▼ ShapeStream 客户端 (examples/redis/src/index.ts) │ insert / update / delete 消息 ▼ Redis Hash (items) ← 通过 pipeline 与 Lua 脚本写入涉及的仓库文件包括同步客户端入口examples/redis/src/index.ts数据库表结构迁移examples/redis/db/migrations/01-create_items_table.sql示例脚本定义examples/redis/package.json后端服务编排.support/docker-compose.ymlpnpm 工作区定义pnpm-workspace.yaml环境准备作为 pnpm workspace 的一分子构建本示例是 ElectricSQL monorepo 的一部分需要先作为 pnpm workspace 中的成员完成安装与构建。首先进入 monorepo 根目录cd ../../安装并构建全部 workspace 包与示例pnpm install pnpm run -r build说明pnpm run -r build会递归构建 workspace 中的每个包。本示例依赖electric-sql/clientworkspace 内部包与redis客户端二者均在 examples/redis/package.json 的dependencies中声明。构建完成后回到示例目录cd examples/redis启动后端服务Postgres Electric sync service使用 Docker Compose 一键拉起示例所需的全部后端服务pnpm backend:up该命令在 examples/redis/package.json 中实际展开为PROJECT_NAMEredis-example pnpm -C ../../ run example-backend:up pnpm db:migrate其中example-backend:up是 monorepo 根 package.json 中的聚合脚本先执行example-backend:down做清理再通过docker compose -f .support/docker-compose.yml up -d启动服务。后端服务构成根据 .support/docker-compose.yml后端包含两个容器服务镜像对外端口说明postgrespostgres:16-alpine54321:5432数据源数据库库名electric用户/密码postgres/passwordbackendelectricsql/electric:canary3000:3000Electric sync service暴露/v1/shape接口backend容器通过DATABASE_URLpostgresql://postgres:passwordpostgres:5432/electric?sslmodedisable连接 Postgres并设置了ELECTRIC_INSECURE: true——注意该配置仅适用于开发环境或已对 Electric API 做了其他安全加固的场景不适用于生产环境。Postgres 数据目录使用tmpfs挂载保证每次启动都是干净的数据库与磁盘状态。⚠️ 重要提示pnpm backend:up总是会先停止并删除其他示例后端容器挂载的卷。这保证了示例每次都以全新的数据库和干净的磁盘启动但如果你同时运行着其他示例请注意其数据会被清空。随后脚本会执行数据库迁移pnpm db:migrate即通过databases/pg-migrations应用./db/migrations目录下的迁移文件。当前唯一的迁移 01-create_items_table.sql 内容如下-- Create a simple items table. CREATE TABLE IF NOT EXISTS items ( id TEXT PRIMARY KEY NOT NULL, title TEXT NOT NULL ); -- Populate the table with 10 items. -- FIXME: Remove this once writing out of band is implemented WITH generate_series AS ( SELECT gen_random_uuid()::text AS id, foo AS title FROM generate_series(1, 10) ) INSERT INTO items (id, title) SELECT id, title FROM generate_series;它创建了一张结构极简的items表id主键 title并用generate_series预填了 10 条初始数据——这让你在启动客户端后能立刻在 Redis 中看到数据方便验证全链路。启动同步客户端后端就绪后启动开发服务器即同步客户端pnpm dev该命令对应 examples/redis/package.json 中的dev: tsx src/index.ts直接用tsx运行 TypeScript 入口。程序启动后会连接localhost:6379的 Redis清除 Hash 中可能存在的旧数据client.del(items)通过SCRIPT_LOAD预加载用于合并部分更新的 Lua 脚本创建订阅items表的ShapeStream开始将变更写入 Redis。用 redis-cli 实时观察 Redis 缓存连接 Redisredis-cli -h 127.0.0.1 -p 6379查看所有同步过来的 keyredis HKEYS items查看所有 key 及对应值items表中的每行记录都作为 Hash 的一个 fieldfield 值为该行数据的 JSON 序列化redis HGETALL items原 README 中此处写的是KGETALL items应为笔误Redis 的正确命令是HGETALL。持续观察所有到达的写入命令观察更新如何实时流入MONITOR端到端验证在 Postgres 中写数据在 Redis 中看到同步保持MONITOR或HGETALL运行的同时打开另一个终端连接 Postgres端口为54321psql postgresql://postgres:passwordlocalhost:54321/electric插入一条新数据并观察它实时同步进 Redisinsert into items (id, title) values (gen_random_uuid(), foo);这条INSERT经 Electric 捕获后会以一条insert变更消息推送给ShapeStream订阅回调最终通过HSET写入 Redis Hash。同理你还可以尝试update items set title bar where id (select id from items limit 1); delete from items where id (select id from items limit 1);分别验证update走 Lua 脚本做字段级合并与delete走HDEL的同步效果。源码深度解析变更消息如何落入 Redis1. 建立订阅ShapeStream 与变更消息入口代码 examples/redis/src/index.ts 使用electric-sql/client创建 shape 订阅import { createClient } from redis import { ShapeStream, Message, isChangeMessage } from electric-sql/client const itemsStream new ShapeStream({ url: http://localhost:3000/v1/shape, params: { table: items, }, }) itemsStream.subscribe(async (messages: Message[]) { ... })ShapeStream负责与 Electric sync service 建立连接、维护增量日志log游标并持续拉取变更其构造选项url、params等定义在 packages/typescript-client/src/client.ts 的ShapeStreamOptions中params.table指定订阅的表名对应服务端/v1/shape?tableitemsisChangeMessage类型守卫实现在 packages/typescript-client/src/helpers.ts用于从消息流中过滤出真正的数据变更消息跳过控制类消息每一条变更消息都携带headers.operationinsert/update/delete、key主键值与value变更后的行数据。2. 按操作类型分发写入回调中对三种操作分别处理switch (message.headers.operation) { case delete: pipeline.hDel(items, message.key) break case insert: pipeline.hSet(items, String(message.key), JSON.stringify(message.value)) break case update: { pipeline.evalSha(updateKeyScriptSha1, { keys: [items, String(message.key)], arguments: [JSON.stringify(message.value)], }) break } }insert整个新行JSON.stringify后HSET进 Hashfield 为iddelete按主键HDEL删除对应 fieldupdate不能简单地整体覆盖因为 shape 日志中的更新可能只携带变更后的部分字段partial update。示例用evalSha调用一段预加载的 Lua 脚本将新值合并进已有 JSON 中避免丢字段。3. Lua 脚本合并部分更新local current redis.call(HGET, KEYS[1], KEYS[2]) local parsed {} if current then parsed cjson.decode(current) end for k, v in pairs(cjson.decode(ARGV[1])) do parsed[k] v end local updated cjson.encode(parsed) return redis.call(HSET, KEYS[1], KEYS[2], updated)逻辑是取出 Hash 中该 key 当前的 JSON → 解析为 Lua table → 用变更消息中的字段逐键覆盖 → 重新编码后写回。脚本在启动时通过client.SCRIPT_LOAD(script)预加载拿到 SHA1 摘要后以EVALSHA调用省去每次传输脚本体的开销。4. pipeline一批变更一次事务提交所有命令先放入client.multi()创建的 pipeline 中最后统一pipeline.exec()执行作为一个 Redis 事务提交避免逐条命令的网络往返const pipeline client.multi() messages.forEach((message) { ... }) try { await pipeline.exec() console.log(Redis hash updated successfully with latest shape updates) } catch (error) { console.error(Error while updating hash:, error) }源码注释中留有一条 FIXME 提醒Redis 官方文档建议一次最多提交约 1 万条命令避免客户端为缓冲命令占用过多内存。对于大批量回放如 shape 首次全量加载场景可能需要分批执行 pipeline。停止服务演示结束后停止后端服务并清理挂载卷pnpm backend:down对应根 package.json 中的example-backend:down即docker compose -f .support/docker-compose.yml down --volumes——连同数据卷一并移除为下一次干净启动做好准备。小结与延伸通过examples/redis示例可以提炼出一条可复用的Postgres → Redis 实时缓存落地模式数据源建模定义带主键的 Postgres 表shape 的更新合并依赖主键订阅变更用ShapeStreamparams.table订阅 shape用isChangeMessage过滤变更消息落盘策略insert 直接覆盖、delete 直接移除、update 用 Lua 脚本做 JSON 字段级合并以容忍部分更新批量提交用 pipeline 将一批消息作为一个事务写入兼顾吞吐与原子性。这一模式的价值在于缓存层始终由 Electric 的增量日志驱动天然免去了手动缓存失效与 TTL 维护让 Redis 从易过期的旁路缓存升级为与数据库强一致的实时镜像。将该订阅逻辑替换为其他存储如其他 KV、搜索引擎或消息队列也只需改动落盘分支订阅与增量同步部分完全复用。【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考