演进全景 让我们回顾小明的二手书平台从诞生到成熟的完整演进路径： 第一阶段：单机时代 └── 问题：并发修改导致数据错乱 └── 方案：数据库事务、隔离级别、乐观锁/悲观锁 第二阶段：读写分离 └── 问题：从库延迟导致读取过期数据 └── 方案：强制读主、时间窗口、会话级路由、同步复制 第三阶段：引入缓存 └── 问题：数据库与缓存数据不一致 └── 方案：Cache-Aside、延迟双删、CDC、LISTEN/NOTIFY 第四阶段：数据分片 └── 问题：跨分片查询和事务 └── 方案：并行查询、全局索引、CQRS、Saga、消息驱动 第五阶段：服务拆分 └── 问题：跨服务操作无法保证原子性 └── 方案：TCC、Saga、本地消息表、AT 模式 第六阶段：消息驱动 └── 问题：消息丢失、重复、乱序 └── 方案：本地消息表、幂等消费、顺序保证、死信队列 每一个阶段都是为了解决上一阶段的瓶颈，但同时也引入了新的一致性挑战。没有银弹，只有权衡。 一致性问题分类 按数据存储分类 类型 问题描述 典型场景 章节 单存储并发 多个请求同时修改同一数据 库存扣减、余额变更 第一章 主从复制延迟 从库数据落后于主库 读写分离架构 第二章 多存储不一致 数据库与缓存数据不同步 缓存架构 第三章 跨分片一致性 分片间数据操作不一致 分库分表架构 第四章 跨服务事务 多服务操作无法原子提交 微服务架构 第五章 消息可靠性 消息丢失/重复/乱序 事件驱动架构 第六章 按一致性强度分类 强一致性 最终一致性 │ │ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ 单机事务 │ │ 同步复制 │ │ TCC │ │ Saga │ │ 消息驱动 │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ │ │ │ │ 性能最低 性能较低 性能中等 性能较高 性能最高 解决方案速查表 单机并发问题 方案 原理 优点 缺点 适用场景 数据库事务 ACID 保证 简单可靠 单机限制 单数据库场景 悲观锁 SELECT FOR UPDATE 强一致 阻塞等待 冲突频繁 乐观锁 版本号检查 无阻塞 需要重试 冲突较少 SERIALIZABLE 最高隔离级别 无并发问题 性能差 金融核心 读写分离一致性 方案 原理 延迟容忍 性能影响 适用场景 强制读主 关键操作读主库 零延迟 高 余额查询 时间窗口 写后短时间读主 秒级 中 普通业务 会话级路由 同会话读主 会话内零延迟 中 用户体验敏感 同步复制 等待从库确认 零延迟 高 强一致要求 业务容忍 接受延迟 秒级 无 非关键数据 缓存一致性 方案 原理 一致性强度 实现复杂度 适用场景 Cache-Aside 先更DB再删缓存 最终一致 低 通用场景 延迟双删 二次删除 最终一致（更强） 中 高一致要求 Write-Through 同时写DB和缓存 最终一致 中 缓存托管 Write-Behind 异步批量写DB 弱一致 高 高写入场景 CDC 监听DB变更 最终一致 高 无侵入需求 LISTEN/NOTIFY PG 通知机制 最终一致 低 PostgreSQL 跨服务事务 方案 原理 隔离性 性能 实现复杂度 适用场景 TCC 预留-确认-取消 好 中 高 短事务、金融 Saga 正向+补偿 差 高 中 长事务 本地消息表 事务写消息 差 高 低 异步场景 AT 模式 自动undo log 中 低 高 低侵入需求 消息可靠性 问题 方案 原理 消息丢失 本地消息表 事务保证写入 消息丢失 事务消息 半消息机制 消息重复 幂等消费 去重+业务幂等 消息乱序 单分区 同 key 同分区 消息乱序 序列号检查 消费端校验 消费失败 死信队列 隔离问题消息 跨分片一致性 问题 方案 一致性 性能 跨分片查询 并行查询 强一致 差 跨分片查询 全局索引 最终一致 中 跨分片查询 CQRS 最终一致 好 跨分片事务 2PC 强一致 差 跨分片事务 Saga 最终一致 好 跨分片事务 消息驱动 最终一致 好 选型决策树 总体决策流程 遇到一致性问题 │ ├─ 1. 先问：真的需要强一致吗？ │ ├─ 金融、交易、核心业务 → 需要 │ └─ 统计、展示、非核心 → 最终一致即可 │ ├─ 2. 再问：能简化架构吗？ │ ├─ 能不拆服务就不拆 │ ├─ 能不分库就不分 │ └─ 能单机解决就单机 │ └─ 3. 最后：选择合适的方案 └─ 见下方详细决策树 详细决策树 ┌─────────────────────────────────────────────────────────────┐ │ 一致性问题决策树 │ └─────────────────────────────────────────────────────────────┘ 问题类型是什么？ │ ├─ 单库并发问题 │ ├─ 冲突频率高？ → 悲观锁 (SELECT FOR UPDATE) │ ├─ 冲突频率低？ → 乐观锁 (版本号) │ └─ 极端要求？ → SERIALIZABLE 隔离级别 │ ├─ 读写分离延迟 │ ├─ 能容忍秒级延迟？ → 业务容忍 │ ├─ 需要写后即读一致？ → 时间窗口读主 │ ├─ 整个会话需要一致？ → 会话级路由 │ └─ 绝对不能延迟？ → 同步复制（慎用） │ ├─ 缓存不一致 │ ├─ 通用场景？ → Cache-Aside │ ├─ 需要更强一致？ → 延迟双删 │ ├─ 不能改代码？ → CDC │ └─ 用 PostgreSQL？ → LISTEN/NOTIFY │ ├─ 跨服务事务 │ ├─ 需要资源隔离？ → TCC │ ├─ 是长事务？ → Saga │ ├─ 可以异步？ → 本地消息表 │ └─ 要低侵入？ → AT 模式（如 Seata） │ ├─ 消息可靠性 │ ├─ 担心丢失？ → 本地消息表 + At-Least-Once │ ├─ 担心重复？ → 幂等消费 │ ├─ 担心乱序？ → 单分区 + 序列号 │ └─ 担心堆积？ → 死信队列 + 告警 │ └─ 跨分片问题 ├─ 查询问题 │ ├─ 查询少？ → 并行查询所有分片 │ ├─ 查询复杂？ → CQRS + 查询库 │ └─ 需要实时？ → 全局索引 └─ 事务问题 ├─ 能避免跨分片？ → 调整分片策略 ├─ 必须强一致？ → 2PC（慎用） └─ 可最终一致？ → Saga 或消息驱动 常见误区 误区一：追求强一致性 很多时候，最终一致性就够了。强一致性的代价是： ...

从这里开始 上一章，小明使用本地消息表解决了跨服务事务问题。随着系统演进，消息队列已经成为服务间通信的核心基础设施。 但新的问题接踵而至。一天早上，运营同学报告了一个奇怪的现象：“有个用户的订单被重复扣款了三次！” 小明查看日志，发现消息确实发送了三次。原因是消息发送后，在收到 broker 确认之前网络超时了，发送端以为失败了就重试。 “那我加个去重逻辑？“小明想。但很快又发现另一个问题：有些消息根本没发出去就丢了。 消息队列不是银弹，它本身也有一致性问题需要解决。 消息投递的三种语义 消息队列的投递保证分为三种： At-Most-Once（最多一次） 消息发送后不等待确认，可能丢失，但不会重复。 // At-Most-Once：发了就不管 async fn send_at_most_once(producer: &KafkaProducer, message: &Message) { // 发送后立即返回，不等待确认 producer.send_fire_and_forget(message).await; // 消息可能丢失，但代码继续执行 } 适用场景：日志收集、监控数据等允许丢失的场景 At-Least-Once（至少一次） 消息发送后等待确认，失败则重试，保证不丢但可能重复。 // At-Least-Once：确认收到才算成功 async fn send_at_least_once( producer: &KafkaProducer, message: &Message, ) -> Result<(), Error> { loop { match producer.send(message).await { Ok(()) => return Ok(()), Err(e) if e.is_retriable() => { tracing::warn!("Send failed, retrying: {}", e); tokio::time::sleep(Duration::from_millis(100)).await; // 继续重试，可能导致重复 } Err(e) => return Err(e), } } } 这是最常用的语义，因为重复可以通过消费端幂等来解决，但丢失无法恢复。 Exactly-Once（恰好一次） 消息既不丢失也不重复，是最理想的状态。 但真正的 Exactly-Once 在分布式系统中几乎不可能实现。所谓的 “Exactly-Once” 实际上是 “At-Least-Once + 幂等消费”。 // "Exactly-Once" = At-Least-Once 发送 + 幂等消费 async fn send_exactly_once( producer: &KafkaProducer, message: &Message, ) -> Result<(), Error> { // 消息必须有唯一 ID let message_id = message.id; // At-Least-Once 发送 producer.send_with_retry(message).await?; Ok(()) } async fn consume_exactly_once( pool: &PgPool, message: &Message, ) -> Result<(), Error> { // 幂等消费（见下文） todo!() } 生产端：确保消息发出 问题：本地事务与消息发送的一致性 // 这段代码有问题 async fn create_order_wrong(pool: &PgPool, producer: &KafkaProducer, order: Order) -> Result<(), Error> { // 1. 写数据库 sqlx::query!("INSERT INTO orders (id, user_id) VALUES ($1, $2)", order.id, order.user_id) .execute(pool) .await?; // 2. 发消息 // 如果这里失败了，订单已创建但消息没发出 producer.send(&OrderCreatedEvent::from(&order)).await?; Ok(()) } 解决方案一：本地消息表（Transactional Outbox） 我们在上一章已经详细介绍过，这里再强调关键点： ...

从这里开始 分库分表解决了数据量问题后，小明的二手书平台继续快速发展。团队也从最初的 3 人扩展到了 20 人。 单体应用的问题开始显现： 代码库越来越大，改一个小功能都要担心影响其他模块 每次发布都是全量部署，风险高 不同模块的技术栈被绑定在一起，无法独立演进 技术负责人建议：“是时候拆分服务了。” 团队经过讨论，决定把系统拆分成多个服务： 用户服务：管理用户账户和余额 库存服务：管理书籍库存 订单服务：处理订单逻辑 拆分后，架构清晰了很多。但第一个需求就让小明犯了难： 用户下单时，需要：1）从用户账户扣款，2）扣减库存，3）创建订单 在单体应用时代，这三个操作可以放在同一个数据库事务里： // 单体时代：一个事务搞定 async fn place_order(pool: &PgPool, order: OrderRequest) -> Result<Order, Error> { let mut tx = pool.begin().await?; // 1. 扣款 sqlx::query!("UPDATE users SET balance = balance - $1 WHERE id = $2", order.amount, order.user_id) .execute(&mut *tx).await?; // 2. 扣库存 sqlx::query!("UPDATE books SET stock = stock - 1 WHERE id = $1", order.book_id) .execute(&mut *tx).await?; // 3. 创建订单 let order = sqlx::query_as!(Order, "INSERT INTO orders (user_id, book_id, amount) VALUES ($1, $2, $3) RETURNING *", order.user_id, order.book_id, order.amount ).fetch_one(&mut *tx).await?; tx.commit().await?; Ok(order) } 但现在，用户余额在用户服务的数据库，库存在库存服务的数据库，订单在订单服务的数据库。三个操作跨越三个独立的数据库，无法在同一个事务中完成。 如果扣款成功了，但扣库存失败了，怎么办？钱扣了，货没发，用户肯定要投诉。 这就是跨服务事务一致性问题。 问题的本质 分布式事务难在哪里？让我们看一个简化的场景： 订单服务 库存服务 │ │ │──── 1. 扣库存请求 ──────>│ │ │ (扣库存成功) │<─── 2. 扣库存响应 ───────│ │ │ │ (创建订单...) │ │ (本地事务失败!) │ │ │ │ 现在怎么办？ │ │ 库存已经扣了，但订单没创建│ 问题的根源是： 没有全局事务协调者：每个服务只能控制自己的本地事务 网络不可靠：调用可能超时、失败、或成功但响应丢失 部分失败：一个操作成功，另一个失败 传统解决方案是两阶段提交（2PC），但它有致命缺陷： ...

从这里开始 经过缓存优化后，小明的二手书平台性能有了很大提升。但新的问题出现了：数据量太大，单个数据库快撑不住了。 订单表已经有上亿条数据，每次查询都很慢，添加索引也无济于事。磁盘空间也快满了，单机存储已经到达瓶颈。 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，这是主从延迟。读写分离带来的第一个一致性问题，你准备好了吗？ ...

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

“模式不是目的，而是手段。当手段成为目的，代码便开始腐烂。” 走完了这趟设计模式的旅程，我们学会了用 Rust 的特性让模式返璞归真。但还有一个更重要的问题：什么时候不该用模式？ 这篇终章，我们来聊聊设计模式的阴暗面——过度设计、模式滥用，以及断舍离的智慧。 过度设计的七宗罪 第一罪：未来幻想症 // ❌ 为"可能的需求"做准备 trait PaymentProcessor { fn process(&self, amount: f64) -> Result<(), Error>; fn refund(&self, transaction_id: &str) -> Result<(), Error>; fn partial_refund(&self, transaction_id: &str, amount: f64) -> Result<(), Error>; fn recurring(&self, schedule: &Schedule) -> Result<(), Error>; fn batch_process(&self, payments: Vec<Payment>) -> Result<(), Error>; fn validate_card(&self, card: &Card) -> Result<bool, Error>; fn tokenize(&self, card: &Card) -> Result<Token, Error>; fn dispute(&self, transaction_id: &str, reason: &str) -> Result<(), Error>; // 还有 20 个方法... } // ✅ 只实现当前需要的 trait PaymentProcessor { fn process(&self, amount: f64) -> Result<(), Error>; } // 需要时再扩展 trait RefundablePayment: PaymentProcessor { fn refund(&self, transaction_id: &str) -> Result<(), Error>; } 症状：接口定义了大量"将来可能用到"的方法，但实现者只用其中两三个。 Rust 的解药：trait 组合。需要什么就定义什么，通过 trait 继承和组合按需扩展。 第二罪：抽象层叠床架屋 // ❌ 抽象套抽象 trait Repository<T> { fn save(&self, entity: T) -> Result<(), Error>; } trait UserRepository: Repository<User> { fn find_by_email(&self, email: &str) -> Result<Option<User>, Error>; } trait UserRepositoryFactory { fn create(&self) -> Box<dyn UserRepository>; } trait UserRepositoryFactoryProvider { fn get_factory(&self) -> Box<dyn UserRepositoryFactory>; } // ✅ 直接了当 struct UserRepository { pool: PgPool, } impl UserRepository { pub fn save(&self, user: &User) -> Result<(), Error> { // 直接实现 } pub fn find_by_email(&self, email: &str) -> Result<Option<User>, Error> { // 直接实现 } } 症状：为了"灵活性"层层包装，但整个系统只有一种实现。 ...

促销规则天天变，每次改都要发版？解释器模式帮你构建 DSL，让业务人员自己配置规则。从简单条件到复杂表达式，看规则引擎如何炼成。 ...

文件系统、组织架构、商品分类、权限树……树形结构无处不在，但处理起来总是一团乱麻。组合、访问者、迭代器、享元四种模式，让你优雅地驯服递归之兽。 ...