Redis

引言 想象这样一个场景：用户在购物车中看到商品显示"有货"，提交订单时却提示"库存不足"。或者更糟糕的情况：订单已支付，仓库却发现没有库存可发货。这些问题的根源都指向一个核心能力：库存预占。 库存预占是OMS系统的生命线。它要解决的核心问题是：在下单到发货的整个周期内，如何保证已承诺给用户的商品库存不被其他订单抢走？本文将从第一性原理出发，系统性地探讨库存预占与释放机制的设计与实现。 库存预占的本质 什么是库存预占？ 库存预占（Inventory Reservation）是指在订单创建时，为订单中的商品临时锁定库存，防止被其他订单占用。 库存状态转换： 可用库存 → 预占库存 → 已出库 ↓ 释放回可用（订单取消） 为什么需要库存预占？ 场景1：下单到支付的时间窗口 用户操作流程： 1. 加入购物车（不预占） 2. 提交订单（预占库存）← 关键点 3. 支付（15分钟内） 4. 发货 问题：如果不预占，步骤2到步骤3之间，库存可能被其他用户购买 场景2：支付到发货的时间窗口 订单处理流程： 1. 用户支付成功 2. 订单进入待发货队列 3. 仓库拣货打包（可能需要几小时） 4. 发货 问题：如果不预占，步骤1到步骤4之间，库存可能被超卖 场景3：高并发秒杀 秒杀场景： - 库存：100件 - 请求：10000个并发请求 - 目标：保证只有前100个请求成功，且不超卖 核心挑战：如何在毫秒级时间内准确预占库存？ 库存预占的核心目标 防止超卖：已售数量 ≤ 实际库存 保证时效：预占操作要快（毫秒级） 支持释放：订单取消/超时后自动释放 高并发：支持万级QPS的库存预占 数据一致性：分布式环境下保证强一致性 库存预占流程设计 标准预占流程 class InventoryReservationService: """库存预占服务""" def reserve(self, order): """ 预占库存的标准流程 1. 校验库存可用性 2. 锁定库存 3. 扣减可用库存 4. 增加预占库存 5. 记录预占日志 """ try: # 1. 校验库存 self._validate_inventory(order) # 2. 获取分布式锁 lock = self._acquire_lock(order.items) # 3. 执行预占 reservations = [] for item in order.items: reservation = self._reserve_single_item( sku_id=item.sku_id, warehouse_id=item.warehouse_id, quantity=item.quantity, order_id=order.order_id ) reservations.append(reservation) # 4. 释放锁 self._release_lock(lock) return ReservationResult( success=True, reservations=reservations ) except InsufficientStockException as e: return ReservationResult( success=False, error=str(e) ) def _reserve_single_item(self, sku_id, warehouse_id, quantity, order_id): """预占单个SKU的库存""" # 原子操作：扣减可用库存，增加预占库存 with self.db.transaction(): # 1. 查询当前库存（加行锁） inventory = self.db.query( """ SELECT available, reserved FROM inventory WHERE sku_id = %s AND warehouse_id = %s FOR UPDATE """, (sku_id, warehouse_id) ) # 2. 检查库存充足性 if inventory.available < quantity: raise InsufficientStockException( f"SKU {sku_id} 库存不足" ) # 3. 更新库存 self.db.execute( """ UPDATE inventory SET available = available - %s, reserved = reserved + %s WHERE sku_id = %s AND warehouse_id = %s """, (quantity, quantity, sku_id, warehouse_id) ) # 4. 记录预占记录 reservation = InventoryReservation( reservation_id=generate_id(), order_id=order_id, sku_id=sku_id, warehouse_id=warehouse_id, quantity=quantity, status='RESERVED', expire_at=datetime.now() + timedelta(minutes=30) ) self.db.insert('inventory_reservations', reservation) return reservation 数据库设计 -- 库存表 CREATE TABLE inventory ( id BIGINT PRIMARY KEY AUTO_INCREMENT, sku_id VARCHAR(64) NOT NULL, warehouse_id VARCHAR(64) NOT NULL, available INT NOT NULL DEFAULT 0, -- 可用库存 reserved INT NOT NULL DEFAULT 0, -- 预占库存 sold INT NOT NULL DEFAULT 0, -- 已售库存 total INT NOT NULL DEFAULT 0, -- 总库存 version INT NOT NULL DEFAULT 0, -- 乐观锁版本号 created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL, UNIQUE KEY uk_sku_warehouse (sku_id, warehouse_id), INDEX idx_sku (sku_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 库存预占记录表 CREATE TABLE inventory_reservations ( id BIGINT PRIMARY KEY AUTO_INCREMENT, reservation_id VARCHAR(64) NOT NULL UNIQUE, order_id VARCHAR(64) NOT NULL, sku_id VARCHAR(64) NOT NULL, warehouse_id VARCHAR(64) NOT NULL, quantity INT NOT NULL, status VARCHAR(32) NOT NULL, -- RESERVED, CONSUMED, RELEASED expire_at DATETIME NOT NULL, -- 预占过期时间 created_at DATETIME NOT NULL, released_at DATETIME, INDEX idx_order (order_id), INDEX idx_expire (expire_at, status) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 库存变更日志表 CREATE TABLE inventory_change_logs ( id BIGINT PRIMARY KEY AUTO_INCREMENT, log_id VARCHAR(64) NOT NULL UNIQUE, sku_id VARCHAR(64) NOT NULL, warehouse_id VARCHAR(64) NOT NULL, change_type VARCHAR(32) NOT NULL, -- RESERVE, RELEASE, CONSUME quantity INT NOT NULL, before_available INT NOT NULL, after_available INT NOT NULL, order_id VARCHAR(64), created_at DATETIME NOT NULL, INDEX idx_sku (sku_id), INDEX idx_order (order_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; 库存预占失败处理 失败场景分类 class ReservationFailureHandler: """预占失败处理器""" def handle_failure(self, order, failure_reason): """处理预占失败""" if isinstance(failure_reason, InsufficientStockException): return self._handle_insufficient_stock(order) elif isinstance(failure_reason, DistributedLockException): return self._handle_lock_timeout(order) elif isinstance(failure_reason, DatabaseException): return self._handle_db_error(order) else: return self._handle_unknown_error(order) def _handle_insufficient_stock(self, order): """处理库存不足""" # 1. 尝试部分预占 partial_result = self._try_partial_reservation(order) if partial_result.has_available_items(): return self._create_partial_order( order, partial_result.available_items ) # 2. 尝试跨仓调拨 transfer_result = self._try_warehouse_transfer(order) if transfer_result.success: return self._retry_reservation(order) # 3. 推荐替代商品 alternatives = self._find_alternatives(order) return ReservationResult( success=False, error="库存不足", alternatives=alternatives ) def _handle_lock_timeout(self, order): """处理锁超时""" # 重试策略：指数退避 for retry in range(3): time.sleep(0.1 * (2 ** retry)) # 100ms, 200ms, 400ms try: return self.reservation_service.reserve(order) except DistributedLockException: continue # 降级策略：异步处理 return self._async_reservation(order) 部分预占策略 class PartialReservationStrategy: """部分预占策略""" def try_partial_reservation(self, order): """尝试部分预占""" available_items = [] unavailable_items = [] for item in order.items: try: # 尝试预占单个商品 reservation = self.reservation_service.reserve_single_item( item ) available_items.append(item) except InsufficientStockException: unavailable_items.append(item) return PartialReservationResult( available_items=available_items, unavailable_items=unavailable_items, fulfillment_rate=len(available_items) / len(order.items) ) def should_accept_partial(self, result, order): """判断是否接受部分预占""" # 策略1：订单金额超过阈值，接受部分预占 if order.total_amount > 500 and result.fulfillment_rate > 0.8: return True # 策略2：重要商品都有货，接受部分预占 critical_items = [i for i in order.items if i.is_critical] if all(i in result.available_items for i in critical_items): return True # 策略3：用户设置允许部分发货 if order.allow_partial_shipment: return True return False 库存释放触发条件 自动释放场景 class InventoryReleaseService: """库存释放服务""" def __init__(self): # 定义释放规则 self.release_rules = [ # 规则1：订单取消 ReleaseRule( name="order_cancelled", condition=lambda order: order.status == 'CANCELLED', action=self._release_all ), # 规则2：支付超时 ReleaseRule( name="payment_timeout", condition=lambda order: ( order.status == 'PENDING_PAYMENT' and datetime.now() > order.payment_deadline ), action=self._release_all ), # 规则3：部分退货 ReleaseRule( name="partial_refund", condition=lambda order: order.has_partial_refund(), action=self._release_partial ), # 规则4：预占过期 ReleaseRule( name="reservation_expired", condition=lambda reservation: ( datetime.now() > reservation.expire_at ), action=self._release_expired ) ] def release_for_order(self, order): """为订单释放库存""" # 1. 查询订单的所有预占记录 reservations = self.reservation_repo.find_by_order(order.order_id) # 2. 执行释放 released_count = 0 for reservation in reservations: if reservation.status != 'RESERVED': continue try: self._release_reservation(reservation) released_count += 1 except Exception as e: self.logger.error( f"释放预占失败: {reservation.reservation_id}", exc_info=e ) return ReleaseResult( success=True, released_count=released_count ) def _release_reservation(self, reservation): """释放单个预占""" with self.db.transaction(): # 1. 更新库存（原子操作） affected_rows = self.db.execute( """ UPDATE inventory SET available = available + %s, reserved = reserved - %s WHERE sku_id = %s AND warehouse_id = %s AND reserved >= %s """, ( reservation.quantity, reservation.quantity, reservation.sku_id, reservation.warehouse_id, reservation.quantity ) ) if affected_rows == 0: raise InventoryMismatchException( f"预占库存不足: {reservation.reservation_id}" ) # 2. 更新预占记录状态 self.db.execute( """ UPDATE inventory_reservations SET status = 'RELEASED', released_at = NOW() WHERE reservation_id = %s """, (reservation.reservation_id,) ) # 3. 记录变更日志 self._log_inventory_change( sku_id=reservation.sku_id, warehouse_id=reservation.warehouse_id, change_type='RELEASE', quantity=reservation.quantity, order_id=reservation.order_id ) 定时释放任务 class ScheduledReleaseJob: """定时释放任务""" def run(self): """ 每分钟执行一次，释放过期预占 批量处理策略： 1. 查询过期预占（分批查询） 2. 批量释放库存 3. 批量更新预占状态 """ batch_size = 1000 offset = 0 while True: # 1. 查询过期预占 expired_reservations = self.reservation_repo.find_expired( limit=batch_size, offset=offset ) if not expired_reservations: break # 2. 按仓库和SKU分组 grouped = defaultdict(list) for res in expired_reservations: key = (res.sku_id, res.warehouse_id) grouped[key].append(res) # 3. 批量释放 for (sku_id, warehouse_id), reservations in grouped.items(): total_quantity = sum(r.quantity for r in reservations) # 批量更新库存 self.db.execute( """ UPDATE inventory SET available = available + %s, reserved = reserved - %s WHERE sku_id = %s AND warehouse_id = %s """, (total_quantity, total_quantity, sku_id, warehouse_id) ) # 批量更新预占状态 reservation_ids = [r.reservation_id for r in reservations] self.db.execute( """ UPDATE inventory_reservations SET status = 'RELEASED', released_at = NOW() WHERE reservation_id IN %s """, (tuple(reservation_ids),) ) offset += batch_size # 限流保护 time.sleep(0.1) 库存预占的并发控制 数据库层面的并发控制 方案1：悲观锁（SELECT FOR UPDATE） ...

三种方案对比 方案 实时性 高可用 运维成本 适用场景 Nacos 秒级 高 低 推荐，微服务首选 Apollo 秒级 高 中 已使用Apollo的项目 Redis 秒级 高 低 轻量级方案 文件 分钟级 低 高 不推荐生产使用 方案1：Nacos（推荐） 配置 spring: cloud: sentinel: datasource: flow: nacos: server-addr: localhost:8848 dataId: ${spring.application.name}-flow-rules groupId: SENTINEL_GROUP rule-type: flow namespace: dev Nacos中配置规则 [ { "resource": "orderCreate", "limitApp": "default", "grade": 1, "count": 1000, "strategy": 0, "controlBehavior": 0 } ] 优点 与Spring Cloud Alibaba深度集成 支持配置热更新 提供Web控制台 支持灰度发布 支持多环境（namespace） 方案2：Apollo 添加依赖 <dependency> <groupId>com.alibaba.csp</groupId> <artifactId>sentinel-datasource-apollo</artifactId> <version>1.8.6</version> </dependency> 配置 spring: cloud: sentinel: datasource: flow: apollo: namespace-name: application flow-rules-key: sentinel.flow.rules default-flow-rule-value: "[]" rule-type: flow Apollo中配置 Namespace: application Key: sentinel.flow.rules Value: ...

一、引子：商品详情接口的性能进化之路 想象你正在开发一个电商平台的商品详情页面，每次用户访问都需要查询商品基本信息、品牌信息、类目信息、库存信息和商品图片。让我们看看三种不同的实现方式，以及它们各自的性能表现。 1.1 场景A：无缓存（直接查数据库） 这是最直接的实现方式：每次请求都查询数据库。 @Service public class ProductService { @Autowired private ProductRepository productRepository; @Autowired private BrandRepository brandRepository; @Autowired private CategoryRepository categoryRepository; @Autowired private InventoryRepository inventoryRepository; @Autowired private ProductImageRepository productImageRepository; /** * 查询商品详情（每次请求都查数据库） * 平均耗时：100ms * QPS上限：500 */ public ProductDetailVO getProductDetail(Long productId) { // 1. 查询商品基本信息（20ms） Product product = productRepository.findById(productId); if (product == null) { throw new ProductNotFoundException("商品不存在：" + productId); } // 2. 查询品牌信息（20ms） Brand brand = brandRepository.findById(product.getBrandId()); // 3. 查询类目信息（20ms） Category category = categoryRepository.findById(product.getCategoryId()); // 4. 查询库存信息（20ms） Inventory inventory = inventoryRepository.findByProductId(productId); // 5. 查询商品图片（20ms，可能有N+1查询问题） List<ProductImage> images = productImageRepository.findByProductId(productId); // 6. 组装返回对象 ProductDetailVO vo = new ProductDetailVO(); vo.setProductId(product.getId()); vo.setProductName(product.getName()); vo.setPrice(product.getPrice()); vo.setBrandName(brand.getName()); vo.setCategoryName(category.getName()); vo.setStock(inventory.getStock()); vo.setImages(images); return vo; } } 性能数据（压测工具：JMeter，1000并发）： ...

一、引子：一个用户会话缓存的演进之路 假设你正在开发一个电商网站的用户会话管理功能。每次用户请求都需要验证身份，最初的实现是每次都查询数据库，但随着用户量增长，数据库压力越来越大。让我们看看这个功能如何一步步演进，从最简单的HashMap到最终的Redis分布式缓存。 1.1 场景0：无缓存（每次查数据库） 最直接的实现：每次请求都查询数据库验证用户身份。 @RestController public class UserController { @Autowired private UserRepository userRepository; /** * 获取用户信息（每次查数据库） * 问题：数据库压力大，响应慢 */ @GetMapping("/api/user/info") public UserVO getUserInfo(@RequestHeader("token") String token) { // 1. 根据token查询用户ID（查数据库） Long userId = tokenRepository.findUserIdByToken(token); if (userId == null) { throw new UnauthorizedException("未登录"); } // 2. 查询用户详细信息（查数据库） User user = userRepository.findById(userId); if (user == null) { throw new UserNotFoundException("用户不存在"); } return convertToVO(user); } } 性能数据： 指标 数值 说明 平均响应时间 50ms 2次SQL查询 QPS上限 1000 数据库连接池限制 数据库压力 100% 每次请求都查库 问题： ...

一、引子：为什么Redis需要五大数据结构？ 很多人的疑问：Memcached只有String一种数据结构，Redis为什么需要五种？ 核心答案：不同的业务场景需要不同的数据结构。 1.1 如果只有String会怎样？ 假设我们要实现一个排行榜功能，只有String的话： // ❌ 方案1：用String存储整个排行榜（JSON序列化） // 问题：每次更新一个用户分数，需要序列化/反序列化整个排行榜 public void updateScore(Long userId, int score) { // 1. 读取整个排行榜（反序列化） String json = redisTemplate.opsForValue().get("rank:list"); List<User> rankList = JSON.parseArray(json, User.class); // 10000个用户 // 2. 更新一个用户的分数 for (User user : rankList) { if (user.getId().equals(userId)) { user.setScore(score); break; } } // 3. 重新排序 rankList.sort((a, b) -> b.getScore() - a.getScore()); // 4. 写入Redis（序列化） String newJson = JSON.toJSONString(rankList); redisTemplate.opsForValue().set("rank:list", newJson); } // 性能问题： // - 读取：反序列化10000个用户，耗时100ms // - 排序：O(NlogN) = 10000*log(10000) ≈ 130000次比较 // - 写入：序列化10000个用户，耗时100ms // 总耗时：200ms+（单次更新） // ✅ 方案2：使用Redis ZSet（有序集合） // 优势：Redis内部维护排序，O(logN)复杂度 public void updateScore(Long userId, int score) { redisTemplate.opsForZSet().add("rank:zset", userId.toString(), score); } // 性能提升： // - 写入：O(logN) = log(10000) ≈ 13次比较 // - 总耗时：1ms // 性能提升：200倍 核心洞察： ...

一、引子：单机Redis的三大困境 假设你正在运行一个电商网站，使用单机Redis作为缓存。随着业务增长，你会遇到三个核心问题： 1.1 问题1：单点故障（可用性） 场景：双11大促，凌晨0点 00:00:00 - Redis单机宕机（内存不足OOM） 00:00:01 - 所有请求打到数据库 00:00:02 - 数据库连接池耗尽（1000个请求/秒） 00:00:03 - 数据库CPU 100% 00:00:05 - 网站503错误，用户无法下单 00:00:10 - 运维手动重启Redis 00:00:15 - Redis启动成功，但缓存为空 00:00:20 - 缓存预热中（需要10分钟） 00:10:00 - 系统恢复正常 损失： - 10分钟宕机时间 - 订单损失：10分钟 × 1000单/分钟 = 1万单 - GMV损失：1万单 × 200元/单 = 200万元 核心问题：单点故障导致系统不可用，无容灾能力。 可用性计算： 假设：Redis每月宕机1次，每次10分钟 可用性 = (30天 × 24小时 × 60分钟 - 10分钟) / (30天 × 24小时 × 60分钟) = (43200 - 10) / 43200 = 99.977% 看起来还不错？但实际上： - 1年12次宕机，累计120分钟 = 2小时 - 如果恰好在大促期间宕机，损失巨大 - 高可用系统要求：99.99%（年宕机时间 < 52分钟） 1.2 问题2：容量瓶颈（可扩展性） 场景：业务增长，数据量暴增 第1个月： - 商品数量：10万 - 缓存数据量：2GB - 单机Redis：4GB内存（够用） 第6个月： - 商品数量：100万 - 缓存数据量：20GB - 单机Redis：4GB内存（不够用！） 解决方案1：垂直扩展（加内存） - 4GB → 16GB（成本翻倍，但有上限） - 最大：512GB（成本极高，且有物理上限） 解决方案2：水平扩展（加机器） - 需要Redis集群（数据分片） 核心问题：单机内存有限，垂直扩展有上限，需要水平扩展。 ...

一、引子：Redis宕机后数据去哪了？ Redis是内存数据库，数据存储在内存中。那么问题来了：Redis宕机后，数据还在吗？ 1.1 场景：双11大促数据丢失事故 场景：电商网站，双11大促 00:00:00 - 大促开始，流量暴增 00:00:01 - Redis缓存：10万个商品详情、50万个用户Session 00:05:00 - 服务器断电（机房故障） 00:05:01 - Redis进程被杀，内存数据全部丢失 00:10:00 - 服务器恢复，Redis重启 00:10:01 - Redis启动成功，但数据为空！ 问题： 1. 所有商品详情缓存丢失 → 10万次数据库查询 2. 所有用户Session丢失 → 50万用户被强制登出 3. 缓存预热需要10分钟 → 数据库压力巨大 4. 用户体验极差 → 大量用户流失 损失： - 用户流失：50万用户 × 10% = 5万用户 - 订单损失：5万用户 × 20% = 1万单 - GMV损失：1万单 × 200元 = 200万元 核心问题：Redis是内存数据库，断电后数据丢失，需要持久化。 1.2 持久化的本质 持久化：将内存中的数据保存到磁盘，重启后可以恢复。 为什么需要持久化？ 原因1：数据安全 - Redis宕机、服务器断电、进程被杀 - 内存数据丢失 - 持久化后可以恢复 原因2：快速恢复 - 没有持久化：需要从数据库重新加载（10分钟+） - 有持久化：直接加载持久化文件（1分钟） 原因3：数据备份 - 定期备份持久化文件 - 灾难恢复（机房火灾、磁盘损坏） Redis的两种持久化方式： ...

一、分布式锁：从SETNX到Redlock 1.1 为什么需要分布式锁？ 场景：秒杀系统的超卖问题 // ❌ 错误：单机锁无法解决分布式超卖 @Service public class SeckillService { @Autowired private ProductRepository productRepository; /** * 秒杀下单（单机锁） */ public synchronized Order seckill(Long productId, Long userId) { // 1. 检查库存 Integer stock = productRepository.getStock(productId); if (stock <= 0) { throw new SoldOutException("商品已售罄"); } // 2. 扣减库存 productRepository.decrementStock(productId); // 3. 创建订单 return orderService.createOrder(productId, userId); } } // 问题： // 假设有3台服务器（Server A、B、C） // T1: Server A：检查库存=1（通过） // T2: Server B：检查库存=1（通过） // T3: Server A：扣减库存=0，创建订单1 // T4: Server B：扣减库存=-1（超卖！），创建订单2 // synchronized只能锁住单机JVM内的线程 // 无法锁住分布式环境的多个进程 核心问题：分布式环境下，单机锁无效，需要分布式锁。 ...

一、引子：商品详情接口的性能进化之路 想象你正在开发一个电商平台的商品详情页面，每次用户访问都需要查询商品基本信息、品牌信息、类目信息、库存信息和商品图片。让我们看看三种不同的实现方式，以及它们各自的性能表现。 1.1 场景A：无缓存（直接查数据库） 这是最直接的实现方式：每次请求都查询数据库。 @Service public class ProductService { @Autowired private ProductRepository productRepository; @Autowired private BrandRepository brandRepository; @Autowired private CategoryRepository categoryRepository; @Autowired private InventoryRepository inventoryRepository; @Autowired private ProductImageRepository productImageRepository; /** * 查询商品详情（每次请求都查数据库） * 平均耗时：100ms * QPS上限：500 */ public ProductDetailVO getProductDetail(Long productId) { // 1. 查询商品基本信息（20ms） Product product = productRepository.findById(productId); if (product == null) { throw new ProductNotFoundException("商品不存在：" + productId); } // 2. 查询品牌信息（20ms） Brand brand = brandRepository.findById(product.getBrandId()); // 3. 查询类目信息（20ms） Category category = categoryRepository.findById(product.getCategoryId()); // 4. 查询库存信息（20ms） Inventory inventory = inventoryRepository.findByProductId(productId); // 5. 查询商品图片（20ms，可能有N+1查询问题） List<ProductImage> images = productImageRepository.findByProductId(productId); // 6. 组装返回对象 ProductDetailVO vo = new ProductDetailVO(); vo.setProductId(product.getId()); vo.setProductName(product.getName()); vo.setPrice(product.getPrice()); vo.setBrandName(brand.getName()); vo.setCategoryName(category.getName()); vo.setStock(inventory.getStock()); vo.setImages(images); return vo; } } 性能数据（压测工具：JMeter，1000并发）： ...

一、引子：一个用户会话缓存的演进之路 假设你正在开发一个电商网站的用户会话管理功能。每次用户请求都需要验证身份，最初的实现是每次都查询数据库，但随着用户量增长，数据库压力越来越大。让我们看看这个功能如何一步步演进，从最简单的HashMap到最终的Redis分布式缓存。 1.1 场景0：无缓存（每次查数据库） 最直接的实现：每次请求都查询数据库验证用户身份。 @RestController public class UserController { @Autowired private UserRepository userRepository; /** * 获取用户信息（每次查数据库） * 问题：数据库压力大，响应慢 */ @GetMapping("/api/user/info") public UserVO getUserInfo(@RequestHeader("token") String token) { // 1. 根据token查询用户ID（查数据库） Long userId = tokenRepository.findUserIdByToken(token); if (userId == null) { throw new UnauthorizedException("未登录"); } // 2. 查询用户详细信息（查数据库） User user = userRepository.findById(userId); if (user == null) { throw new UserNotFoundException("用户不存在"); } return convertToVO(user); } } 性能数据： 指标 数值 说明 平均响应时间 50ms 2次SQL查询 QPS上限 1000 数据库连接池限制 数据库压力 100% 每次请求都查库 问题： ...