PostgreSQL

你的数据库快爆了 凌晨 3 点，报警短信把你吵醒：数据库 CPU 100%，响应超时。 你揉着眼睛打开监控，订单表已经 2 亿行，每次查询都在全表扫描。加索引？早加过了。分表？业务代码要大改。 DBA 说：“该分库分表了。” 你打开《一致性实战（四）》，看到分片路由、跨分片查询聚合、Saga 事务补偿… 头皮发麻。 有没有更简单的方案？ 有。让我介绍 Citus——PostgreSQL 的分布式扩展。它让分片对应用几乎透明，大部分 SQL 不用改，ORM 和 sqlx 可以直接用。 Citus 是什么 想象你有一个图书馆，书越来越多，一个管理员忙不过来了。 传统方案（手动分片）： 你把书分到 4 个房间，每个房间一个管理员。但你得记住每本书在哪个房间，跨房间借书要自己协调。 Citus 方案： 还是 4 个房间 4 个管理员，但前台有一个总调度。你只跟总调度说话，它自动知道书在哪，跨房间的事它帮你协调。 这就是 Citus 的架构： ┌─────────────────────────────────────────────────────────┐ │ 应用程序 │ │ (sqlx, Diesel, 任何 PG 客户端) │ └─────────────────────┬───────────────────────────────────┘ │ 普通 PostgreSQL 协议 ▼ ┌─────────────────────────────────────────────────────────┐ │ Coordinator (协调器) │ │ 看起来就是一个普通 PostgreSQL │ │ • 解析 SQL │ │ • 路由到正确的分片 │ │ • 聚合跨分片结果 │ └───────┬─────────────┬─────────────┬─────────────────────┘ │ │ │ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ Worker1 │ │ Worker2 │ │ Worker3 │ │ (分片0-3)│ │ (分片4-7)│ │(分片8-11)│ └─────────┘ └─────────┘ └─────────┘ 关键点：应用只连接 Coordinator，它看起来、用起来就是一个普通 PostgreSQL。 五分钟上手 1. 启动 Citus 集群 用 Docker Compose 快速启动一个本地集群： # docker-compose.yml services: coordinator: image: citusdata/citus:12.1 container_name: citus_coordinator ports: - "5432:5432" environment: POSTGRES_USER: postgres POSTGRES_PASSWORD: postgres POSTGRES_DB: mydb healthcheck: test: ["CMD-SHELL", "pg_isready -U postgres"] interval: 5s timeout: 5s retries: 5 networks: - citus-network worker1: image: citusdata/citus:12.1 container_name: citus_worker1 environment: POSTGRES_USER: postgres POSTGRES_PASSWORD: postgres healthcheck: test: ["CMD-SHELL", "pg_isready -U postgres"] interval: 5s timeout: 5s retries: 5 networks: - citus-network worker2: image: citusdata/citus:12.1 container_name: citus_worker2 environment: POSTGRES_USER: postgres POSTGRES_PASSWORD: postgres healthcheck: test: ["CMD-SHELL", "pg_isready -U postgres"] interval: 5s timeout: 5s retries: 5 networks: - citus-network networks: citus-network: driver: bridge 启动集群并等待健康检查通过： ...

从这里开始 经过缓存优化后，小明的二手书平台性能有了很大提升。但新的问题出现了：数据量太大，单个数据库快撑不住了。 订单表已经有上亿条数据，每次查询都很慢，添加索引也无济于事。磁盘空间也快满了，单机存储已经到达瓶颈。 DBA 看了监控后说：“是时候分库分表了。” 小明决定按 user_id 将订单数据分到 4 个数据库分片中： user_id % 4 = 0 → shard_0 user_id % 4 = 1 → shard_1 user_id % 4 = 2 → shard_2 user_id % 4 = 3 → shard_3 分片后，单个用户的订单查询飞快了。但很快遇到了新问题： 问题一：跨分片查询 “给我查最近 7 天所有用户的订单总额。“运营说。 这意味着要查询所有 4 个分片，然后合并结果。 问题二：跨分片事务 “用户 A（在 shard_0）想把一本书转让给用户 B（在 shard_1）。” 这涉及两个分片的数据修改，如何保证原子性？ 问题三：分片键变更 “用户要改手机号，但我们是按手机号分片的…” 分片键变更意味着数据要迁移到另一个分片。 这就是数据分片带来的一致性挑战。 分片基础 分片路由 首先实现一个分片路由器： use sqlx::PgPool; use std::collections::HashMap; use std::sync::Arc; /// 分片路由器 pub struct ShardRouter { shards: Vec<Arc<PgPool>>, shard_count: usize, } impl ShardRouter { pub fn new(shard_pools: Vec<PgPool>) -> Self { let shard_count = shard_pools.len(); Self { shards: shard_pools.into_iter().map(Arc::new).collect(), shard_count, } } /// 根据分片键获取分片 pub fn get_shard(&self, shard_key: i64) -> &PgPool { let shard_index = (shard_key as usize) % self.shard_count; &self.shards[shard_index] } /// 获取所有分片（用于跨分片查询） pub fn all_shards(&self) -> &[Arc<PgPool>] { &self.shards } /// 根据分片键计算分片索引 pub fn shard_index(&self, shard_key: i64) -> usize { (shard_key as usize) % self.shard_count } } 基本的分片读写 pub struct ShardedOrderRepository { router: ShardRouter, } impl ShardedOrderRepository { /// 写入订单（路由到对应分片） pub async fn create_order(&self, order: &Order) -> Result<(), Error> { let pool = self.router.get_shard(order.user_id); sqlx::query!( r#" INSERT INTO orders (id, user_id, book_id, amount, status, created_at) VALUES ($1, $2, $3, $4, $5, NOW()) "#, order.id, order.user_id, order.book_id, order.amount, order.status, ) .execute(pool) .await?; Ok(()) } /// 查询用户订单（单分片查询） pub async fn get_user_orders(&self, user_id: i64) -> Result<Vec<Order>, Error> { let pool = self.router.get_shard(user_id); let orders = sqlx::query_as!(Order, "SELECT * FROM orders WHERE user_id = $1 ORDER BY created_at DESC", user_id ) .fetch_all(pool) .await?; Ok(orders) } } 问题一：跨分片查询 当查询条件不包含分片键时，需要查询所有分片。 ...

从这里开始 小明的二手书平台在读写分离上线后，数据库终于稳定了。但好景不长，运营同学兴奋地告诉他：“我们要搞一次促销活动！” 活动当天，流量暴涨。虽然有了从库分担读请求，但热门书籍的详情页依然让数据库不堪重负。监控显示，相同书籍的查询每秒重复了上万次。 “这些数据明明很少变化，为什么每次都要查数据库？“小明决定引入 Redis 缓存。 代码很快写好了： async fn get_book(pool: &PgPool, redis: &RedisClient, book_id: i64) -> Result<Book> { // 先查缓存 if let Some(cached) = redis.get(&format!("book:{}", book_id)).await? { return Ok(serde_json::from_str(&cached)?); } // 缓存未命中，查数据库 let book = sqlx::query_as!(Book, "SELECT * FROM books WHERE id = $1", book_id) .fetch_one(pool) .await?; // 写入缓存 redis.set_ex(&format!("book:{}", book_id), &serde_json::to_string(&book)?, 3600).await?; Ok(book) } 促销当天，一切看起来都很顺利。直到客服开始收到投诉：“我明明改了价格，为什么页面显示的还是旧的？” 小明查看日志，发现问题了：卖家更新了价格，数据库确实变了，但缓存里还是旧数据。用户看到的是缓存中的过期信息。 数据同时存在于两个地方，它们之间失去了同步——这就是缓存一致性问题。 问题的本质 缓存一致性问题的根源在于：数据库和缓存是两个独立的存储系统，对它们的操作无法在同一个事务中完成。 无论你以什么顺序操作，都可能出问题： 先更新数据库，再更新缓存： 时刻1: 请求A 更新数据库 price=100 时刻2: 请求B 更新数据库 price=200 时刻3: 请求B 更新缓存 price=200 时刻4: 请求A 更新缓存 price=100 // 并发导致缓存最终是旧值！ 先更新缓存，再更新数据库： 时刻1: 请求A 更新缓存 price=100 时刻2: 请求A 更新数据库失败，回滚 时刻3: 缓存中 price=100，但数据库还是旧值 // 脏数据！ 先删除缓存，再更新数据库： 时刻1: 请求A 删除缓存 时刻2: 请求B 读取，缓存未命中，从数据库读取旧值 时刻3: 请求B 将旧值写入缓存 时刻4: 请求A 更新数据库 // 缓存中又是旧值！ 先更新数据库，再删除缓存（最常用）： 时刻1: 请求A 读取，缓存未命中 时刻2: 请求A 从数据库读取 price=100 时刻3: 请求B 更新数据库 price=200 时刻4: 请求B 删除缓存 时刻5: 请求A 将 price=100 写入缓存 // 又是旧值！ 看起来无论怎么做都不对？别急，让我们看看业界是如何解决的。 方案一：Cache-Aside（旁路缓存） 这是最经典、最广泛使用的方案。核心思想是：读时填充缓存，写时只删除缓存。 为什么"删除"而不是"更新”？ 删除缓存的好处是： 避免并发写冲突：删除是幂等的，多次删除和一次删除效果相同 简化逻辑：不需要计算新值，让下一次读请求自然地填充正确的值 避免无效更新：如果数据更新后根本没人读，更新缓存就是浪费 实现 use redis::AsyncCommands; use sqlx::PgPool; use serde::{Deserialize, Serialize}; #[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)] pub struct Book { pub id: i64, pub title: String, pub price: i64, // 以分为单位 pub stock: i32, pub updated_at: chrono::DateTime<chrono::Utc>, } pub struct BookService { pool: PgPool, redis: redis::Client, } impl BookService { /// 读取书籍信息（Cache-Aside 模式） pub async fn get_book(&self, book_id: i64) -> Result<Book, Error> { let cache_key = format!("book:{}", book_id); let mut conn = self.redis.get_multiplexed_async_connection().await?; // 1. 先查缓存 let cached: Option<String> = conn.get(&cache_key).await?; if let Some(json) = cached { return Ok(serde_json::from_str(&json)?); } // 2. 缓存未命中，查数据库 let book = sqlx::query_as!(Book, "SELECT * FROM books WHERE id = $1", book_id) .fetch_one(&self.pool) .await?; // 3. 填充缓存（设置过期时间作为兜底） let json = serde_json::to_string(&book)?; conn.set_ex(&cache_key, &json, 3600).await?; Ok(book) } /// 更新书籍价格（先更新数据库，再删除缓存） pub async fn update_price(&self, book_id: i64, new_price: i64) -> Result<(), Error> { // 1. 更新数据库 sqlx::query!( "UPDATE books SET price = $1, updated_at = NOW() WHERE id = $2", new_price, book_id ) .execute(&self.pool) .await?; // 2. 删除缓存 let cache_key = format!("book:{}", book_id); let mut conn = self.redis.get_multiplexed_async_connection().await?; conn.del(&cache_key).await?; Ok(()) } } 仍然存在的问题 前面我们分析过，“先更新数据库，再删除缓存"在极端并发下仍可能出现不一致： ...

用户刚改完昵称，刷新页面发现还是旧的。这不是 Bug，这是主从延迟。读写分离带来的第一个一致性问题，你准备好了吗？ ...

当你的系统只有一台机器、一个数据库时，一致性问题几乎不存在。数据库事务是你的保护伞。但这种幸福，注定是短暂的。 ...

回顾这个系列： 关系型数据库解决了数据一致性 NoSQL 解决了扩展性和灵活性 嵌入式数据库解决了零部署需求 分布式数据库解决了单机瓶颈 分析型数据库解决了海量数据分析 时序数据库解决了时间序列场景 向量数据库解决了语义搜索 每个问题都催生出新的数据库类型。而 PostgreSQL 选择了一条不同的路：成为一个可以被无限扩展的平台。 ...

“找出所有与这张图片相似的商品。” “搜索语义上相关的文档，而不只是关键词匹配。” “推荐用户可能喜欢的内容。” 这些需求有什么共同点？它们都需要找**“相似”而不是“相等”**的数据。传统数据库在这里彻底无能为力了。 ...

你的服务器每秒产生上千条监控指标。用户行为日志像瀑布一样涌入。IoT 设备 7×24 小时不停地汇报传感器数据。 传统数据库处理这些数据时，已经开始喘不过气了。 ...

1970 年，IBM 研究员 Edgar F. Codd 发表了一篇论文：《A Relational Model of Data for Large Shared Data Banks》。 这篇论文没有写任何代码，却定义了接下来 50 年数据库的基本形态。 为什么一个 50 年前的模型，至今仍是主流？ ...