Day 2 · Java 后端工程代码审查 — 深入审查 Issue 与解决方案教学文

配套要求：Java 17 / Spring Boot 3 / MySQL 8 / Redis Cluster / RabbitMQ；全部代码示例力求可编译、可运行。

今日审查主线速览

一句话主线：这四道题表面上分别是缓存、安全、MQ、线程池，实际上都在回答同一个工程问题——“系统的边界在哪里？”

缓存要守住一致性边界（Cache Aside 的删除时序与失败处理）；文件下载要守住信任边界（客户端输入永远不可信）；MQ 要守住投递语义边界（at-least-once 是默认现实，幂等是消费者的第一原则）；线程池要守住资源边界（有界 = 背压，线程数由下游决定而非由流量决定）。

代码本身能跑，只代表通过了最低门槛。真正的 Code Review 要逐级追问：正常情况能跑 → 异常情况会怎样 → 并发情况下会怎样 → 依赖故障会怎样 → 流量扩大 100 倍会怎样 → 失败以后能不能恢复。 这四道题就是这六级追问的四个具体切片。

题 1：用户资料缓存（Cache Aside 一致性 / 热点 Key 击穿 / 穿透·击穿·雪崩）

🚩 审查 Issue（结论先行）

结论先行：这是一套"能跑但边界模糊"的经典 Cache Aside 实现——它假设了"DB 提交和 Redis 删除必然同时成功、且不会有并发读旧写回"，而现实是两个系统之间不存在分布式事务。必须围绕"删除失败"与"并发写回"两条失败路径重新设计。

📖 深入讲解（为什么有问题）

1. 更新数据库 → 删除缓存，是否一定安全？

不一定。 这是 Cache Aside（旁路缓存）模式的"阿喀琉斯之踵"。Cache Aside 的写路径规则是"先更新 DB，再删除缓存"（Write-Around 思路：让缓存下次读取时自然重建，而不是去更新缓存）。这个规则本身是对的——删除缓存比更新缓存更不容易引入不一致（更新缓存需要把 DB 数据完整序列化进缓存，任何字段遗漏都会造成脏数据；删除则简单粗暴）。但"删除"这一步和"DB commit"发生在两个不同的系统里，没有事务边界，所以存在两个经典失败点：

• 失败点 1：DB commit 成功，Redis delete 失败。 缓存里躺着旧数据，新数据永远进不来，直到 TTL 到期。
• 失败点 2：并发时序错乱（A miss / B update）。 见下文第 3 点。

为什么"更新缓存"（Write-Through）反而更危险？因为 DB 与缓存是两套存储，更新缓存成功、DB 失败 → 缓存是新的、DB 是旧的，且没有 TTL 兜底，错误可能永久存在；而删除缓存后，最坏情况是缓存 miss 回源，错误窗口被 TTL 截断。所以业界共识：写路径用"删"不用"改"，问题从"如何保证双写一致"降级为"如何保证删除可靠"。

2. DB commit 成功、Redis delete 失败，会怎样？

后果链：commit 成功 → delete 失败 → 缓存里保留旧昵称 → 后续所有读都命中旧缓存 → 用户持续看到旧昵称 → 直到 24h TTL 到期才自动恢复 → 不一致窗口最长 24 小时。

为什么 delete 会失败？Redis Cluster 下可能的原因：网络抖动、Redis 节点故障、连接池耗尽、主从切换期间请求超时。而在本例中更隐蔽的一点是：redisTemplate.delete() 抛出异常时，@Transactional 方法会把异常传播出去导致事务回滚——也就是说 delete 失败不但没删掉缓存，还让已经成功的 DB 更新也回滚了（这倒是"歪打正着"地避免了一致性破坏，但用户的操作直接报错，可用性受损）。这是"缓存操作和 DB 事务耦合"的典型反面教材：缓存删除不应该影响业务事务的成败。

3. 两个线程并发：A cache miss / B update，最终结果？

时间线：

线程 A（读） 线程 B（写）
1. GET cache -> miss
2. SELECT user（读到旧昵称）
3. UPDATE user SET nickname=新昵称
4. DELETE cache
5. SET cache = 旧昵称 ← 脏数据写回！

最终结果：缓存里是旧昵称，DB 里是新昵称，缓存与 DB 不一致，且这个脏数据没有"下一次删除"来纠正——B 已经删过缓存了，A 是在 B 删除之后才写回的。这个脏值会一直躺到 TTL 到期。注意：即使 B 用的是"更新缓存"也一样脏，因为 A 拿到的是删除/更新之前的旧快照。只要"读-写回"不是原子的，"最后写者胜"就会产生脏数据。

业界对策（后面候选方案逐一对比）：

• delayed double delete（延迟双删）：B 在删除后延迟 1s 再删一次，把 A 写回的窗口"事后擦掉"——治标，但窗口仍然存在（若 A 的写回发生在第二次删除之后呢？仍有理论窗口，只是概率极低）。
• 读时加锁回源（mutex/single-flight）：让"读 DB + 写缓存"被同一把锁保护，B 的删除与 A 的回源串行化——从根上消除脏写回（推荐，见解决方案）。
• CDC / MQ 失效：把"删缓存"变成异步可靠事件，配合重试，从可靠性上兜底。

4. 24 小时 TTL 合理吗？

不合理，双层不合理。

• 对一致性而言：TTL 是不一致的"自愈时间"。24h 意味着一旦 delete 失败或脏写回发生，用户最长要等一天才能看到正确数据。用户资料这种"展示型数据"，业界一般把 TTL 压到分钟级（如 30 分钟 + 抖动），让最坏不一致窗口可控。
• 对击穿而言：TTL 越长，Key 到期瞬间累积的"过期前热度"越高，击穿冲击越大。而且所有 Key 在同一时刻（写入时）起算 24h，会导致同一批写入的 Key 在同一时刻集中过期——这正是雪崩的温床。
• TTL 的真正职责是"防缓存永不失效"（兜底），而不是"承担一致性修复"。一致性要由删除策略负责，TTL 只做最后防线。

5. 热点 Key 失效会发生什么？（击穿）

明星用户 = 单 Key 超高 QPS。该 Key 到期瞬间，所有读请求同时 miss、同时回源 MySQL。假设该 Key 平时 5000 QPS 全部命中缓存，到期后这 5000 QPS 在一瞬间全部变成 DB 查询——即使 MySQL 单次查询只要 1ms，5000 并发也远超连接池上限（通常几十到几百），导致：连接池排队 → 请求超时 → 雪球式放大。这就是线上现象 2。注意击穿（单个热点 Key 过期）与雪崩（大量 Key 同时过期）的区别：一个是"点"的尖峰，一个是"面"的洪峰。

6. 缓存穿透 / 击穿 / 雪崩，分别有没有风险？

三者对策不同：穿透 → 空值缓存（短 TTL）+ 参数校验 + 布隆过滤器；击穿 → mutex / single-flight 单飞回源 + logical expire；雪崩 → 随机 TTL + Redis 高可用（Cluster/哨兵）+ 多级缓存（本地 Caffeine 兜底）。

7. 候选方案对比（逐项回答"是否需要"）

8. 最终一致还是强一致？

用户资料（昵称、头像等展示型数据）应当允许"秒级最终一致"，不应追求强一致。 理由：

• 为什么不能强一致？强一致 = 读必须读到最新值 = 读路径要校验/绕过缓存 = 每个读都打 DB 或做分布式事务，缓存存在的意义就没了。
• 为什么最终一致可接受？昵称变更对用户是"低敏操作"，容忍 1~5 秒的展示延迟，产品上无感知；而它换来的读吞吐（缓存命中率 99%+）是刚需。
• 工程红线："最终"要有界——必须通过"删除重试 + 短 TTL + 延迟双删"把不一致窗口压到秒级，而不是放任 24h。

对比：如果是余额、库存、支付状态这类资金/强约束数据，结论完全不同——要么不缓存，要么缓存只做"读展示 + 严格失效 + 事务内对账"，且必须允许读库兜底。先判断业务性质，再决定一致性级别。

🛠 解决方案

候选方案 → 推荐组合拳：

推荐方案 = Cache Aside 改进版（四件套组合）：

Trade-off 说明：为什么不用 CDC 一步到位？——引入 Canal 增加运维复杂度，对"用户资料"这个量级是过度设计；"业务主动删 + MQ 补偿"已能覆盖 99% 的失败路径，且代码直观可测。为什么 mutex 锁 TTL 取 3s？——锁的生命周期必须覆盖"DB 查询 + 序列化 + 写缓存"的最坏耗时（本场景 <100ms，3s 充裕）；锁过期早于回源完成会导致第二个线程也拿到锁（锁失效），因此锁 TTL 宁大勿小，回源完成后主动释放。

// ============ 1. 配置：RedisTemplate 改用 JSON 序列化（替代 JDK 序列化） ============
package com.example.user.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration
public class RedisConfig {

@Bean
public RedisTemplate<String, Object> redisTemplate(
RedisConnectionFactory connectionFactory,
ObjectMapper objectMapper) {
RedisTemplate<String, Object> template = new RedisTemplate<>();
template.setConnectionFactory(connectionFactory);
// Key 用 String，Value 用 JSON，避免 JDK 序列化的体积大/不可读/反序列化风险
StringRedisSerializer stringSerializer = new StringRedisSerializer();
GenericJackson2JsonRedisSerializer jsonSerializer =
new GenericJackson2JsonRedisSerializer(objectMapper);
template.setKeySerializer(stringSerializer);
template.setHashKeySerializer(stringSerializer);
template.setValueSerializer(jsonSerializer);
template.setHashValueSerializer(jsonSerializer);
template.afterPropertiesSet();
return template;
}
}

// ============ 2. 领域对象：UserProfile 需要可被 JSON 序列化 ============
package com.example.user.model;

public record UserProfile(
Long userId,
String nickname,
String avatarUrl,
String email
) {
public static UserProfile from(User user) {
return new UserProfile(user.getId(), user.getNickname(),
user.getAvatarUrl(), user.getEmail());
}
}

// ============ 3. 缓存失效服务：重试 + 延迟双删 + MQ 补偿 ============
package com.example.user.support;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
@RequiredArgsConstructor
public class CacheInvalidationService {

private static final int MAX_SYNC_RETRY = 3;
private static final long SECOND_DELETE_DELAY_MS = 1000L;

// 延迟双删用的定时线程（守护线程，不阻塞 JVM 退出）
private static final ScheduledExecutorService DELAYED_DELETE_POOL =
Executors.newSingleThreadScheduledExecutor(r -> {
Thread t = new Thread(r, "cache-delayed-delete");
t.setDaemon(true);
return t;
});

private final RedisTemplate<String, Object> redisTemplate;
// 生产环境注入 MqInvalidationPublisher（可选），delete 重试失败后发补偿消息

/**
* 删除缓存（带重试 + 延迟双删）。
* 为什么延迟双删？——覆盖"线程 A 读旧值 → 线程 B 删缓存 → A 写回旧值"的并发窗口：
* 第一次删除尽量缩短不一致窗口；第二次删除把 A 写回的脏值擦掉。
* 注意：延迟双删不是严格正确（A 的写回若晚于第二次删除仍有理论窗口），
* 所以真正的根治靠读路径的 single-flight（见 UserProfileService），双删只是兜底。
*/
public void deleteCache(String key) {
deleteWithRetry(key, MAX_SYNC_RETRY);
DELAYED_DELETE_POOL.schedule(() -> deleteWithRetry(key, MAX_SYNC_RETRY),
SECOND_DELETE_DELAY_MS, TimeUnit.MILLISECONDS);
}

private void deleteWithRetry(String key, int maxAttempts) {
for (int attempt = 1; attempt <= maxAttempts; attempt++) {
try {
redisTemplate.delete(key);
return;
} catch (Exception e) {
log.warn("cache delete failed, key={}, attempt={}", key, attempt, e);
try {
Thread.sleep(100L * attempt); // 100ms / 200ms / 300ms 退避
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
return;
}
}
}
// 同步重试仍失败：发 MQ 由独立消费者补偿删除（解耦、可持久化、可重试）
log.error("cache delete failed after {} attempts, publishing compensation event, key={}",
maxAttempts, key);
// mqPublisher.publish(new CacheInvalidationEvent(key));
}
}

// ============ 4. 核心服务：读路径 single-flight + 写路径 afterCommit 删缓存 ============
package com.example.user.service;

import com.example.user.model.User;
import com.example.user.model.UserProfile;
import com.example.user.repo.UserRepository;
import com.example.user.support.CacheInvalidationService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import java.time.Duration;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.CompletableFuture;

@Slf4j
@Service
@RequiredArgsConstructor
public class UserProfileService {

private static final Duration TTL_BASE = Duration.ofMinutes(30); // 基础 TTL
private static final Duration TTL_JITTER = Duration.ofMinutes(5); // 随机抖动，防雪崩
private static final Duration NULL_TTL = Duration.ofMinutes(5); // 空值缓存，防穿透
private static final Duration LOCK_TTL = Duration.ofSeconds(3); // 回源锁生命周期（必须 > 回源耗时）
private static final String KEY_PREFIX = "user:profile:";
private static final String LOCK_PREFIX = "user:profile:lock:";

private final UserRepository userRepository;
private final RedisTemplate<String, Object> redisTemplate;
private final CacheInvalidationService invalidationService;

// single-flight：进程内按 key 合并并发回源（跨实例互斥由 Redis 分布式锁承担）
private final ConcurrentHashMap<Long, CompletableFuture<UserProfile>> inFlight =
new ConcurrentHashMap<>();

// —————- 读路径 —————-

public UserProfile getProfile(Long userId) {
String key = KEY_PREFIX + userId;
Object cached = redisTemplate.opsForValue().get(key);
if (cached instanceof UserProfile profile) {
return profile;
}
if (cached instanceof NullProfile) { // 命中空值缓存：直接 404，不再打 DB
throw new UserNotFoundException(userId);
}
return singleFlightLoad(userId, key);
}

/**
* single-flight 回源：同一 userId 的并发请求合并为一次 DB 查询。
* 为什么能根治脏写回？——"查 DB + 写缓存"整个过程被合并/互斥，
* 不会出现"读旧值→别人删缓存→我写回旧值"的交错。
*/
private UserProfile singleFlightLoad(Long userId, String key) {
CompletableFuture<UserProfile> future = inFlight.computeIfAbsent(userId,
id -> CompletableFuture.supplyAsync(() -> loadFromDbWithMutex(id, key))
.whenComplete((p, e) -> inFlight.remove(id)));
try {
return future.join();
} catch (Exception e) {
Throwable cause = e.getCause() instanceof RuntimeException re ? re : e;
if (cause instanceof UserNotFoundException) {
throw (UserNotFoundException) cause;
}
throw new IllegalStateException("load profile failed", cause);
}
}

private UserProfile loadFromDbWithMutex(Long userId, String key) {
String lockKey = LOCK_PREFIX + userId;
Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "1", LOCK_TTL);
if (Boolean.TRUE.equals(locked)) {
try {
// 拿到锁的线程：双重检查，避免"锁内又查一次"（其实 single-flight 已合并，这里是双保险）
Object again = redisTemplate.opsForValue().get(key);
if (again instanceof UserProfile profile) {
return profile;
}
UserProfile profile = userRepository.findById(userId)
.map(UserProfile::from)
.orElse(null);
if (profile == null) {
// 防穿透：把"不存在"也缓存起来，短 TTL
redisTemplate.opsForValue().set(key, NullProfile.INSTANCE, NULL_TTL);
throw new UserNotFoundException(userId);
}
long ttlSeconds = TTL_BASE.toSeconds()
+ ThreadLocalRandom.current().nextLong(0, TTL_JITTER.toSeconds());
redisTemplate.opsForValue().set(key, profile, Duration.ofSeconds(ttlSeconds));
return profile;
} finally {
redisTemplate.delete(lockKey); // 回源完成主动释放锁
}
}
// 没拿到锁：说明其他实例正在回源，短暂自旋等缓存
for (int i = 0; i < 10; i++) {
try {
Thread.sleep(50);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
Object cached = redisTemplate.opsForValue().get(key);
if (cached instanceof UserProfile profile) {
return profile;
}
}
// 等待超时仍无缓存（回源方失败）：直接读 DB 兜底，但不写缓存（避免脏写回）
return userRepository.findById(userId).map(UserProfile::from)
.orElseThrow(() -> new UserNotFoundException(userId));
}

// —————- 写路径 —————-

@Transactional
public void updateNickname(Long userId, String nickname) {
User user = userRepository.findById(userId)
.orElseThrow(() -> new UserNotFoundException(userId));
user.setNickname(nickname);
userRepository.save(user);

// 关键点 1：缓存删除必须等事务提交成功后再做（afterCommit），
// 否则 delete 在 commit 之前执行：DB 回滚时缓存被白白删除（损失命中），
// 更糟的是 delete 抛异常会把整个业务事务一起回滚（缓存操作不该影响业务成败）。
// 关键点 2：afterCommit 保证"commit 成功 → 删缓存"的次序，把失败面收窄到 Redis 一侧。
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override
public void afterCommit() {
invalidationService.deleteCache(KEY_PREFIX + userId);
}
});
}
}

// ============ 5. 空值标记（可被 JSON 序列化的 record） ============
package com.example.user.service;

/**
* 空值缓存标记：表示"该 userId 不存在"。
* 为什么不能直接存 null？——RedisTemplate 存 null 会抛异常；
* 为什么要有独立类型？——区分"缓存未命中"（继续回源）与"确定不存在"（直接 404）。
*/
public record NullProfile() {
public static final NullProfile INSTANCE = new NullProfile();
}

// ============ 6.（可选）热点用户识别：命中计数超阈值 → 改用 logical expire ============
package com.example.user.service;

import lombok.RequiredArgsConstructor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.ThreadLocalRandom;

/**
* logical expire（逻辑过期）：缓存物理永不过期，value 携带逻辑过期时间。
* 读时若逻辑过期：先返回旧值给调用方（用户无感知），后台抢锁刷新。
* 彻底消灭击穿——过期瞬间没有任何请求会 miss 到 DB。
* 适用于明星用户这类"高并发 + 高热点"的 Key；普通用户用短 TTL + 抖动即可。
*/
@Component
@RequiredArgsConstructor
public class LogicalExpireCache {

private final RedisTemplate<String, Object> redisTemplate;

public record LogicalValue<T>(T value, Instant expireAt) {}

public <T> T getOrRefresh(String key, Duration logicalTtl,
Loader<T> loader, String lockKey) {
Object cached = redisTemplate.opsForValue().get(key);
if (cached instanceof LogicalValue<?> lv) {
@SuppressWarnings("unchecked")
T value = (T) lv.value();
if (lv.expireAt().isAfter(Instant.now())) {
return value; // 未逻辑过期：直接返回
}
// 逻辑过期：先返回旧值，后台抢锁刷新（stale-while-revalidate）
if (Boolean.TRUE.equals(redisTemplate.opsForValue()
.setIfAbsent(lockKey, "1", Duration.ofSeconds(5)))) {
try {
T fresh = loader.load();
redisTemplate.opsForValue().set(key,
new LogicalValue<>(fresh, Instant.now().plus(logicalTtl)));
} finally {
redisTemplate.delete(lockKey);
}
}
return value;
}
// 物理 miss：同步加载（首访，无旧值可返回）
T fresh = loader.load();
redisTemplate.opsForValue().set(key,
new LogicalValue<>(fresh, Instant.now().plus(logicalTtl)));
return fresh;
}

@FunctionalInterface
public interface Loader<T> {
T load();
}
}

关于锁生命周期的补充讲解：Redis 分布式锁的 TTL（这里取 3s）必须大于"临界区最坏执行时间"（DB 查询 + 网络 + 序列化 + 写缓存）。为什么？——因为 SET key NX EX 3 的锁在 3s 后自动消失；若回源耗时超过 3s，第二个线程会在锁过期后再次拿到锁，造成双重回源，锁就失效了。生产中更稳妥的做法是给锁续期（Watch Dog 机制，Redisson 内置），或者把锁 TTL 设得远大于回源耗时。反过来说，回源正常都在几十毫秒内完成，3s 锁 + 完成后主动释放，是最简单的够用方案。

🧪 需要补充的测试

package com.example.user.service;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import com.example.user.model.User;
import com.example.user.model.UserProfile;
import com.example.user.repo.UserRepository;
import com.example.user.support.CacheInvalidationService;

import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.*;

@ExtendWith(MockitoExtension.class)
class UserProfileServiceTest {

@Mock UserRepository userRepository;
@Mock RedisTemplate<String, Object> redisTemplate;
@Mock ValueOperations<String, Object> valueOperations;
@Mock CacheInvalidationService invalidationService;

UserProfileService service;

@BeforeEach
void setUp() {
when(redisTemplate.opsForValue()).thenReturn(valueOperations);
service = new UserProfileService(userRepository, redisTemplate, invalidationService);
}

@Test
void concurrentMiss_shouldQueryDbOnlyOnce() throws Exception {
// 单飞验证：20 个并发 miss，DB 只被查 1 次
AtomicInteger dbCalls = new AtomicInteger();
when(valueOperations.get("user:profile:1")).thenReturn(null);
when(valueOperations.setIfAbsent(eq("user:profile:lock:1"), eq("1"), any()))
.thenReturn(true);
when(userRepository.findById(1L)).thenAnswer(inv -> {
dbCalls.incrementAndGet();
return Optional.of(user(1L, "old"));
});

int n = 20;
CountDownLatch start = new CountDownLatch(1);
CountDownLatch done = new CountDownLatch(n);
ExecutorService pool = Executors.newFixedThreadPool(n);
for (int i = 0; i < n; i++) {
pool.submit(() -> {
try {
start.await();
service.getProfile(1L);
} catch (Exception ignored) {
} finally {
done.countDown();
}
});
}
start.countDown();
done.await();
pool.shutdown();

assertThat(dbCalls.get()).isEqualTo(1);
}

@Test
void notFoundUser_shouldBeCachedAsNull() {
when(valueOperations.get("user:profile:999")).thenReturn(null);
when(valueOperations.setIfAbsent(eq("user:profile:lock:999"), eq("1"), any()))
.thenReturn(true);
when(userRepository.findById(999L)).thenReturn(Optional.empty());

org.junit.jupiter.api.Assertions.assertThrows(
UserNotFoundException.class, () -> service.getProfile(999L));
// 空值写入了缓存（防穿透）
verify(valueOperations).set(eq("user:profile:999"), any(NullProfile.class), any());
}

@Test
void updateNickname_deleteCacheMustRunAfterCommit() {
when(userRepository.findById(1L)).thenReturn(Optional.of(user(1L, "old")));
// 模拟"事务提交后删除缓存"
service.updateNickname(1L, "new");
// afterCommit 在事务提交时触发；这里直接断言注册了同步（事务管理器为 mock 时 afterCommit 不触发，
// 集成测试中应配合 TransactionTemplate 验证"commit 成功后才 delete"的次序）
verify(invalidationService, never()).deleteCache("user:profile:1");
}

private User user(Long id, String nickname) {
User u = new User();
u.setId(id);
u.setNickname(nickname);
return u;
}
}

补充说明：上例中 updateNickname 的单测因为 TransactionSynchronizationManager 需要真实事务才触发 afterCommit，单测只能验证"未提交前不删缓存"。真正的集成测试应该用 @SpringBootTest + 真实 Redis（Testcontainers）+ TransactionTemplate，断言：事务回滚 → 缓存不删；事务提交 → 缓存删除且延迟 1s 后再次删除。

💡 工程要点（教学总结）

• 写路径三大纪律：先 DB 后缓存、commit 成功才删缓存（afterCommit）、删除失败必须重试/补偿——把"两系统无事务"的缺陷用"时序 + 重试"补上。
• 读路径两大杀器：single-flight/mutex 消灭击穿与脏写回；空值缓存消灭穿透。"读-写回"不原子，就必须让并发读合并。
• TTL 三原则：短（分钟级而非小时级）、随机（防雪崩）、只做兜底不做一致性修复。
• 三兄弟口诀：穿透是"查不存在的 Key"，击穿是"一个热点 Key 过期"，雪崩是"一堆 Key 同时过期"——对策分别是空值缓存、单飞回源、随机 TTL。
• 先定业务性质再定一致性：展示型数据接受秒级最终一致（有界），资金类数据不许缓存或严格对账。
• 生产级细节：Redis 用 JSON 序列化而非 JDK 序列化；锁的 TTL 必须覆盖临界区耗时；多实例下进程内 single-flight 必须叠加分布式锁。

题 2：文件下载接口（路径穿越 / 越权 / Content-Disposition / 大文件下载）

🚩 审查 Issue（结论先行）

结论先行：这是一道 P0 级安全审查题。三个叠加的致命伤：信任前端传路径（路径穿越 + 任意文件读取）→ 无归属校验（越权下载）→ 头注入与性能方案错误。任何一条都足以让这个接口在安全评审中被打回重做。

📖 深入讲解（为什么有问题）

1. 漏洞是什么？风险定 P 几？

• ?path=/etc/passwd：new File("/etc/passwd") 是绝对路径，直接读取系统文件。因为服务运行账户能访问 /etc/、/opt/app/、/var/log/，攻击者可以读取：/etc/passwd（用户列表）、/opt/app/application.yml（数据库口令、Redis 密码、云厂商 AK/SK）、/var/log/*.log（业务数据）、甚至 ~/.ssh/id_rsa。→ P0，任意文件读取。
• ?path=/data/uploads/../../etc/passwd：.. 目录回溯。文件系统解析路径时，/data/uploads/../../etc/passwd 规范化后等于 /etc/passwd。→ P0，路径穿越。
• 两个请求等价，都是 P0。为什么是 P0 而不是 P1？——因为这不是"特定条件下才触发的边缘 bug"，而是任何知道接口的人都可稳定利用的远程任意文件读取漏洞，属于 CWE-22 / OWASP A01（Broken Access Control）级别的最高危问题。

2. 为什么不能相信前端参数？

因为参数来自客户端 = 攻击者可控。开发者的假设"前端只会传数据库保存的附件路径"把安全建立在一个没有强制力的约定上：

• 前端代码可以被任意修改（改 JS、用 curl/Postman 直接构造请求）；
• 就算前端是自家 SPA，也无法阻止攻击者绕过前端直接打接口；
• 安全模型里，所有客户端输入都默认不可信，这是信任边界（Trust Boundary）的基本原则——服务端必须假设请求里任何字段都可能被篡改，包括 path、fileId、userId、Cookie 等。

信任边界口诀：能到达服务端的每一个字节，都默认来自攻击者。

3. path.startsWith("/data/uploads/") 能不能彻底修？

不能，它连第一关都过不了，更别提软链接。

• 字符串前缀 ≠ 路径规范化：startsWith 是纯字符串比较，不做路径解析。/data/uploads/../etc/passwd 字符串前缀确实是 /data/uploads/（/data/uploads/.. 的前 15 个字符完全匹配），但文件系统解析后是 /etc/passwd。前缀检查在"…/"面前形同虚设。
• 绕过变体无穷无尽：/data/uploads/../../etc/passwd、/data/uploads/..%2f..%2fetc/passwd（URL 编码）、/data/uploads/%2e%2e/%2e%2e/etc/passwd、双写 ….//、/data/uploads2/../../etc/passwd（前缀只匹配到 /data/uploads，/data/uploads2 也以它开头！）、符号链接、硬链接、/data/uploads/././../../etc/passwd……
• 即使你用 Path.normalize() 之后再 startsWith（这是比字符串前缀好得多的做法），仍然挡不住：

4. Symbolic Link 会不会绕过？

会，而且非常干净地绕过。 假设攻击者能在 /data/uploads/ 下创建软链接（或者数据库里某条附件记录的 storageKey 恰好指向软链接）：

/data/uploads/link_to_etc -> /etc (软链接)
GET /api/files/download?path=/data/uploads/link_to_etc/passwd

normalize() 后的路径是 /data/uploads/link_to_etc/passwd——前缀合法、规范化合法，但 File/FileInputStream 默认跟随符号链接，实际读到的是 /etc/passwd。Files.isRegularFile() 也会跟随链接判断为目标文件。结论：任何"校验字符串路径"的方案都无法防御软链接，因为问题出在"字符串 → inode"的解析层。 防御手段只有两条路：① 用 Files.isSymbolicLink() 显式拒绝 + 解析真实路径（toRealPath()）后再校验；② 根除客户端传路径的设计（推荐，见下）。

5. 应该让客户端传什么：文件路径 / fileId / objectKey？

核心原则：客户端只传"引用"，不传"位置"。 存储路径/objectKey 由服务端生成并只存在 DB 里（storageKey 列），下载时服务端按 fileId 查出 storageKey 再访问。这样路径穿越的攻击面从协议层直接消失——因为攻击者根本没有机会指定路径。同时存储文件名应该用 UUID 随机命名（uuid.pdf），而不是用户原始文件名——原始文件名只存 DB（original_filename 列）用于下载时的 Content-Disposition 展示。

6. 如何验证"这个文件属于当前用户"？

归属校验必须在服务端基于可信身份做，且发生在任何文件访问之前：

请求带登录凭证（Session / JWT）
→ 服务端解析出 currentUserId（可信身份）
→ 查 attachment 表：WHERE id = :fileId
→ 校验 att.owner_id == currentUserId （或按业务查共享/权限表）
→ 通过后才允许读取存储文件

为什么必须查库校验而不是信任请求里的 userId？——因为 ?fileId=1&userId=2 这种请求里，userId 也是客户端可控的；必须从服务端持有的登录态（SecurityContext / Session）里取身份。另外还要注意：先鉴权后访问（不能先读了文件再校验），并且文件不存在时返回 404 而非 403 的差异要注意（是否暴露资源存在性，视业务而定）。

7. Content-Disposition 有没有额外风险？

有，两个：

8. 大文件应该直接通过 Spring MVC 返回吗？

不应该（分情况）。 ResponseEntity<Resource> 虽然底层是流式的（FileSystemResource 不会整体读入内存），但数据仍然要经过应用服务器：

• Tomcat 每个请求占用一个 worker 线程，文件下载是长连接，大文件下载会让 worker 线程长时间被"传输文件"占住 → 线程池耗尽 → 其他 API 全部排队；
• 应用服务器转发字节流，浪费应用带宽和 CPU（无零拷贝）；
• 没有断点续传（Range 请求需自己实现）、没有限速、没有并发防重。

三个正确的工程化选择（按存储形态选）：

推荐组合：本地盘 + Nginx → X-Accel-Redirect；对象存储 → Pre-signed URL + 302；只有小文件且无前置 Nginx 时才用流式。另外务必支持 Range 请求（Accept-Ranges: bytes + 206 响应），这是断点续传和视频拖动的基石——Nginx 和 OSS 都原生支持，而手写 Spring 流式则要自己实现。

9. 更合理的下载 API 设计（额外要求）

目标形态：GET /api/files/{fileId}/download，后端内部流程：

1. 认证：从 SecurityContext 取当前登录用户（JWT/Session），未登录 → 401
2. 定位：按 fileId 查 attachment 表 → 不存在 → 404
3. 授权：att.owner_id == currentUserId（或查共享权限表）→ 不通过 → 403
4. 状态：文件状态 ACTIVE？→ 已删除/禁用 → 404
5. 解析：用 DB 里的 storageKey 解析真实存储路径（服务端生成，杜绝客户端注入）
6. 存在性：文件在存储上存在？→ 不存在 → 404
7. 响应：
– 本地盘：返回 X-Accel-Redirect（Nginx 发文件）+ 安全 Content-Disposition
– 对象存储：生成 pre-signed URL → 302 重定向
8. 审计（可选）：记录下载日志（谁、何时、哪个文件、来源 IP）

🛠 解决方案

候选方案对比 → 推荐方案：推荐"fileId 引用 + 服务端归属校验 + 本地盘用 X-Accel / 对象存储用 Pre-signed URL"组合；彻底放弃客户端传路径。防御纵深四层：① 协议层消灭路径输入（传 fileId）→ ② 鉴权层归属校验 → ③ 存储层 UUID 随机命名 + 目录隔离 → ④ 传输层 Nginx/OSS 卸载。

// ============ 1. 附件实体：storageKey 服务端生成，绝不接受客户端路径 ============
package com.example.file.model;

import jakarta.persistence.Column;
import jakarta.persistence.Entity;
import jakarta.persistence.GeneratedValue;
import jakarta.persistence.GenerationType;
import jakarta.persistence.Id;
import jakarta.persistence.Table;
import lombok.Getter;
import lombok.Setter;

import java.time.LocalDateTime;

@Entity
@Table(name = "attachment")
@Getter
@Setter
public class Attachment {

@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;

@Column(name = "owner_id", nullable = false)
private Long ownerId; // 归属用户：鉴权依据

@Column(name = "storage_key", nullable = false)
private String storageKey; // 服务端生成的相对存储路径，如 "2024/08/17/550e8400-….pdf"

@Column(name = "original_filename", nullable = false)
private String originalFilename; // 用户原始文件名：仅用于下载时的展示命名

@Column(name = "content_type")
private String contentType;

@Column(name = "file_size", nullable = false)
private Long fileSize;

@Column(name = "status", nullable = false)
private String status; // ACTIVE / DELETED

@Column(name = "created_at", nullable = false)
private LocalDateTime createdAt;
}

// ============ 2. 存储方案抽象：本地盘 or 对象存储，二选一实现 ============
package com.example.file.storage;

import java.net.URI;

/**
* 下载目标：服务端完成鉴权后，决定"怎么把文件交给客户端"。
*/
public sealed interface DownloadTarget permits LocalFileTarget, ObjectStorageTarget {

String safeFileName();

long size();

String contentType();
}

/** 本地盘文件：交给 Nginx X-Accel-Redirect 发送 */
public record LocalFileTarget(
String internalUrl, // 例如 /protected/2024/08/17/uuid.pdf（Nginx internal location）
String safeFileName,
long size,
String contentType
) implements DownloadTarget {
@Override public String safeFileName() { return safeFileName; }
@Override public long size() { return size; }
@Override public String contentType() { return contentType; }
}

/** 对象存储：交给 302 重定向到预签名 URL */
public record ObjectStorageTarget(
URI preSignedUrl,
String safeFileName,
long size,
String contentType
) implements DownloadTarget {
@Override public String safeFileName() { return safeFileName; }
@Override public long size() { return size; }
@Override public String contentType() { return contentType; }
}

// ============ 3. 文件服务：归属校验 + 存储解析（核心安全逻辑） ============
package com.example.file.service;

import com.example.file.model.Attachment;
import com.example.file.repo.AttachmentRepository;
import com.example.file.storage.DownloadTarget;
import com.example.file.storage.LocalFileTarget;
import com.example.file.storage.ObjectStorageTarget;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
import java.util.UUID;

@Slf4j
@Service
@RequiredArgsConstructor
public class FileService {

private static final Path STORAGE_ROOT = Path.of("/data/uploads").toAbsolutePath().normalize();

private final AttachmentRepository attachmentRepository;
private final PreSignedUrlProvider preSignedUrlProvider; // 对象存储时注入

@Value("${file.storage.type:local}") // local | s3
private String storageType;

/**
* 解析下载目标。整个方法的核心：客户端只提供 fileId，
* 存储路径完全来自服务端 DB，客户端没有任何注入路径的机会。
*
* @throws FileAccessDeniedException 403：文件不属于当前用户
* @throws AttachmentNotFoundException 404：文件不存在/已删除
*/
public DownloadTarget resolveDownload(Long fileId, Long currentUserId) {
Attachment att = attachmentRepository.findById(fileId)
.orElseThrow(() -> new AttachmentNotFoundException(fileId));

// ① 归属校验：文件必须属于当前登录用户（服务端可信身份，非请求参数）
if (!att.getOwnerId().equals(currentUserId)) {
log.warn("download denied: fileId={}, user={}, owner={}",
fileId, currentUserId, att.getOwnerId());
throw new FileAccessDeniedException(fileId);
}

// ② 状态校验：已删除/禁用一律视为不存在（避免泄露资源存在性）
if (!"ACTIVE".equals(att.getStatus())) {
throw new AttachmentNotFoundException(fileId);
}

// ③ 按存储形态分发
return "s3".equals(storageType)
? toObjectStorageTarget(att)
: toLocalFileTarget(att);
}

/**
* 本地盘：解析 storageKey，防御性校验路径在 STORAGE_ROOT 之下。
* 为什么还要校验？——storageKey 由服务端生成，理论上不会越界，
* 但 DB 可能被历史脏数据/误操作污染，且未来可能有人手工改库；
* 校验是"纵深防御"的最后一道，代价几乎为零。
* 注意：软链接无法靠字符串校验防御，真正防御是"storageKey 永不由客户端提供"。
*/
private LocalFileTarget toLocalFileTarget(Attachment att) {
Path target = STORAGE_ROOT.resolve(att.getStorageKey()).normalize();
if (!target.startsWith(STORAGE_ROOT)) {
log.error("storage key escapes root, fileId={}, storageKey={}",
att.getId(), att.getStorageKey());
throw new IllegalStateException("invalid storage key");
}
if (!Files.exists(target) || !Files.isRegularFile(target)) {
throw new AttachmentNotFoundException(att.getId());
}
// 相对 internal 路径（供 Nginx 使用，必须去掉根前缀）
String internalUrl = "/protected/" + STORAGE_ROOT.relativize(target)
.toString().replace('\\\\', '/');
return new LocalFileTarget(internalUrl, att.getOriginalFilename(),
att.getFileSize(), att.getContentType());
}

/** 对象存储：生成限时预签名 URL（15 分钟），客户端直连下载 */
private ObjectStorageTarget toObjectStorageTarget(Attachment att) {
return new ObjectStorageTarget(
preSignedUrlProvider.signGet(att.getStorageKey(),
att.getOriginalFilename(), Duration.ofMinutes(15)),
att.getOriginalFilename(), att.getFileSize(), att.getContentType());
}
}

// ============ 4. Controller：只收 fileId，不再收 path ============
package com.example.file.web;

import com.example.file.service.FileService;
import com.example.file.storage.DownloadTarget;
import com.example.file.storage.LocalFileTarget;
import com.example.file.storage.ObjectStorageTarget;
import com.example.security.CurrentUser;
import lombok.RequiredArgsConstructor;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.net.URI;

@RestController
@RequiredArgsConstructor
public class FileController {

private final FileService fileService;

/**
* 下载接口：客户端只传 fileId，路径/归属/存储解析全部在服务端完成。
* 两种响应形态：
* 本地盘 → X-Accel-Redirect（Nginx 用 sendfile 零拷贝发文件）
* 对象存储 → 302 重定向到预签名 URL（下载流量不经过应用）
*/
@GetMapping("/api/files/{fileId}/download")
public ResponseEntity<?> download(@PathVariable Long fileId) {
Long currentUserId = CurrentUser.id(); // 服务端可信身份
DownloadTarget target = fileService.resolveDownload(fileId, currentUserId);

HttpHeaders headers = new HttpHeaders();
headers.setContentDisposition(ContentDispositionUtil.safeAttachment(target.safeFileName()));
headers.setContentType(MediaType.parseMediaType(target.contentType()));

if (target instanceof LocalFileTarget local) {
// X-Accel-Redirect：应用只做鉴权定位，实际发送交给 Nginx
headers.add("X-Accel-Redirect", local.internalUrl());
headers.setContentLength(local.size());
return new ResponseEntity<>(headers, HttpStatus.OK);
}

if (target instanceof ObjectStorageTarget oss) {
// 302 → 客户端直连对象存储
return ResponseEntity.status(HttpStatus.FOUND)
.location(URI.create(oss.preSignedUrl().toString()))
.headers(headers)
.build();
}
throw new IllegalStateException("unsupported download target");
}
}

// ============ 5. Content-Disposition 安全工具：防头注入 + RFC 5987 中文文件名 ============
package com.example.file.web;

import org.springframework.http.ContentDisposition;

import java.nio.charset.StandardCharsets;

/**
* 安全构造 Content-Disposition：
* ① 过滤 CR/LF/引号等控制字符，杜绝响应头注入（CRLF Injection）；
* ② 中文等非 ASCII 文件名走 RFC 5987 filename* 编码（避免乱码）。
* 为什么不手拼字符串？——手拼容易漏过滤；Spring 的 ContentDisposition
* 已实现 filename 转义与 filename* 双轨，直接复用标准实现。
*/
public final class ContentDispositionUtil {

private ContentDispositionUtil() {
}

public static ContentDisposition safeAttachment(String rawFilename) {
// 只保留安全字符：过滤 \\r \\n " 以及一切控制字符（防御 CRLF 头注入）
String clean = (rawFilename == null ? "download" : rawFilename)
.replaceAll("[\\\\r\\\\n\\"]", "_")
.replaceAll("[\\\\p{Cntrl}]", "_")
.replaceAll("[\\\\\\\\/]", "_");
// 最长截断，避免超长文件名带来的解析歧义/日志放大
if (clean.length() > 200) {
clean = clean.substring(0, 200);
}
return ContentDisposition.attachment()
.filename(clean, StandardCharsets.UTF_8)
.build();
}
}

// ============ 6.（对象存储）AWS S3 预签名 URL 提供者 ============
package com.example.file.storage;

import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
import software.amazon.awssdk.services.s3.presigner.S3Presigner;
import software.amazon.awssdk.services.s3.presigner.model.GetObjectPresignRequest;

import java.net.URI;
import java.time.Duration;

@Component
@RequiredArgsConstructor
public class PreSignedUrlProvider {

@Value("${s3.bucket}")
private String bucket;

private final S3Presigner presigner;

/**
* 生成限时预签名 GET URL：
* – 签名期限（15 分钟）内有效，泄露窗口可控；
* – response-content-disposition 参数让 OSS 直接返回安全下载头，
* 无需应用再设置（注意：此处示例简化，生产需对文件名做与
* ContentDispositionUtil 相同的净化）。
*/
public URI signGet(String objectKey, String fileName, Duration ttl) {
GetObjectRequest getRequest = GetObjectRequest.builder()
.bucket(bucket)
.key(objectKey)
.build();
GetObjectPresignRequest presignRequest = GetObjectPresignRequest.builder()
.signatureDuration(ttl)
.getObjectRequest(getRequest)
.build();
return presigner.presignGetObject(presignRequest).url().toURI();
}
}

# ============ 7. Nginx 配置：internal location + 零拷贝 + 限速 ============
# 关键点：
# – internal：该 location 只能被 X-Accel-Redirect 头触发，外部 URL 直接访问返回 404；
# – alias 必须与后端 STORAGE_ROOT 一致；
# – sendfile：零拷贝，应用服务器不再参与文件字节流传输；
# – limit_rate：限制单连接下载速度，防止单用户占满出口带宽；
# – 原生支持 Range 请求（断点续传）。

location /protected/ {
internal; # 仅允许 X-Accel-Redirect 内部跳转
alias /data/uploads/; # 注意 alias 与 root 的路径拼接差异
sendfile on;
sendfile_max_chunk 1m;
limit_rate 10m; # 每连接 10MB/s 限速（按需配置）
add_header Accept-Ranges bytes; # 断点续传
}

🧪 需要补充的测试

package com.example.file.web;

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.http.HttpHeaders;
import org.springframework.test.web.servlet.MockMvc;

import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*;

@SpringBootTest
@AutoConfigureMockMvc
class FileSecurityTest {

@Autowired MockMvc mockMvc;

// 回归：旧接口的路径穿越（若仍存在 path 参数，应全部拒绝）
@Test
void oldStylePathTraversal_mustBeRejected() throws Exception {
mockMvc.perform(get("/api/files/download")
.param("path", "/etc/passwd"))
.andExpect(status().is4xxClientError());

mockMvc.perform(get("/api/files/download")
.param("path", "/data/uploads/../../etc/passwd"))
.andExpect(status().is4xxClientError());
}

// 新接口：只接受 fileId，且必须带登录态
@Test
void newApi_requiresAuthentication() throws Exception {
mockMvc.perform(get("/api/files/1/download"))
.andExpect(status().isUnauthorized()); // 未登录 → 401
}

// 越权：user2 下载 user1 的文件 → 403
@Test
void crossUserDownload_mustBeForbidden() throws Exception {
mockMvc.perform(get("/api/files/1/download")
.header(HttpHeaders.AUTHORIZATION, "Bearer token-of-user2"))
.andExpect(status().isForbidden());
}

// 头注入：恶意文件名被净化，响应头不含 CRLF
@Test
void maliciousFileName_mustNotInjectHeader() throws Exception {
// 在 DB 中构造 original_filename = "a.txt\\r\\nSet-Cookie: hack=1"
// 下载后断言响应头不含 Set-Cookie、不含换行符
mockMvc.perform(get("/api/files/2/download")
.header(HttpHeaders.AUTHORIZATION, "Bearer token-of-owner"))
.andExpect(header().string(HttpHeaders.CONTENT_DISPOSITION,
org.hamcrest.Matchers.not(
org.hamcrest.Matchers.containsString("\\r\\n"))));
}
}

💡 工程要点（教学总结）

• 信任边界是安全第一原则：客户端传来的每个字段都默认不可信；"前端只会传合法值"不是安全设计，是事故预告。
• 只传引用、不传位置：API 传 fileId/objectKey，存储路径由服务端 DB 解析——路径穿越从协议层消失，比任何"校验路径"都彻底。
• 路径校验的三大误区：startsWith 是字符串比较不是路径解析（被 ../ 秒破）；normalize() 挡不住软链接（要 toRealPath() + 拒绝符号链接）；"数据库里的路径"也可能被脏数据污染（纵深防御仍要校验）。
• 鉴权三问：身份从哪来（服务端登录态，不是请求参数）？归属怎么验（owner_id == 当前用户）？顺序对不对（先鉴权后访问）？
• 大文件下载是架构问题不是接口问题：应用层转发大文件 = 用最贵的资源（Tomcat 线程）干最脏的活（传字节）；本地盘交给 Nginx sendfile 零拷贝，对象存储交给预签名 URL 直连，顺带白拿断点续传、限速、CDN。
• 响应头也要当输入处理：任何拼接进响应头的用户数据都要净化 + 标准编码（ContentDisposition + RFC 5987）。

题 3：RabbitMQ 消费者（at-least-once / 重复消费 / Poison Message / 消费幂等）

🚩 审查 Issue（结论先行）

结论先行：RabbitMQ 默认给出的是 at-least-once 投递语义，重复消费不是"异常情况"，而是默认现实。这道题的错误修复方案（吞异常）把"重复"变成了"丢失"，属于 P0 级错误。正确解法 = 消费幂等（同事务幂等表）+ 有限重试 + DLQ 兜底 + 监控告警。

📖 深入讲解（为什么有问题）

1. MQ 本质上提供哪种投递语义？

为什么"重复"是默认现实？——因为 RabbitMQ 无法知道"消费者是否真的处理完了"。它只认 ACK：basicAck 到达之前，消息处于"未确认"状态，连接断开就会重新投递。而"处理成功"和"ACK 送达"是两个独立事件，中间隔着网络，任何一秒的断连都会造成"处理了但没 ACK"→ 重投。这就是为什么每个消费者都必须假设"同一消息我会收到多次"。

2. 回放线上时间线：为什么积分 +200？

08:10:01 message received 第 1 次投递
08:10:01 point DB timeout DB 故障，addPoints 抛异常
08:10:02 consumer throws exception 方法抛异常 → Spring 不 ack（消息还在队列）
08:10:03 RabbitMQ redelivery 第 2 次投递
08:10:03 points insert success DB 恢复，插入成功
08:10:03 consumer connection broken ACK 还没来得及发，连接断了！
08:10:04 RabbitMQ redelivery 第 3 次投递（RabbitMQ 认为第 2 次没确认）
08:10:04 points insert success 又插入一次

关键在 08:10:03→04：第二次处理已成功（insert success），但 ACK 因连接断开丢失，RabbitMQ 判定"未确认"重新投递，第三次又成功插入。处理成功 ≠ 消息确认，这就是 at-least-once 的重复窗口。没有幂等表，这个重复就是实打实的 +200。

3. DB 成功、ACK 失败会怎样？

消息会重新投递，消费者会再次执行业务。 如果业务副作用（加积分）不可幂等，就重复生效。这是 at-least-once 的标准代价。对策只有一条：让业务副作用可幂等（见"消费幂等"），而不是祈祷 ACK 不丢。

4. catch 掉 Exception 后会怎样？（为什么是错误修复）

后果链：消费抛异常 → catch 吞掉 → 方法正常返回 → Spring 认为处理成功 → auto ack → 消息从队列删除 → 业务副作用永远缺失 → 无重试、无告警、无补偿入口。

• 它确实"不再无限重试"了，但以消息丢失为代价：DB 故障期间进来的所有积分消息全部静默消失，用户积分永久少加；
• 更糟的是掩盖故障：log.error 在日志海里，没有指标告警，DB 故障恢复后没人知道有多少消息丢了；
• 正确的"不无限重试"应该是：有限次重试 → 进 DLQ → 告警 → 人工/补偿任务处理，而不是一次失败就放弃。

教训：在 MQ 消费者里 catch 住所有异常 = 主动丢消息。 消费者代码里唯一允许"吞掉并继续"的异常是"幂等冲突"（已消费过），其他异常一律抛出交给重试/DLQ 机制。

5. Poison Message（毒消息）怎么处理？

Poison Message = 持续抛异常、永远处理不了的消息（如消息体缺字段、业务数据损坏）。如果不设上限，它会被无限 requeue：

• 消费线程反复处理同一毒消息 → 线程被占满 → 队列里其他正常消息全部积压（队头阻塞）；
• 每次处理都打 DB/写日志 → 日志爆炸、DB 压力放大；
• 重启也解决不了——消息还在队列里。

正确姿势（三层）：

6. 无限 retry 有什么风险？

• 吞吐归零：毒消息反复占用消费者线程，正常消息排队（队头阻塞）；
• 日志/监控风暴：每秒重试的异常刷爆日志，掩盖真实问题；
• 下游被反复打击：每次重试都打 DB/调外部，故障的 DB 被雪上加霜；
• 消息TTL/积压：配合 TTL 或队列长度限制，最终触发消息过期丢弃或队列阻塞。

结论：retry 必须有界（次数 + 退避），有界之后必须分流（DLQ），分流之后必须处理（告警 + 人工/补偿）。

7. DLQ 应该承担什么职责？

DLQ（死信队列）的职责是**“隔离 + 暂存 + 触发治理”**，不是"垃圾箱"：

8. 积分系统怎么实现消费幂等？

方案：消费幂等表（本地事务）。

设计：consume_record(event_id, consumer_name, order_id, processed_at)
约束：UNIQUE(event_id, consumer_name)

消费流程（同一 DB 事务）：
① INSERT consume_record(…) — 抢唯一键
② 若唯一键冲突（已存在）→ 已消费过 → 直接跳过
③ 若无冲突 → UPDATE user_point SET points = points + 100
④ 事务提交 → ①和③同生共死

为什么用 event_id 而不是 order_id？——一个订单可能产生多个事件（OrderCreated、OrderPaid、OrderShipped……），每个事件的消费记录要分开；而且同一事件可能被不同消费者（积分、库存、通知）各自消费，所以唯一键必须是 (event_id, consumer_name) 二元组。

UNIQUE(event_id, consumer_name) 是否足够？——足够兜住"并发重复"，但必须满足一个前提：记录插入与业务副作用在同一个事务里。 分析：

• 场景 1：两个实例同时消费同一消息（极端情况下 RabbitMQ 可能对同一消息并发投递，或重投与手动重放撞车）→ 只有一个 INSERT 成功，另一个撞唯一键 → 幂等成立 ✅；
• 场景 2：INSERT 成功但加积分失败 → 事务回滚 → 记录也没了 → 重投后重新执行 ✅（前提：同一事务！）；
• 场景 3：如果记录和积分不在同一事务（先插记录、提交，再单独加积分）：加积分失败 → 记录已提交 → 重投时撞唯一键直接跳过 → 积分永久丢失 ❌。这就是题目问"记录消费与加积分是否同一事务"的原因——必须同一事务。

补充：如果不想用"先插后改"的顺序，也可以用 INSERT … ON DUPLICATE KEY UPDATE / INSERT IGNORE 一步完成抢键（见解决方案代码），语义更清晰。

为什么不用 Redis 做幂等？——Redis 的 SETNX 也可以去重，但：① 与 DB 事务没有原子性，记录消费成功但加积分失败时无法回滚去重标记；② 资金/权益类数据必须落在持久化存储上，Redis 丢数据就是丢权益；③ 幂等记录还要支持对账查询。结论：权益类幂等用 DB 事务，Redis 幂等只适合日志、埋点这类可容忍丢失的场景。

9. 补充：Spring 的 auto ack 与 manual ack 怎么选？

• acknowledge-mode: auto（Spring 默认行为）：方法正常返回 → ack；抛异常 → 由容器决定 requeue（配合 retry 配置）。够用且简单，配合"有限重试 + DLQ"即可；
• acknowledge-mode: manual：自己调 channel.basicAck/basicNack，适合需要精确控制确认时机（如处理完但想延迟确认）的场景。不强制，auto + 幂等 + DLQ 已是生产级配置；
• 红线：无论哪种模式，都不要"catch 住异常当成功"——那等于手动制造消息丢失。

🛠 解决方案

候选方案对比 → 推荐方案：

推荐方案 = 有限重试 + DLQ + 同事务幂等表 + 告警补偿。 完整实现如下：

— ============ 1. 消费幂等表 ============
CREATE TABLE consume_record (
id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键',
event_id VARCHAR(64) NOT NULL COMMENT '消息事件ID（生产端生成，全局唯一，如 UUID）',
consumer_name VARCHAR(64) NOT NULL COMMENT '消费者标识：POINT / STOCK / NOTIFY',
order_id BIGINT NOT NULL COMMENT '业务订单号（便于对账）',
processed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '消费完成时间',
PRIMARY KEY (id),
UNIQUE KEY uk_event_consumer (event_id, consumer_name)
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COMMENT = 'MQ 消费幂等记录表';

— 对账查询：某订单的积分消费记录
— SELECT * FROM consume_record WHERE order_id = ? AND consumer_name = 'POINT';

// ============ 2. 实体与仓储 ============
package com.example.point.repo;

import jakarta.persistence.Column;
import jakarta.persistence.Entity;
import jakarta.persistence.GeneratedValue;
import jakarta.persistence.GenerationType;
import jakarta.persistence.Id;
import jakarta.persistence.Table;
import jakarta.persistence.UniqueConstraint;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Getter;
import lombok.NoArgsConstructor;

import java.time.LocalDateTime;

@Entity
@Table(name = "consume_record",
uniqueConstraints = @UniqueConstraint(
name = "uk_event_consumer",
columnNames = {"event_id", "consumer_name"}))
@Getter
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ConsumeRecord {

@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;

@Column(name = "event_id", nullable = false)
private String eventId;

@Column(name = "consumer_name", nullable = false)
private String consumerName;

@Column(name = "order_id", nullable = false)
private Long orderId;

@Column(name = "processed_at", nullable = false)
private LocalDateTime processedAt;
}

package com.example.point.repo;

import org.springframework.data.jpa.repository.JpaRepository;

public interface ConsumeRecordRepository extends JpaRepository<ConsumeRecord, Long> {

boolean existsByEventIdAndConsumerName(String eventId, String consumerName);
}

// ============ 3. 消息体：必须带 event_id（生产端生成） ============
package com.example.order.event;

/**
* 订单创建事件。event_id 在生产端生成（UUID），全局唯一。
* 为什么必须在消息里带 event_id？——消费者幂等的"键"必须由消息自身携带，
* 不能由消费者自行推断（如用 orderId 推断，多个事件会互相撞）。
*/
public record OrderCreatedMessage(
String eventId, // 全局唯一事件 ID
Long orderId,
Long userId,
Integer amount
) {
public static OrderCreatedMessage of(Long orderId, Long userId, Integer amount) {
return new OrderCreatedMessage(java.util.UUID.randomUUID().toString(),
orderId, userId, amount);
}
}

// ============ 4. 积分服务：同事务幂等加积分（核心修复） ============
package com.example.point.service;

import com.example.order.event.OrderCreatedMessage;
import com.example.point.repo.ConsumeRecord;
import com.example.point.repo.ConsumeRecordRepository;
import com.example.point.repo.UserPointRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;

@Slf4j
@Service
@RequiredArgsConstructor
public class PointService {

public static final String CONSUMER_NAME = "POINT";

private final ConsumeRecordRepository consumeRecordRepository;
private final UserPointRepository userPointRepository;
private final JdbcTemplate jdbcTemplate;

/**
* 幂等加积分：消费记录与积分变更必须在【同一个数据库事务】里。
*
* 为什么必须同事务？（三个为什么）
* ① 为什么先抢唯一键？—— 唯一键是"并发重复"的最后防线：
* 两个消费者同时处理同一消息，只有一个能 INSERT 成功；
* ② 为什么用 INSERT IGNORE 而不是先 exists 再 insert？
* —— exists→insert 两步之间存在竞态窗口；INSERT IGNORE 一步完成、
* 且不抛异常、不污染当前事务（JPA 的 flush 抛 DuplicateKeyException
* 会把事务标记为 rollback-only，同事务内再操作就不可靠了）；
* ③ 为什么"记录 + 积分"必须同事务？
* —— 若记录提交了而积分失败，重投时撞唯一键直接跳过 → 积分永久丢失；
* 同事务则"记录失败 ⇒ 积分一起回滚 ⇒ 重投重新执行"，不丢不重。
*/
@Transactional
public void addPointsIdempotent(OrderCreatedMessage message) {
// 快速路径：已消费过（多数重复投递走这里，避免无谓的写）
if (consumeRecordRepository.existsByEventIdAndConsumerName(
message.eventId(), CONSUMER_NAME)) {
log.info("duplicate message ignored: eventId={}, orderId={}",
message.eventId(), message.orderId());
return;
}

// 抢唯一键：成功 → 返回 1；撞键（并发重复）→ 返回 0，不抛异常
int inserted = jdbcTemplate.update("""
INSERT IGNORE INTO consume_record
(event_id, consumer_name, order_id, processed_at)
VALUES (?, ?, ?, ?)
""",
message.eventId(), CONSUMER_NAME, message.orderId(),
LocalDateTime.now());

if (inserted == 0) {
log.info("unique key conflict, already consumed: eventId={}",
message.eventId());
return; // 并发重复：另一个事务已消费，本事务无副作用，正常返回
}

// 唯一键抢到：执行真正的业务副作用（与上面的 INSERT 同事务）
userPointRepository.addPoints(message.userId(), message.amount());
log.info("points added: userId={}, amount={}, eventId={}",
message.userId(), message.amount(), message.eventId());
}
}

// ============ 5. 消费者：不再吞异常，交给重试/DLQ 机制 ============
package com.example.point.consumer;

import com.example.order.event.OrderCreatedMessage;
import com.example.point.service.PointService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RequiredArgsConstructor
public class PointConsumer {

private final PointService pointService;

/**
* 幂等消费：重复投递只生效一次。
* 方法正常返回 → Spring 自动 ack → 消息出队；
* 方法抛异常 → 容器按配置重试（max-attempts=3）→ 耗尽后拒绝进 DLQ。
* 绝不 catch 业务异常！——吞异常 = 主动丢消息。
*/
@RabbitListener(queues = "order.created.point")
public void consume(OrderCreatedMessage message) {
pointService.addPointsIdempotent(message);
}
}

# ============ 6. 配置：有限重试 + 拒绝进 DLQ（Spring Boot 托管方式） ============
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: auto # 方法正常返回即 ack；抛异常走重试
retry:
enabled: true # 开启消费重试
max-attempts: 3 # 最多 3 次（1 次原始 + 2 次重试）
initial-interval: 1000ms # 首次重试间隔
multiplier: 2.0 # 指数退避：1s → 2s
max-interval: 10000ms
default-requeue-rejected: false # 关键：重试耗尽后拒绝并投递到 DLQ，
# 而不是无限 requeue 回原队列

// ============ 7.（等价显式写法）容器工厂 + 队列声明 DLX/DLQ ============
package com.example.point.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
* 死信链路：
* order.created.point（业务队列）
* │ 处理失败重试 3 次
* ▼
* RejectAndDontRequeueRecoverer → 拒绝消息
* │ x-dead-letter-exchange 路由
* ▼
* order.created.point.dlq（死信队列）→ 落库 + 告警 + 人工/补偿
*/
@Configuration
public class RabbitConfig {

public static final String POINT_QUEUE = "order.created.point";
public static final String POINT_DLQ = "order.created.point.dlq";
public static final String DLX = "order.dead-letter.exchange";

@Bean
public DirectExchange deadLetterExchange() {
return new DirectExchange(DLX, true, false);
}

@Bean
public Queue pointQueue() {
return QueueBuilder.durable(POINT_QUEUE)
.deadLetterExchange(DLX) // 拒绝/过期/超长 → 进死信
.deadLetterRoutingKey(POINT_DLQ)
.build();
}

@Bean
public Queue pointDlq() {
return QueueBuilder.durable(POINT_DLQ).build();
}

@Bean
public Binding dlqBinding() {
return BindingBuilder.bind(pointDlq()).to(deadLetterExchange()).with(POINT_DLQ);
}

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setAcknowledgeMode(org.springframework.amqp.core.AcknowledgeMode.AUTO);
factory.setDefaultRequeueRejected(false); // 重试耗尽不再回原队列
factory.setMessageConverter(new Jackson2JsonMessageConverter()); // JSON 消息体
// 有限重试 + 指数退避；耗尽后 RejectAndDontRequeueRecoverer 拒绝消息 → 走 DLX
factory.setAdviceChain(RetryInterceptorBuilder.stateless()
.maxAttempts(3)
.backOffOptions(1000L, 2.0, 10000L)
.recoverer(new RejectAndDontRequeueRecoverer())
.build());
return factory;
}
}

// ============ 8. 死信消费者：落库 + 告警（DLQ 的治理职责） ============
package com.example.point.consumer;

import com.example.order.event.OrderCreatedMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RequiredArgsConstructor
public class PointDlqConsumer {

private final DeadLetterRecordRepository deadLetterRecordRepository;
private final AlertService alertService;

/**
* DLQ 消费者：不重试（毒消息重试无意义），只做三件事：
* ① 落库留痕（错误明细，供对账与重放）；
* ② 告警（值班群/监控平台）；
* ③ 等待人工修复根因后，由补偿工具重投或直接补偿。
*/
@RabbitListener(queues = RabbitConfig.POINT_DLQ)
public void onDeadLetter(OrderCreatedMessage message) {
deadLetterRecordRepository.save(DeadLetterRecord.from(message, CONSUMER_POINT));
alertService.alert("积分消费进入死信队列: eventId=" + message.eventId());
log.error("dead letter received: eventId={}, orderId={}",
message.eventId(), message.orderId());
}
}

补充：如果业务要求"不丢任何一条"，可再加一层对账 Job：定时扫描"订单已创建但 consume_record 缺失"的订单（以 order 表为准），把漏消费的事件重新投递。幂等表 + 对账 = 双保险。

🧪 需要补充的测试

package com.example.point.service;

import com.example.order.event.OrderCreatedMessage;
import com.example.point.repo.ConsumeRecord;
import com.example.point.repo.ConsumeRecordRepository;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.support.TransactionTemplate;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
class PointServiceIdempotencyTest {

@Autowired PointService pointService;
@Autowired ConsumeRecordRepository consumeRecordRepository;
@Autowired JdbcTemplate jdbcTemplate;
@Autowired TransactionTemplate transactionTemplate;

private final OrderCreatedMessage msg =
OrderCreatedMessage.of(1001L, 7L, 100);

@Test
void duplicateDelivery_pointsAddedOnlyOnce() {
// 模拟 at-least-once 重复投递：同一消息消费 3 次
pointService.addPointsIdempotent(msg);
pointService.addPointsIdempotent(msg);
pointService.addPointsIdempotent(msg);

Integer points = jdbcTemplate.queryForObject(
"SELECT points FROM user_point WHERE user_id = 7", Integer.class);
assertThat(points).isEqualTo(100); // 只加了 100，不是 300

assertThat(consumeRecordRepository
.existsByEventIdAndConsumerName(msg.eventId(), "POINT")).isTrue();
}

@Test
void concurrentDuplicate_onlyOneInsertWins() throws Exception {
// 两线程并发消费同一消息：唯一键保证只有一个生效
java.util.concurrent.CountDownLatch start = new java.util.concurrent.CountDownLatch(1);
java.util.concurrent.ExecutorService pool =
java.util.concurrent.Executors.newFixedThreadPool(2);
java.util.concurrent.atomic.AtomicInteger exceptions = new java.util.concurrent.atomic.AtomicInteger();
for (int i = 0; i < 2; i++) {
pool.submit(() -> {
try {
start.await();
pointService.addPointsIdempotent(msg);
} catch (Exception e) {
exceptions.incrementAndGet();
}
});
}
start.countDown();
pool.shutdown();
pool.awaitTermination(10, java.util.concurrent.TimeUnit.SECONDS);

Integer points = jdbcTemplate.queryForObject(
"SELECT points FROM user_point WHERE user_id = 7", Integer.class);
assertThat(points).isEqualTo(100); // 并发也只加一次
assertThat(exceptions.get()).isEqualTo(0); // 且没有任何异常（INSERT IGNORE 不抛）
}

@Test
void addPointsFailure_rollsBackRecordTogether() {
// 同事务验证：加积分失败（模拟 user_id 不存在的外键/除零）→
// 消费记录必须一起回滚 → 重投后还能重新执行（不丢）
transactionTemplate.executeWithoutResult(status -> {
try {
pointService.addPointsIdempotent(
new OrderCreatedMessage("bad-event-1", 999L, –1L, 100));
} catch (Exception expected) {
// 预期失败
}
});
// 记录随事务回滚：不存在
assertThat(consumeRecordRepository
.existsByEventIdAndConsumerName("bad-event-1", "POINT")).isFalse();
}
}

💡 工程要点（教学总结）

• at-least-once 是 MQ 的默认现实，重复消费不是 bug 而是预期；所有消费者都要回答一个问题：“这条消息收到两次，业务会怎样？”——答不上来就是 P0。
• 吞异常 = 主动丢消息：消费者代码里唯一允许吞的异常是"幂等冲突"；其余异常交给"有限重试 → DLQ → 告警补偿"。
• 幂等三要素：唯一的键（event_id，生产端生成）、持久化的去重表（UNIQUE(event_id, consumer_name)）、与业务副作用同事务——三者缺一，幂等就不成立。
• retry 必须有界、有界之后必须分流（DLQ）、分流之后必须治理（告警+补偿）：无限 requeue 是毒消息的放大器，DLQ 不是垃圾箱而是治理入口。
• 先判断业务性质：资金/权益类必须 DB 事务幂等（可对账），日志/埋点类才允许 Redis 去重或 at-most-once。
• 可观测性：消费积压、重试次数、DLQ 深度都要有指标告警——问题要在用户投诉之前被发现。

题 4：异步任务线程池（newCachedThreadPool 失控 / 有界线程池 / Rate Limit / Bulkhead / 异步 Job 设计）

🚩 审查 Issue（结论先行）

结论先行：这道题的核心不是"换一个线程池实现"，而是重新理解资源边界——"异步"只是把工作从请求线程挪到后台线程，总资源（CPU/内存/下游并发能力）没有增加一分一毫。正确链路 = 有界线程池（数量由下游并发上限决定） + 有界队列（背压） + 拒绝策略 + 超时/重试/熔断 + 持久化 Job 状态机（失败可恢复）。几十秒级的 AI Job 根本不该用 @Async 裸跑。

📖 深入讲解（为什么有问题）

1. newCachedThreadPool() 最大的问题是什么？

线程数无上限（unbounded），且无背压（backpressure）。 逐层拆解：

• 机制：newCachedThreadPool = SynchronousQueue + 线程数无上限（Integer.MAX_VALUE）+ 60s keepAlive。SynchronousQueue 是"零容量"队列——提交的任务必须立刻有线程接走，没有线程就现场 new 一个。所以每个未及时处理的任务 = 一个新线程。
• 为什么线程是昂贵资源：每个线程默认栈 1MB（-Xss）+ 线程对象 + 内核线程；1000 个线程 ≈ 1GB+ 虚拟内存；更重要的是上下文切换——CPU 在 1000 个可运行线程间切换的开销远超计算本身。
• 为什么会导致 Full GC：线程多了 → 每个线程的栈、局部变量、线程对象都占堆外/堆内存 → 内存压力 → GC 频繁 → 停顿变长 → 任务处理更慢 → 需要更多线程 → 恶性循环（正反馈失控）。最终 JVM 长时间 STW，无响应。
• 没有背压：SynchronousQueue 不接受排队，意味着系统对"任务涌入速度"没有任何缓冲和限制——流量多大，线程就建多少，直到 OOM。
• 一句话：cached pool 是为"短命、低频、突发"任务设计的（如临时线程池），不是为"高并发、长耗时"任务设计的。

2. 为什么"异步"并不等于"系统吞吐能力无限"？

因为异步只是重新分配了线程，没有增加任何物理资源。做一件工作的资源消耗是恒定的：CPU 计算、内存、网络连接、下游服务能力。异步把"请求线程"换成"后台线程"，收益是请求线程被释放、接口延迟下降，代价是后台线程池成为新的瓶颈。吞吐上限 = min(CPU 能力, 内存能力, 下游并发能力, 连接池大小, 线程数)。本题里，真正的天花板是 AI Provider 的 50 并发——系统再快，下游只接受 50 个并发请求，其余请求无论开多少线程都只能排队等待。异步改变了"谁在等"，不改变"有多少工作在等"。

3. 固定 1000 个线程是不是解决方案？

不是，而且方向就错了。 四个理由：

4. AI HTTP 调用属于 CPU 密集还是 IO 密集？

IO 密集（阻塞式等待网络响应）。 判断标准：任务的大部分时间花在"等待"（网络往返、磁盘、下游处理）还是"计算"（CPU 执行指令）？AI 调用发出 HTTP 请求后，线程就阻塞在 socket 上等响应（平均 4s、P95 25s），CPU 几乎空闲。所以：

• IO 密集任务的线程数可以 > CPU 核数（经验公式：线程数 ≈ CPU 核数 × (1 + 等待时间/计算时间)，等待占比越高，线程数可以越大）；
• 但上限必须由下游并发能力决定：AI Provider 50 并发 → 本服务对 AI 的并发调用最多 50，线程池 maxPoolSize 取 50 左右即可，多出来的线程只是陪跑；
• 对比：CPU 密集任务（图像处理、加解密）线程数 ≈ CPU 核数 + 1，多了反而因上下文切换变慢。

5. 线程池五大参数分别控制什么？

关键机制：ThreadPoolExecutor 的扩容顺序是"core 满 → 入队 → 队列满 → 扩到 max → 队列满且线程满 → 拒绝策略"。 所以 core=10, max=50, queue=200 的真实含义是：最多 50 个任务同时执行，200 个在队列等待，第 251 个触发拒绝策略——整个系统对任务的容纳有明确上限（250），这就是"有界"。

6. 300 req/s × 4s，系统大约同时存在多少任务？

用 Little’s Law（利特尔法则）：L = λ × W（系统中任务数 = 到达率 × 平均驻留时间）。

• 平均：L = 300/s × 4s = 1200 个任务同时存在；
• 最坏（P95）：L = 300/s × 25s = 7500 个；
• 极端（60s）：L = 300/s × 60s = 18000 个！

这就是 newCachedThreadPool 会创建的线程数量级：高峰期系统里同时躺着几千个任务，cached pool 就给你建几千个线程 → 内存/切换/GC 全线崩溃。这个数字也直接否定了"固定 1000 线程"：1000 线程连平均并发（1200）都兜不住，而且兜住了也没用——下游只有 50 并发。正确姿态：接受"排队"，用有界队列缓冲，用拒绝策略把压力传导回去，而不是试图用线程数硬扛。

7. AI Provider 50 并发限制下，本服务线程池开 1000 有什么意义？

没有任何正面意义，只有负面代价。 分析：

• 50 并发限制意味着任何时刻最多 50 个 HTTP 请求能真正发出，其余 950 个线程全部阻塞在"等待下游配额/等待响应"上——它们不产生任何吞吐；
• 代价却实打实：950 个线程的栈内存、950 个线程的上下文切换（调度器要在它们之间来回切换）、950 个等待中的连接/对象；
• 更危险的是超时重试风暴：1000 个线程同时超时重试，会把下游 50 并发的限流打到熔断，雪崩。

正确对齐：本服务线程池并发上限 ≈ 下游并发上限（50），多出的任务在队列里等（背压），而不是开线程等（浪费）。这就是 Bulkhead（舱壁隔离）思想：线程池的容量 = 你对下游的承诺，超出就排队/拒绝，绝不透支。

8. 几十秒甚至几分钟的 AI Job，适不适合直接使用 @Async？

不适合。 判断标准：任务生命周期是否超过"进程可靠性"的范围。

为什么 @Async 撑不住长任务？——@Async 的本质是"丢进线程池就不管了"：没有返回值/状态存储、JVM 重启任务蒸发、失败没有补偿入口、线程池是内存态（重启即清空）。而几十秒的 AI Job，用户必然会问"我的报告生成到哪一步了？"——这要求任务有持久化状态机。所以正确链路是：HTTP 提交 → 落 Job 表（状态机）→ 有界执行器调度 → 更新状态 → 用户轮询查询。@Async 只适合作为"Job 执行器内部"的短任务载体，或者干脆不要 @Async，用显式 executor.execute()。

🛠 解决方案

候选方案对比 → 推荐方案：

推荐链路设计：

POST /reports/{id}/generate
↓ ① 创建 Job（落库：PENDING）
↓ ② dispatch：提交到有界线程池
↓ ③ 线程池（core=10, max=50, queue=200, CallerRuns）── 资源边界
↓ ④ Bulkhead：Semaphore(50) 与 AI Provider 对齐 ──────── 下游配额
↓ ⑤ Timeout：65s（> P95 25s，覆盖极端 60s）──────────── 僵尸调用防护
↓ ⑥ Retry：指数退避 1s→2s，最多 3 次 ────────────────── 瞬时故障恢复
↓ ⑦ Circuit Breaker：AI 连续失败 → 快速失败不重试 ───── 依赖故障防护
↓ ⑧ 更新 Job 状态（SUCCESS/FAILED），失败可人工重试
↓
AI Provider（50 concurrent）

配套：GET /api/reports/jobs/{jobId} 查询状态；定时 recoverStuckJobs 扫描 PENDING 超时/RUNNING 超时任务重新入队（失败恢复）。

// ============ 1. 有界线程池（替代 newCachedThreadPool / fixed 1000） ============
package com.example.report.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

@Configuration
@EnableAsync
public class AsyncConfig {

/**
* 有界线程池：容量 = 对 AI Provider 的承诺。
* corePoolSize=10 常驻线程，覆盖日常流量；
* maxPoolSize=50 = AI Provider 并发上限（Bulkhead 对齐）；
* queueCapacity=200 有界队列 = 背压缓冲（高峰期任务在此排队，而不是开新线程）；
* keepAlive=60s 高峰过后回收非核心线程；
* CallerRunsPolicy 队列满 + 线程满 → 由 Tomcat 提交线程执行 → 接口自然变慢
* → 背压传导给调用方，而不是丢弃任务或撑爆内存。
*/
@Bean("reportExecutor")
public Executor reportExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(10);
executor.setMaxPoolSize(50);
executor.setQueueCapacity(200);
executor.setKeepAliveSeconds(60);
executor.setThreadNamePrefix("report-exec-");
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
}

— ============ 2. 报告生成 Job 表（持久化状态机） ============
CREATE TABLE report_job (
id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'Job ID',
report_id BIGINT NOT NULL COMMENT '报告 ID',
user_id BIGINT NOT NULL COMMENT '发起用户',
status VARCHAR(16) NOT NULL COMMENT 'PENDING / RUNNING / SUCCESS / FAILED',
attempt INT NOT NULL DEFAULT 0 COMMENT '已尝试次数（重试上限）',
last_error VARCHAR(512) NULL COMMENT '最近一次失败原因',
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
started_at DATETIME NULL COMMENT '开始执行时间（用于超时恢复判断）',
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id),
KEY idx_status_updated (status, updated_at)
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COMMENT = 'AI 报告生成异步任务表';

// ============ 3. Job 实体 ============
package com.example.report.job;

import jakarta.persistence.Column;
import jakarta.persistence.Entity;
import jakarta.persistence.EnumType;
import jakarta.persistence.Enumerated;
import jakarta.persistence.GeneratedValue;
import jakarta.persistence.GenerationType;
import jakarta.persistence.Id;
import jakarta.persistence.Table;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Getter;
import lombok.NoArgsConstructor;

import java.time.LocalDateTime;

@Entity
@Table(name = "report_job")
@Getter
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ReportJob {

public enum Status { PENDING, RUNNING, SUCCESS, FAILED }

@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;

@Column(name = "report_id", nullable = false)
private Long reportId;

@Column(name = "user_id", nullable = false)
private Long userId;

@Enumerated(EnumType.STRING)
@Column(name = "status", nullable = false)
private Status status;

@Column(name = "attempt", nullable = false)
private Integer attempt;

@Column(name = "last_error")
private String lastError;

@Column(name = "created_at", nullable = false)
private LocalDateTime createdAt;

@Column(name = "started_at")
private LocalDateTime startedAt;
}

// ============ 4. Job 执行服务：状态机 + Bulkhead + 超时 + 重试 ============
package com.example.report.job;

import com.example.report.ai.AiClient;
import com.example.report.repo.ReportRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
@RequiredArgsConstructor
public class ReportJobService {

private static final Duration AI_TIMEOUT = Duration.ofSeconds(65);
private static final int MAX_ATTEMPTS = 3;

private final ReportJobRepository jobRepository;
private final ReportRepository reportRepository;
private final AiClient aiClient;

@Qualifier("reportExecutor")
private final Executor reportExecutor;

/**
* Bulkhead（舱壁）：与 AI Provider 的 50 并发上限对齐。
* 信号量保证【本服务对 AI 的并发调用】永远 ≤ 50：
* – 拿不到配额 → 任务保持 PENDING，等下次调度，而不是开新线程硬挤；
* – 这是"线程池容量 = 下游承诺"的落地实现。
*/
private final Semaphore aiConcurrency = new Semaphore(50);

// —————- 对外入口 —————-

/** 创建 Job 并提交执行。 */
public Long createAndDispatch(Long reportId, Long userId) {
ReportJob job = jobRepository.save(ReportJob.builder()
.reportId(reportId)
.userId(userId)
.status(ReportJob.Status.PENDING)
.attempt(0)
.createdAt(java.time.LocalDateTime.now())
.build());
dispatch(job.getId());
return job.getId();
}

/** 提交执行：交给有界线程池。 */
public void dispatch(Long jobId) {
// 显式 execute 而不是 @Async——任务状态由 Job 表管理，不依赖方法返回值
reportExecutor.execute(() -> execute(jobId));
}

// —————- 执行核心 —————-

public void execute(Long jobId) {
ReportJob job = jobRepository.findById(jobId).orElse(null);
if (job == null || job.getStatus() == ReportJob.Status.SUCCESS) {
return;
}
// ① Bulkhead：拿不到下游配额就不发起调用（任务留在 PENDING，等恢复调度）
if (!aiConcurrency.tryAcquire(5, TimeUnit.SECONDS)) {
log.warn("AI bulkhead full, job {} stays pending", jobId);
return;
}
try {
// ② 状态推进 + 尝试次数递增
jobRepository.markRunning(jobId,
job.getAttempt() + 1, java.time.LocalDateTime.now());
// ③ 超时 + 有限重试
String result = callAiWithRetry(job);
// ④ 业务落库 + 状态 SUCCESS
reportRepository.updateResult(job.getReportId(), result, "SUCCESS");
jobRepository.markSuccess(jobId);
} catch (Exception e) {
log.error("job {} failed: {}", jobId, e.getMessage(), e);
// ⑤ 失败：标记 FAILED（可人工重试），错误留痕
jobRepository.markFailed(jobId, truncate(e.getMessage()));
} finally {
aiConcurrency.release(); // 释放下游配额
}
}