高并发系统设计 - 从零到精通

课程目标

学完这套课程，你将能够：

• 理解高并发的核心瓶颈和解决思路
• 掌握高并发架构设计模式
• 熟练使用各种缓存、消息队列、数据库优化技术
• 能够设计QPS 10万+的系统
• 掌握限流、降级、熔断等保护机制
• 面试时自信回答各种高并发问题

---

第一部分：高并发基础概念

1.1 什么是高并发？

核心概念：

• QPS（Queries Per Second）： 每秒查询数
• TPS（Transactions Per Second）： 每秒事务数
• 并发数： 同时处理的请求数
• RT（Response Time）： 响应时间

并发量级参考：	并发级别	QPS	场景示例
小型	100-1000	个人博客、小公司官网
中型	1000-10000	中型企业应用
大型	1万-10万	电商平台、社交应用
超大型	10万+	双11大促、秒杀系统

核心公式（面试必背！）：

系统容量 = 并发数 × 平均响应时间
QPS = 并发数 / 平均响应时间

举例：
- 100个并发用户，平均响应时间100ms
- QPS = 100 / 0.1 = 1000 QPS

1.2 高并发的三大瓶颈

面试必答：

"高并发系统主要面临三大瓶颈：1）CPU瓶颈，大量计算消耗CPU；2）I/O瓶颈，包括磁盘I/O和网络I/O；3）内存瓶颈，内存不足导致频繁GC或swap。解决思路是：CPU密集型任务并行化，I/O密集型任务异步化，内存不足使用缓存或增加服务器。"

瓶颈一：CPU瓶颈

场景： 复杂计算、加解密、图片处理

解决方法：

1. 算法优化：减少不必要的计算
2. 缓存：缓存计算结果
3. 并行计算：利用多核CPU
4. 异步处理：将耗时操作放到后台

// 原始：CPU密集型计算
public Result heavyCalculation(Input input) {
    // 复杂计算，耗时长
    return doComplexCalculation(input);
}

// 优化：异步计算
public CompletableFuture<Result> heavyCalculationAsync(Input input) {
    return CompletableFuture.supplyAsync(() ->
        doComplexCalculation(input),
        threadPool
    );
}

瓶颈二：I/O瓶颈

场景： 数据库查询、文件读写、网络请求

解决方法：

1. 使用缓存：Redis、Memcached
2. 异步I/O：NIO、Netty
3. 批量操作：减少I/O次数
4. 连接池：复用连接

// 原始：同步I/O，阻塞
public User getUser(Long userId) {
    return repository.findById(userId);  // 阻塞等待
}

// 优化：缓存优先
public User getUser(Long userId) {
    // 先查缓存
    User user = redisTemplate.opsForValue().get("user:" + userId);
    if (user != null) {
        return user;
    }
    // 缓存未命中查数据库
    user = repository.findById(userId);
    // 写入缓存
    redisTemplate.opsForValue().set("user:" + userId, user, 3600);
    return user;
}

瓶颈三：内存瓶颈

场景： 大对象、缓存过多、内存泄漏

解决方法：

1. 对象池：复用对象
2. 缓存淘汰策略：LRU、LFU
3. 分级缓存：本地缓存+分布式缓存
4. 优化数据结构：减少内存占用

1.3 性能测试方法

工具：

• JMeter： 压力测试
• ab (Apache Bench)： 简单压测
• wrk： 高性能压测工具
• Gatling： 更强大的压测工具

使用wrk压测：

# 安装wrk
sudo apt install wrk

# 基础压测
wrk -t12 -c400 -d30s http://localhost:8080/api/test

# 参数说明：
# -t12: 12个线程
# -c400: 400个并发连接
# -d30s: 持续30秒

# 输出示例：
# Running 30s test @ http://localhost:8080/api/test
#   12 threads and 400 connections
#   Thread Stats   Avg      Stdev     Max   +/- Stdev
#     Latency    45.32ms   12.45ms  200.34ms   87.23%
#     Req/Sec     8.50k     1.23k   10.00k    68.92%
#   3056780 requests in 30.00s, 2.35GB read
# Requests/sec: 101892.60
# Transfer/sec:   80.15MB

---

第二部分：缓存策略（并发加速器）

2.1 缓存架构设计

三级缓存架构：

请求
  ↓
本地缓存（Caffeine/Guava） ← 命中返回
  ↓ 未命中
分布式缓存（Redis/Cluster）
  ↓ 未命中
数据库（MySQL）
  ↓
回写缓存

为什么需要多级缓存？

1. 本地缓存：速度快（内存），不占用网络带宽
2. 分布式缓存：数据一致性好，容量大
3. 数据库：持久化存储

2.2 本地缓存（Caffeine/Guava）

Caffeine（推荐）：

// Caffeine缓存配置
Cache<String, User> userCache = Caffeine.newBuilder()
    .maximumSize(10000)              // 最大10000条
    .expireAfterWrite(10, TimeUnit.MINUTES)  // 写入10分钟过期
    .refreshAfterWrite(5, TimeUnit.MINUTES) // 5分钟刷新
    .recordStats()                    // 记录统计信息
    .build(key -> loadUserFromDB(key));  // 缓存加载器

// 使用
User user = userCache.get("user:123");

// 缓存统计
CacheStats stats = userCache.stats();
System.out.println("命中率: " + stats.hitRate());
System.out.println("加载数: " + stats.loadCount());

面试问题：本地缓存的缺点？

回答：

"本地缓存的缺点：1）容量受限，受限于服务器内存；2）数据不一致，多台服务器缓存可能不同；3）无法主动失效，需要等待过期；4）内存占用过高影响应用稳定性。因此通常作为一级缓存，配合分布式缓存使用。"

2.3 分布式缓存（Redis）

Redis数据类型和使用场景

数据类型	适用场景	示例
String	简单KV、计数器	用户信息、点赞数
Hash	对象存储	商品详情、用户画像
List	消息队列、列表	评论列表、待办任务
Set	去重、交集差集	标签、共同关注
ZSet	排行榜、延迟队列	排行榜、定时任务
Bitmap	位统计	用户签到、在线状态
HyperLogLog	基数统计	UV统计（精准度99%）

Redis集群方案

方案一：主从复制 + 哨兵

Master
  ↓ 写入
Slave1 ← 复制
Slave2 ← 复制
  ↓ 读取
Sentinel 监控

配置示例：

# redis.conf - Master
port 6379
requirepass yourpassword

# redis.conf - Slave
port 6380
replicaof master_ip 6379
masterauth yourpassword

# sentinel.conf - Sentinel
port 26379
sentinel monitor mymaster master_ip 6379 2
sentinel auth-pass mymaster yourpassword
sentinel down-after-milliseconds mymaster 5000

方案二：Redis Cluster（分片集群）

客户端
  ↓
Slot分片 (0-16383)
  ↓
┌─────┬─────┬─────┬─────┬─────┬─────┐
│Slot0│Slot1│Slot2│Slot3│Slot...│Slot │
│  -  │  -  │  -  │  -  │ -27 │16383│
└──┬──┴──┬──┴──┬──┴──┬──┴──┬──┴──┬──┘
   ↓     ↓     ↓     ↓     ↓     ↓
Node1  Node2  Node3  Node1  Node2  Node3

面试回答：

"Redis集群有两种方案：主从复制+哨兵适合中小规模，读写分离，高可用；Redis Cluster适合大规模，自动分片，无中心节点。对于高并发场景，我选择Redis Cluster，因为它支持水平扩展，故障时自动迁移数据。同时配合本地缓存减少Redis压力。"

2.4 缓存穿透、击穿、雪崩（面试三剑客！）

问题一：缓存穿透

现象： 查询不存在的数据，请求全部打到数据库

请求查询 user_id=-1
  ↓
缓存不存在
  ↓
数据库也不存在
  ↓
再次请求，继续查数据库
  ↓
数据库被打爆！

解决方案：

方案1：缓存空值

public User getUser(Long userId) {
    String key = "user:" + userId;

    // 先查缓存
    User user = redisTemplate.opsForValue().get(key);
    if (user != null) {
        return user;  // 返回空值或真实值
    }

    // 查数据库
    user = userRepository.findById(userId);

    // 即使是null也缓存
    redisTemplate.opsForValue().set(key, user, 300, TimeUnit.SECONDS);
    return user;
}

方案2：布隆过滤器

// 初始化布隆过滤器
BloomFilter<Long> bloomFilter = BloomFilter.create(
    Funnels.longFunnel(),
    1000000,  // 预计数据量
    0.01      // 误判率1%
);

// 加载所有存在的ID到布隆过滤器
for (Long userId : existingUserIds) {
    bloomFilter.put(userId);
}

// 查询时先检查
public User getUser(Long userId) {
    // 布隆过滤器判断不存在，直接返回
    if (!bloomFilter.mightContain(userId)) {
        return null;
    }
    // 可能存在，继续查缓存和数据库
    return getUserFromCacheOrDB(userId);
}

面试回答：

"缓存穿透的解决方案：1）缓存空值，将查询为空的结果也缓存，设置较短过期时间；2）布隆过滤器，将所有可能存在的key预加载到布隆过滤器，查询时先过滤肯定不存在的key。布隆过滤器有误判率但性能极高，适合海量数据场景。"

问题二：缓存击穿

现象： 热点key过期，大量请求同时打到数据库

时间点T: 热点key过期
      ↓
10000个并发请求同时到达
      ↓
全部查数据库
      ↓
数据库压力暴增

解决方案：

方案1：互斥锁（分布式锁）

public User getUser(Long userId) {
    String key = "user:" + userId;

    // 先查缓存
    User user = redisTemplate.opsForValue().get(key);
    if (user != null) {
        return user;
    }

    // 获取分布式锁
    String lockKey = "lock: user:" + userId;
    try {
        boolean locked = redisTemplate.opsForValue()
            .setIfAbsent(lockKey, "1", 10, TimeUnit.SECONDS);

        if (locked) {
            // 获取锁成功，查数据库
            user = userRepository.findById(userId);
            // 写入缓存，设置过期时间
            redisTemplate.opsForValue().set(key, user, 3600, TimeUnit.SECONDS);
        } else {
            // 获取锁失败，等待一下再查缓存
            Thread.sleep(100);
            return getUser(userId);
        }
    } finally {
        // 释放锁
        redisTemplate.delete(lockKey);
    }

    return user;
}

方案2：逻辑过期（热点永不过期）

public User getUser(Long userId) {
    String key = "user:" + userId;

    // 查缓存
    User user = redisTemplate.opsForValue().get(key);

    // 判断是否逻辑过期
    if (user != null && !isExpired(user)) {
        return user;
    }

    // 异步刷新缓存
    if (user == null || isExpired(user)) {
        CompletableFuture.runAsync(() -> refreshCache(userId));
    }

    // 返回旧数据
    return user;
}

面试回答：

"缓存击穿是热点key过期导致的数据库压力暴增。解决方案：1）互斥锁，只允许一个线程查数据库重建缓存，其他线程等待；2）逻辑过期，缓存永不过期，通过后台线程定期刷新，返回旧数据的同时异步刷新。互斥锁保证一致性但性能较低，逻辑过期性能好但数据有短暂延迟。"

问题三：缓存雪崩

现象： 大量key同时过期，数据库压力暴增

时间点T: 10000个key同时过期
      ↓
所有请求查数据库
      ↓
数据库崩溃

解决方案：

方案1：随机过期时间

public void setCache(String key, Object value) {
    // 基础过期时间1小时
    int baseExpire = 3600;
    // 随机增加0-30分钟
    int randomExpire = (int) (Math.random() * 1800);

    redisTemplate.opsForValue().set(
        key,
        value,
        baseExpire + randomExpire,
        TimeUnit.SECONDS
    );
}

方案2：缓存预热

@Scheduled(cron = "0 0 2 * * ?")  // 每天凌晨2点
public void warmUpCache() {
    // 加载热点数据到缓存
    List<User> hotUsers = userRepository.findHotUsers();
    for (User user : hotUsers) {
        redisTemplate.opsForValue().set(
            "user:" + user.getId(),
            user,
            3600,
            TimeUnit.SECONDS
        );
    }
}

方案3：多级缓存 + 限流保护

// 一级Caffeine本地缓存
// 二级Redis分布式缓存
// 查询时先查本地，再查Redis，最后查DB
// 并在DB前加限流

面试回答：

"缓存雪崩是大量key同时过期导致的。解决方案：1）设置随机过期时间，避免同时过期；2）缓存预热，在系统低峰期提前加载热点数据；3）多级缓存+限流，本地缓存作为第一道防线，配合限流保护数据库。实际项目中我们组合使用多种方案，确保系统稳定性。"

2.5 缓存更新策略（面试必问！）

策略一：Cache Aside（旁路缓存，最常用）

读缓存：先查缓存，未命中查数据库，回写缓存
写数据：先更新数据库，再删除缓存

// 读
public User getUser(Long id) {
    User user = cache.get(id);
    if (user == null) {
        user = db.findById(id);
        cache.put(id, user);
    }
    return user;
}

// 写
public void updateUser(User user) {
    db.update(user);        // 先更新DB
    cache.remove(user.getId());  // 再删除缓存
}

为什么是删除而不是更新？

"删除缓存是因为：1）并发写时可能导致数据不一致；2）删除开销小；3）下次读取时自动加载最新数据。只有在缓存更新计算非常昂贵时才考虑更新缓存。"

策略二：Write Through（写穿透）

写数据：同时写入缓存和数据库
读缓存：直接读缓存

public void updateUser(User user) {
    db.update(user);        // 写数据库
    cache.put(user.getId(), user);  // 写缓存
}

策略三：Write Behind（异步写回）

写数据：只写缓存，异步写数据库
读缓存：直接读缓存

适用场景： 写入频繁、一致性要求不高

---

第三部分：消息队列（异步解耦神器）

3.1 为什么需要消息队列？

核心作用：

1. 异步处理： 耗时操作异步执行
2. 削峰填谷： 平滑流量峰值
3. 解耦： 生产者和消费者解耦

场景对比：

场景1：同步处理（慢）

用户下单
  ↓
创建订单 (50ms)
  ↓
扣减库存 (50ms)
  ↓
发送短信 (100ms)  ← 耗时操作
  ↓
发送通知 (100ms)  ← 耗时操作
  ↓
返回 (300ms)     ← 响应慢

场景2：异步处理（快）

用户下单
  ↓
创建订单 (50ms)
  ↓
扣减库存 (50ms)
  ↓
发送到MQ (10ms)
  ↓
返回 (110ms)     ← 响应快

MQ消费者异步：
  ↓
发送短信
发送通知

3.2 消息队列选型

MQ	优点	缺点	适用场景
RabbitMQ	功能丰富、可靠性强	性能一般、资源占用高	复杂业务、可靠投递
Kafka	吞吐量极高、持久化好	实时性一般、消费复杂	日志收集、大数据
RocketMQ	功能全面、阿里系	学习曲线陡峭	电商、金融
Pulsar	云原生、存算分离	较新、生态一般	云架构

面试回答：

"我们选择RabbitMQ作为消息队列，因为它：1）支持多种消息模式（直连、主题、RPC、广播）；2）消息可靠性高，支持持久化、ACK机制、死信队列；3）管理界面友好，运维方便；4）社区活跃，文档完善。如果需要超大数据吞吐，会选择Kafka。"

3.3 RabbitMQ实战

消息模型

1. 直连模式（Direct）

// 生产者
rabbitTemplate.convertAndSend(
    "order-exchange",
    "create",
    orderMessage
);

// 消费者
@RabbitListener(
    queues = "order.create.queue",
    queueBinding = @QueueBinding(
        value = @Queue(value = "order.create.queue"),
        exchange = @Exchange(value = "order-exchange", type = "direct"),
        key = "create"
    )
)
public void handleOrderCreate(OrderMessage message) {
    // 处理订单创建
}

2. 主题模式（Topic）

// 生产者
rabbitTemplate.convertAndSend(
    "user-exchange",
    "user.created",  // 通配符匹配 user.*
    userMessage
);

// 消费者
@RabbitListener(
    bindings = @QueueBinding(
        value = @Queue(value = "user-event.queue"),
        exchange = @Exchange(value = "user-exchange", type = "topic"),
        key = "user.*"  // 匹配所有用户事件
    )
)
public void handleUserEvent(UserMessage message) {
    // 处理用户事件
}

3. 延迟队列

// 发送延迟消息（5分钟后执行）
Message message = MessageBuilder
    .withPayload(task)
    .setHeader("x-delay", 300000)  // 5分钟延迟
    .build();

rabbitTemplate.send("delay-exchange", "delay-key", message);

消息可靠性保证

1. 消息持久化

# application.yml
spring:
  rabbitmq:
    publisher-confirm-type: correlated  # 发送确认
    publisher-returns: true              # 退回确认
    listener:
      simple:
        acknowledge-mode: manual         # 手动ACK

2. 死信队列（处理失败消息）

@Bean
public Queue deadLetterQueue() {
    return QueueBuilder.durable("order.dead-letter.queue").build();
}

@Bean
public Queue orderQueue() {
    return QueueBuilder.durable("order.queue")
        .withArgument("x-dead-letter-exchange", "dead-letter-exchange")
        .withArgument("x-dead-letter-routing-key", "dead-letter")
        .build();
}

@RabbitListener(queues = "order.dead-letter.queue")
public void handleDeadLetterMessage(Message message) {
    // 处理失败消息，记录日志或重试
    log.error("处理失败消息: {}", message);
}

3. 消息幂等性（防止重复消费）

@RabbitListener(queues = "order.queue")
public void handleOrder(OrderMessage message) {
    // 使用唯一ID保证幂等性
    String messageId = message.getId();

    // 检查是否已处理
    if (redisTemplate.hasKey("processed:" + messageId)) {
        return;  // 已处理，跳过
    }

    try {
        // 处理消息
        processOrder(message);

        // 标记已处理
        redisTemplate.opsForValue().set(
            "processed:" + messageId,
            "1",
            24, TimeUnit.HOURS
        );
    } catch (Exception e) {
        // 处理失败，抛出异常触发重试
        throw new RuntimeException(e);
    }
}

面试回答：

"消息可靠性保证从三个方面实现：1）生产端，开启confirm机制确保消息发送成功，开启return机制处理路由失败；2）服务端，消息持久化到磁盘，集群部署保证高可用；3）消费端，手动ACK确认，失败消息进入死信队列。同时使用唯一ID实现幂等性，防止重复消费。"

3.4 消息堆积处理（面试高频）

现象： 消费速度 < 生产速度，队列堆积

解决方案：

方案1：增加消费者

# application.yml - 增加并发消费
spring:
  rabbitmq:
    listener:
      simple:
        concurrency: 10      # 最小并发数
        max-concurrency: 50  # 最大并发数

方案2：批量消费

@RabbitListener(
    queues = "order.queue",
    containerFactory = "batchContainerFactory"
)
public void handleBatchOrders(List<OrderMessage> messages) {
    // 批量处理，减少数据库操作
    batchProcessOrders(messages);
}

@Bean
public SimpleRabbitListenerContainerFactory batchContainerFactory() {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setBatchListener(true);  // 开启批量
    factory.setBatchSize(100);       // 每次100条
    factory.setConsumerBatchEnabled(true);
    return factory;
}

方案3：临时扩容队列

// 1. 创建临时队列，增加分区
// 2. 停止原消费者
// 3. 使用脚本将堆积消息转移到临时队列
// 4. 启动更多消费者消费临时队列

方案4：丢弃非关键消息

// 监控队列长度
if (queueSize > threshold) {
    // 删除低优先级消息
    rabbitTemplate.purgeQueue("low-priority.queue");
}

面试回答：

"消息堆积处理：1）紧急情况下增加消费者数量；2）使用批量消费减少网络开销和数据库操作；3）监控队列长度，超过阈值时告警；4）非关键消息可丢弃，保护核心业务；5）长期方案是优化消费者性能，增加分区。实际项目中我们设置监控告警，队列堆积超过1万条时自动扩容消费者。"

---

第四部分：数据库优化（并发瓶颈突破）

4.1 读写分离

架构：

应用
  ↓
主库（写）
  ↓ 复制
从库1（读）
从库2（读）
从库3（读）

实现方式：

方案1：Spring动态数据源

public class DynamicDataSource extends AbstractRoutingDataSource {
    @Override
    protected Object determineCurrentLookupKey() {
        return DataSourceContextHolder.getDataSourceType();
    }
}

// 读操作
@Service
public class UserService {
    @DataSource("slave")
    public User getUser(Long id) {
        return userRepository.findById(id);
    }
}

// 写操作
@Service
public class UserService {
    @DataSource("master")
    public void saveUser(User user) {
        userRepository.save(user);
    }
}

方案2：ShardingSphere（推荐）

# application.yml
spring:
  shardingsphere:
    datasource:
      names: master,slave1,slave2
      master:
        type: com.zaxxer.hikari.HikariDataSource
        url: jdbc:mysql://master:3306/db
      slave1:
        type: com.zaxxer.hikari.HikariDataSource
        url: jdbc:mysql://slave1:3306/db
      slave2:
        type: com.zaxxer.hikari.HikariDataSource
        url: jdbc:mysql://slave2:3306/db

    rules:
      readwrite-splitting:
        data-sources:
          readwrite_ds:
            write-data-source-name: master
            read-data-source-names: slave1,slave2
            load-balancer-name: round_robin
        props:
          load-balance-algorithm-type: round_robin

面试问题：读写分离的数据一致性问题？

回答：

"读写分离存在数据一致性问题：主库写入后，从库可能有延迟。解决方案：1）强制读主库，关键业务读写都走主库；2）缓存兜底，读取时先查缓存；3）接受短暂不一致，如用户更新头像后几秒内看不到新头像；4）使用半同步复制，主库等待至少一个从库确认才返回。实际项目中我们结合业务场景选择合适的方案。"

4.2 分库分表

为什么需要分库分表？

• 单表数据量超过1000万，查询变慢
• 单机存储空间不足
• 单机CPU/内存瓶颈

分表策略：

策略1：范围分表

-- 按ID范围分表
orders_0: 0 - 100万
orders_1: 100万 - 200万
orders_2: 200万 - 300万
...

// 路由算法
public String getTable(Long orderId) {
    // 假设每张表100万条
    int tableIndex = (int)(orderId / 1000000);
    return "orders_" + tableIndex;
}

策略2：Hash分表（推荐）

// 哈希分表
public String getTable(Long orderId) {
    // 对订单ID取模，均匀分布
    int tableCount = 10;
    int tableIndex = (int)(orderId % tableCount);
    return "orders_" + tableIndex;
}

ShardingSphere配置：

spring:
  shardingsphere:
    rules:
      sharding:
        tables:
          orders:
            actual-data-nodes: ds_${0..1}.orders_${0..9}
            table-strategy:
              standard:
                sharding-column: order_id
                sharding-algorithm-name: order_mod
        sharding-algorithms:
          order_mod:
            type: MOD
            props:
              sharding-count: 10

面试问题：分库分表后的join问题？

回答：

"分库分表后跨表join确实是个问题。解决方案：1）将关联数据放在同一个分片，使用相同的分片键；2）应用层join，先查询主表获取关联ID，再分别查询子表；3）使用ES或宽表，将关联数据冗余存储；4）对于不需要实时性的报表，使用数据仓库定时聚合。实际项目中我们优先设计表结构避免跨分片join。"

4.3 索引优化

索引设计原则：

1. 最左前缀原则： 联合索引从左向右匹配
2. 选择性高的列在前： 区分度高的列优先索引
3. 避免覆盖索引失效： 不要在索引列上做运算

正确示例：

-- 好的索引
CREATE INDEX idx_user_time ON orders(user_id, create_time);

-- 好的查询
SELECT * FROM orders
WHERE user_id = 123
  AND create_time > '2024-01-01';

-- 坏的查询（索引失效）
SELECT * FROM orders
WHERE create_time > '2024-01-01';  -- 跳过最左列

索引优化案例：

-- 问题查询：慢
SELECT * FROM orders
WHERE user_id = 123
  AND status = 'PAID'
  AND create_time > '2024-01-01'
ORDER BY create_time DESC
LIMIT 10;

-- 优化：添加覆盖索引
CREATE INDEX idx_user_status_time ON orders(user_id, status, create_time)
INCLUDE (id, amount);  -- 覆盖索引

-- 查询可以直接从索引获取数据，无需回表

面试回答：

"索引优化原则：1）区分度高的列优先建索引；2）联合索引遵循最左前缀原则；3）避免在索引列上做运算导致索引失效；4）使用覆盖索引减少回表；5）定期分析慢查询，使用EXPLAIN分析执行计划。实际项目中，我们监控慢查询日志，定期优化索引。"

4.4 连接池优化

HikariCP（推荐）：

spring:
  datasource:
    hikari:
      minimum-idle: 10          # 最小空闲连接
      maximum-pool-size: 100     # 最大连接数
      connection-timeout: 30000 # 连接超时
      idle-timeout: 600000      # 空闲超时
      max-lifetime: 1800000     # 连接最大生命周期
      leak-detection-threshold: 60000  # 连接泄露检测

连接池配置计算：

最大连接数 = (CPU核心数 * 2) + 有效磁盘数

例如：
- 8核CPU，1个磁盘
- 最大连接数 = 8 * 2 + 1 = 17

但实际配置要结合应用场景：
- 读多写少：可以增加连接数
- CPU密集型：减少连接数避免上下文切换

---

第五部分：限流、降级、熔断（系统保护三剑客）

5.1 限流

限流算法：

算法1：计数器（固定窗口）

public class RateLimiter {
    private final AtomicInteger counter = new AtomicInteger(0);
    private final long interval;
    private final int limit;

    public boolean allow() {
        if (counter.get() >= limit) {
            return false;
        }
        counter.incrementAndGet();
        return true;
    }
}

缺点： 边界问题，2倍流量可能通过

算法2：滑动窗口（推荐）

public class SlidingWindowRateLimiter {
    private final ConcurrentLinkedDeque<Long> timestamps = new ConcurrentLinkedDeque<>();
    private final long windowSize;  // 窗口大小（毫秒）
    private final int limit;         // 限制数量

Q    public boolean allow() {
        long now = System.currentTimeMillis();
        long windowStart = now - windowSize;

        // 清理过期请求
        while (!timestamps.isEmpty() && timestamps.peek() < windowStart) {
            timestamps.poll();
        }

        // 检查是否超过限制
        if (timestamps.size() >= limit) {
            return false;
        }

        timestamps.add(now);
        return true;
    }
}

算法3：令牌桶（生产消费模型）

public class TokenBucketRateLimiter {
    private final long capacity;      // 桶容量
    private final long rate;          // 令牌生成速率（个/毫秒）
    private long tokens;              // 当前令牌数
    private long lastRefillTime;       // 上次填充时间

    public synchronized boolean allow() {
        long now = System.currentTimeMillis();
        long elapsed = now - lastRefillTime;

        // 填充令牌
        tokens = Math.min(capacity, tokens + elapsed * rate);
        lastRefillTime = now;

;

        if (tokens >= 1) {
            tokens--;
            return true;
        }
        return false;
    }
}

算法4：漏桶（恒定输出）

public class LeakyBucketRateLimiter {
    private final long capacity;      // 桶容量
    private final long rate;          // 漏水速率（个/毫秒）
    private long water;                // 当前水量
    private long lastLeakTime;         // 上次漏水时间

    public synchronized boolean allow() {
        long now = System.currentTimeMillis();
        long elapsed = now - lastLeakTime;

        // 漏水
        water = Math.max(0, water - elapsed * rate);
        lastLeakTime = now;

        if (water < capacity) {
            water++;
            return true;
        }
        return false;
    }
}

实战：Sentinel限流

# application.yml
spring:
  cloud:
    sentinel:
      datasource:
        flow:
          nacos:
            server-addr: localhost:8848
            data-id: sentinel-flow-rules
            rule-type: flow

Nacos配置规则：

[{
  "resource": "/api/order/create",
  "limitApp": "default",
  "grade": 1,
  "count": 1000,
  "strategy": 0,
  "controlBehavior": 0,
  "clusterMode": false
}]

Java代码：

@RestController
public class OrderController {
    @GetMapping("/api/order/create")
    public Result createOrder(@RequestParam Long userId) {
        try (Entry entry = SentinelResourceUtil.enter("createOrder")) {
            // 业务逻辑
            return Result.success(orderService.create(userId));
        } catch (BlockException e) {
            // 限流处理
返回 Result.error("系统繁忙，请稍后重试");
        }
    }
}

面试回答：

"限流我们使用阿里Sentinel，支持QPS限流、并发限流、Warm Up预热、匀速排队等多种模式。针对不同接口设置不同限流规则：核心接口严格限流保证稳定性，非核心接口可以适当放行。限流后返回友好提示，引导用户重试。通过Nacos动态配置规则，无需重启服务。"

5.2 降级

降级策略：

降级类型	适用场景	处理方式
自动降级	超时、异常、限流	返回默认值、缓存值
手动降级	运维开关	关闭非核心功能
兜底数据	数据获取失败	返回历史数据

实现方式：

方式1：注解降级

@RestController
public class UserController {
    @HystrixCommand(
        fallbackMethod = "getUserFallback",
        commandProperties = {
            @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds", value = "3000")
        }
    )
    @GetMapping("/user/{id}")
    public User getUser(@PathVariable Long id) {
        return userService.getUser(id);
    }

    // 降级方法
    public User getUserFallback(@PathVariable Long id) {
        // 返回缓存数据或默认值
        return cacheService.getUserFromCache(id);
    }
}

方式2：Sentinel降级

@SentinelResource(value = "getUser", fallback = "getUserFallback")
public User getUser(Long id) {
    return userRepository.findById(id);
}

// 降级方法
public User getUserFallback(Long id, Throwable ex) {
    log.warn("getUser降级, id={}, ex={}", id, ex.getMessage());
    return User.builder().id(id).name("默认用户").build();
}

降级策略配置：

// 配置降级规则
List<DegradeRule> rules = new ArrayList<>();
DegradeRule rule = new DegradeRule();
rule.setResource("getUser");
rule.setGrade(RuleConstant.DEGRADE_GRADE_EXCEPTION_RATIO);  // 异常比例
rule.setCount(0.5);  // 异常比例超过50%
rule.setTimeWindow(10);  // 统计窗口10秒
rule.setMinRequestAmount(5);  // 最小请求数5
rules.add(rule);
DegradeRuleManager.loadRules(rules);

面试回答：

"降级策略包括：1）超时降级，超过3秒未响应自动降级；2）异常降级，异常比例超过50%触发降级；3）自动降级，结合监控指标自动触发；4）手动降级，运维人员开关控制。降级后返回缓存数据或默认值，保证核心功能可用。非核心功能如推荐、评论优先降级。"

5.3 熔断

熔断机制（电路熔断器）：

正常状态 ──(失败率>阈值)──> 熔断状态
    ↑                            ↓
    └────────(半开尝试)─────────┘

熔断状态流转：

1. Closed（关闭）： 正常请求通过，统计失败率
2. Open（打开）： 熔断器打开，请求直接失败
3. Half-Open（半开）： 尝试恢复，允许少量请求通过

Resilience4j熔断：

// 配置熔断器
CircuitBreakerConfig config = CircuitBreakerConfig.custom()
    .failureRateThreshold(50)          // 失败率50%
    .waitDurationInOpenState(Duration.ofSeconds(10))  // 熔断10时间
    .slidingWindowSize(10)             // 滑动窗口大小
    .build();

CircuitBreaker circuitBreaker = CircuitBreaker.of("userService", config);

// 使用熔断器
Supplier<User> supplier = CircuitBreaker.decorateSupplier(
    circuitBreaker,
    () -> userService.getUser(id)
);

try {
    User user = supplier.get();
} catch (CallNotPermittedException e) {
    // 熔断打开，执行降级
    user = getUserFromCache(id);
} catch (Exception e) {
    // 其他异常
    throw e;
}

面试回答：

"熔断使用Resilience4j实现，配置失败率阈值50%、熔断时间10秒。当失败率超过阈值时熔断器打开，直接拒绝请求；熔断时间后进入半开状态，允许少量请求通过尝试恢复；如果恢复成功则关闭熔断器。熔断和降级配合使用，保护系统不被雪崩拖垮。"

---

第六部分：异步处理（提升并发能力）

6.1 异步方案对比

方案	优点	缺点	适用场景
线程池	简单直接	资源有限、阻塞	简单异步任务
CompletableFuture	链式调用、组合灵活	学习曲线	复杂异步编排
消息队列	解耦、削峰	系统复杂	跨服务异步
事件驱动	松耦合、扩展性好	调试困难	复杂业务流

6.2 CompletableFuture（推荐）

基础使用：

// 异步执行任务
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 耗时操作
    Thread.sleep(1000);
    return "Hello";
});

// 获取结果
String result = future.get();  // 阻塞等待
String result = future.get(2, TimeUnit.SECONDS);  // 带超时

链式调用：

CompletableFuture.supplyAsync(() -> {
    // 第一步：获取用户
    return userService.getUser(userId);
}, executor)
.thenApplyAsync(user -> {
    // 第二步：获取用户订单
    return orderService.getOrders(user.getId());
}, executor)
.thenAcceptAsync(orders -> {
    // 第三步：发送通知
    notificationService.send(orders);
}, executor)
.exceptionally(ex -> {
    // 异常处理
    log.error("异步任务失败", ex);
    return null;
});

组合多个Future：

// 场景：同时获取用户、商品、优惠券，再创建订单
CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(
    () -> userService.getUser(userId)
);
CompletableFuture<Product> productFuture = CompletableFuture.supplyAsync(
    () -> productService.get(productId)
);
CompletableFuture<Coupon> couponFuture = CompletableFuture.supplyAsync(
    () -> couponService.getCoupon(couponId)
);

// 等待所有完成
CompletableFuture.allOf(userFuture, productFuture, couponFuture)
    .thenRun(() -> {
        // 创建订单
        Order order = orderService.create(
            userFuture.join(),
            productFuture.join(),
            couponFuture.join()
        );
    });

面试回答：

"异步处理我们使用CompletableFuture，它支持链式调用和组合操作。将耗时操作如发送通知、计算积分等异步执行，提升接口响应速度。同时配置独立的线程池，避免影响核心业务。对于跨服务异步，使用消息队列解耦。通过异步化，接口响应时间从500ms降低到100ms。"

6.3 异步事件（Spring Event）

// 定义事件
public class OrderCreatedEvent extends ApplicationEvent {
    private final Order order;

    public OrderCreatedEvent(Object source, Order order) {
        super(source);
        this.order = order;
    }
}

// 发布事件
@Service
public class OrderService {
    @Autowired
    private ApplicationEventPublisher eventPublisher;

    public void createOrder(Order order) {
        // 创建订单
        orderRepository.save(order);

        // 发布事件（异步）
        eventPublisher.publishEvent(new OrderCreatedEvent(this, order));
    }
}

// 监听事件
@Service
public class OrderEventListener {
    @EventListener
    @Async  // 异步处理
    public void handleOrderCreated(OrderCreatedEvent event) {
        Order order = event.getOrder();

        // 发送通知
        notificationService.send(order);

        // 更新积分
        pointService.add(order.getUserId(), order.getAmount());
    }
}

---

第七部分：实战案例（面试题库）

案例1：秒杀系统设计

需求：

• QPS 10万+
• 不能超卖
• 高可用

架构设计：

用户
  ↓
[限流层] Sentinel限流
  ↓
[缓存层] Redis预扣库存
  ↓
[消息层] MQ异步下单
  ↓
[服务层) 下单、扣减库存
  ↓
[数据层] MySQL持久化

核心代码：

Redis预扣库存（原子操作）：

@Service
public class SeckillService {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    public boolean preDeductStock(Long productId, Long userId) {
        String key = "seckill:stock:" + productId;

        // 使用Lua脚本保证原子性
        String script = "local stock = redis.call('get', KEYS[1]) " +
                         "if stock and tonumber(stock) > 0 then " +
                         "  redis.call('decr', KEYS[1]) " +
                         "  return 1 " +
                         "else " +
                         "  return 0 " +
                         "end";

        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptText(script);
        redisScript.setResultType(Long.class);

        Long result = redisTemplate.execute(
            redisScript,
            Collections.singletonList(key)
        );

        return result == 1;
    }
}

Redisson分布式锁：

@Service
public class SeckillService {
    @Autowired
    private RedissonClient redissonClient;

    public boolean seckill(Long productId, Long userId) {
        String lockKey = "seckill:lock:" + productId;
        RLock lock = redissonClient.getLock(lockKey);

        try {
            // 尝试获取锁，等待3秒，锁10秒自动释放
            if (lock.tryLock(3, 10, TimeUnit.SECONDS)) {
                // 检查是否已购买
                if (hasPurchased(productId, userId)) {
                    return false;
                }

                // 预扣库存
                if (preDeductStock(productId, userId)) {
                    // 发送消息异步下单
                    sendToMQ(productId, userId);
                    return true;
                }
            }
            return false;
        } finally {
            lock.unlock();
        }
    }
}

MySQL防超卖（乐观锁）：

@Service
public class OrderService {
    @Transactional
    public void createOrder(Long productId, Long userId) {
        // 乐观锁扣减库存
        int updated = productRepository.deductStock(productId, 1);
        if (updated == 0) {
            throw new BusinessException("库存不足");
        }

        // 创建订单
        Order order = Order.builder()
            .productId(productId)
            .userId(userId)
            .status("PENDING")
            .build();
        orderRepository.save(order);
    }
}

// Repository
@Modifying
@Query("UPDATE Product p SET p.stock = p.stock - :count " +
       "WHERE p.id = :productId AND p.stock >= :count")
int deductStock(@Param("productId") Long productId,
               @Param("count") Integer count);

面试回答：

"秒杀系统核心是防止超卖和高并发。架构上采用：1）限流层，SentinelQPS限流；2）缓存层，Redis预扣库存，使用Lua脚本保证原子性；3）消息层，MQ异步处理下单；4）数据层，MySQL乐观锁扣库存。通过多级防护：Redis保证高并发预扣，MySQL保证数据一致性，MQ异步处理削峰。同时使用分布式锁防止重复购买。"

---

案例2：评论系统设计

需求：

• 文章评论树形结构
• 支持无限嵌套
• 高并发读写

表设计：

CREATE TABLE comments (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    article_id BIGINT NOT NULL,
    user_id BIGINT NOT NULL,
    content TEXT NOT NULL,
    parent_id BIGINT DEFAULT 0,  -- 父评论ID，0表示顶级评论
    root_id BIGINT DEFAULT 0,    -- 根评论ID，用于查询整个评论树
    level INT DEFAULT 0,         -- 评论层级
    reply_to BIGINT DEFAULT 0,   -- 回复的用户ID
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    is_deleted TINYINT DEFAULT 0,
    INDEX idx_article (article_id),
    INDEX idx_root (root_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

查询方案：

方案1：递归查询（少量数据）

-- MySQL 8.0+ 使用CTE递归查询
WITH RECURSIVE comment_tree AS (
    -- 查询顶级评论
    SELECT * FROM comments
    WHERE article_id = 123 AND parent_id = 0

    UNION ALL

    -- 递归查询子评论
    SELECT c.* FROM comments c
    JOIN comment_tree ct ON c.parent_id = ct.id
)
SELECT * FROM comment_tree ORDER BY create_time;

方案2：闭包表（大量数据）

-- 闭包表存储所有祖先-后代关系
CREATE TABLE comment_closure (
    ancestor_id BIGINT NOT NULL,
    descendant_id BIGINT NOT NULL,
    depth INT NOT NULL,
    PRIMARY KEY (ancestor_id, descendant_id)
);

-- 查询某个评论的所有子孙
SELECT c.*
FROM comments c
JOIN comment_closure cc ON c.id = cc.descendant_id
WHERE cc.ancestor_id = 123
ORDER BY c.create_time;

-- 查询某个评论的所有祖先
SELECT c.*
FROM comments c
JOIN comment_closure cc ON c.id = cc.ancestor_id
WHERE cc.descendant_id = 456
ORDER BY cc.depth DESC;

方案3：物化路径（推荐）

-- 添加path字段存储路径
ALTER TABLE comments ADD COLUMN path VARCHAR(500);

-- 插入数据
-- 顶级评论：path = '1'
-- 二级评论：path = '1.2'
-- 三级评论：path = '1.2.3'

-- 查询某个评论的所有子孙
SELECT * FROM comments
WHERE article_id = 123
  AND path LIKE '1.%'  -- 查询评论1的所有子孙
ORDER BY path;

-- 查询某个评论的所有祖先
SELECT * FROM comments
WHERE article_id = 123
  AND FIND_IN_SET('1', REPLACE(path, '.', ',')) > 0
ORDER BY path;

缓存优化：

@Service
public class CommentService {
    @Cacheable(value = "comment:tree", key = "#articleId")
    public List<Comment> getCommentTree(Long articleId) {
        // 查询评论树
        return queryCommentTree(articleId);
    }

    @CacheEvict(value = "comment:tree", key = "#comment.articleId")
    public Comment createComment(Comment comment) {
        // 创建评论
        Comment saved = commentRepository.save(comment);

        // 更新path
        updateCommentPath(saved);

        return saved;
    }
}

面试回答：

"评论系统设计考虑树形结构和并发性能。表设计存储parent_id和path，path使用物化路径方案存储'1.2.3'格式，便于查询所有子孙和祖先。查询时使用path LIKE '1.%'快速获取子树。使用Redis缓存整个评论树，增删改时失效缓存。对于超高并发场景，使用ES存储评论，支持全文搜索和复杂查询。"

---

案例3：点赞系统设计

需求：

• 用户可以点赞/取消点赞
• 实时获取点赞数
• 查询用户是否点赞

存储方案：

Redis方案（推荐）：

@Service
public class LikeService {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    private static final String LIKE_COUNT_KEY = "like:count:";
    private static final String USER_LIKED_KEY = "user:liked:";

    // 点赞
    public boolean like(Long userId, Long targetId) {
        String userLikedKey = USER_LIKED_KEY + userId + ":" + targetId;

        // 检查是否已点赞
        Boolean isLiked = redisTemplate.opsForValue().setIfAbsent(userLikedKey, "1", 30, TimeUnit.DAYS);

        if (Boolean.TRUE.equals(isLiked)) {
            // 增加点赞数
            redisTemplate.opsForValue().increment(LIKE_COUNT_KEY + targetId);
            return true;
        }
        return false;
    }

    // 取消点赞
    public boolean unlike(Long userId, Long targetId) {
        String userLikedKey = USER_LIKED_KEY + userId + ":" + targetId;

        // 检查是否点赞
        String value = (String) redisTemplate.opsForValue().get(userLikedKey);
        if (value != null) {
            // 删除点赞记录
            redisTemplate.delete(userLikedKey);
            // 减少点赞数
            redisTemplate.opsForValue().decrement(LIKE_COUNT_KEY + targetId);
            return true;
        }
        return false;
    }

    // 获取点赞数
    public Long getLikeCount(Long targetId) {
        Long count = (Long) redisTemplate.opsForValue().get(LIKE_COUNT_KEY + targetId);
        return count != null ? count : 0L;
    }

    // 检查用户是否点赞
    public boolean isLiked(Long userId, Long targetId) {
        String userLikedKey = USER_LIKED_KEY + userId + ":" + targetId;
        return redisTemplate.hasKey(userLikedKey);
    }
}

持久化方案：

// 定时将Redis数据同步到MySQL
@Scheduled(cron = "0 0 2 * * ?")  // 每天凌晨2点
public void syncToMySQL() {
    // 1. 查询Redis所有点赞数据
    Set<String> keys = redisTemplate.keys("user:liked:*");

    // 2. 批量处理
    for (String key : keys) {
        // 解析key
        String[] parts = key.split(":");
        Long userId = Long.parseLong(parts[2]);
        Long targetId = Long.parseLong(parts[3]);

        // 更新MySQL
        likeRepository.upsert(userId, targetId);
    }

    // 3. 删除已处理的数据（或设置标志）
}

MySQL表设计：

CREATE TABLE likes (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    user_id BIGINT NOT NULL,
    target_id BIGINT NOT NULL,
    target_type VARCHAR(20) NOT NULL,  -- POST, COMMENT, USER
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_user_target (user_id, target_id, target_type),
    INDEX idx_target (target_id, target_type)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

BitMap方案（用户维度）：

// 使用BitMap记录用户是否点赞某篇文章
// 每个bit代表一个用户，1表示点赞，0表示未点赞

public void setLikeBit(Long articleId, Long userId) {
    long bitIndex = userId % (Long.MAX_VALUE - 1);
    redisTemplate.opsForValue().setBit("article:likes:" + articleId, bitIndex, true);
}

public boolean isLiked(Long articleId, Long userId) {
    long bitIndex = userId % (Long.MAX_VALUE - 1);
    return redisTemplate.opsForValue().getBit("article:likes:" + articleId, bitIndex);
}

public long getLikeCount(Long articleId) {
    // 统计bit数
    return redisTemplate.execute((RedisCallback<Long>) connection -> {
        return (long) connection.bitCount("article:likes:" + articleId.getBytes());
    });
}

面试回答：

"点赞系统采用Redis+MySQL双层存储。Redis使用String存储点赞数，使用user:liked:userId:targetId记录点赞关系，操作简单快速。定时任务将数据同步到MySQL持久化。对于用户维度的点赞状态，可以使用BitMap进一步优化，一个bit代表一个用户，内存占用极小。实际项目中我们支持Post、Comment、User等多种点赞类型，使用target_type区分。"

---

案例4：排行榜系统设计

需求：

• 实时排行榜
• 支持分页查询
• 支持按时间范围查询

方案设计：

方案1：Redis ZSet（推荐）

@Service
public class LeaderboardService {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    private static final String LEADERBOARD_KEY = "leaderboard:";

    // 增加分数
    public void addScore(Long userId, double score) {
        redisTemplate.opsForZSet().incrementScore(
            LEADERBOARD_KEY + "all",
            userId,
            score
        );
    }

    // 获取Top N
    public List<RankItem> getTopN(int n) {
        Set<ZSetOperations.TypedTuple<Object>> set =
            redisTemplate.opsForZSet().reverseRangeWithScores(
                LEADERBOARD_KEY + "all", 0, n - 1
            );

        return set.stream()
            .map(tuple -> RankItem.builder()
                .userId((Long) tuple.getValue())
                .score(tuple.getScore())
                .build())
            .collect(Collectors.toList());
    }

    // 获取用户排名
    public Long getUserRank(Long userId) {
        Long rank = redisTemplate.opsForZSet().reverseRank(
            LEADERBOARD_KEY + "all", userId
        );
        return rank != null ? rank + 1 : null;
    }

    // 分页查询
    public List<RankItem> getRange(int page, int size) {
        long start = (page - 1) * size;
        long end = start + size - 1;

        Set<ZSetOperations.TypedTuple<Object>> set =
            redisTemplate.opsForZSet().reverseRangeWithScores(
                LEADERBOARD_KEY + "all", start, end
            );

        return set.stream()
            .map(tuple -> RankItem.builder()
                .userId((Long) tuple.getValue())
                .score(tuple.getScore())
                .rank(start++)
                .build())
            .collect(Collectors.toList());
    }
}

定时排行榜（周榜、月榜）：

@Service
public class LeaderboardScheduler {
    @Autowired
    private LeaderboardService leaderboardService;

    // 每天创建新key
    @Scheduled(cron = "0 0 0 * * ?")
    public void createDailyLeaderboard() {
        String date = LocalDate.now().format(DateTimeFormatter.ISO_DATE);
        // 当天的key
        String dailyKey = "leaderboard:daily:" + date;
    }

    // 每周汇总周榜
    @Scheduled(cron = "0 0 0 ? * MON")
    public void createWeeklyLeaderboard() {
        LocalDate today = LocalDate.now();
        LocalDate weekStart = today.minusDays(6);

        // 合并一周的数据
        String weeklyKey = "leaderboard:weekly:" + today;
        for (int i = 0; i < 7; i++) {
            LocalDate date = weekStart.plusDays(i);
            String dailyKey = "leaderboard:daily:" + date;
            // 合并到周榜
            mergeToWeekly(dailyKey, weeklyKey);
        }
    }
}

持久化到MySQL：

CREATE TABLE leaderboard (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    user_id BIGINT NOT NULL,
    score DECIMAL(10, 2) NOT NULL,
    rank INT NOT NULL,
    date DATE NOT NULL,
    UNIQUE KEY uk_user_date (user_id, date),
    INDEX idx_date_rank (date, rank)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

@Scheduled(cron = "0 0 1 * * ?")  // 每天凌晨1点
public void syncLeaderboard() {
    // 1. 获取Redis排行榜
    List<RankItem> ranks = leaderboardService.getTopN(1000);

    // 2. 转换并插入MySQL
    LocalDate today = LocalDate.now();
    for (int i = 0; i < ranks.size(); i++) {
        LeaderboardEntity entity = LeaderboardEntity.builder()
            .userId(ranks.get(i).getUserId())
            .score(ranks.get(i).getScore())
            .rank(i + 1)
            .date(today)
            .build();
        leaderboardRepository.save(entity);
    }
}

面试回答：

"排行榜系统使用Redis ZSet实现，支持实时分数更新、排名查询、分页获取。对于多时间维度的排行榜（日榜、周榜、月榜），使用定时任务创建不同key分别存储。定时任务将排行榜数据持久化到MySQL，用于历史查询和数据统计。ZSet的时间复杂度是O(log N)，支持千万级数据的高性能查询。对于更复杂的场景，可以结合MySQL做二次查询。"

---

第八部分：面试题库（逐题分析）

面试题1：如何设计一个高并发系统？

回答模板（完整版）：

"设计高并发系统从以下几个层面考虑：

1. 架构层面：

• 使用微服务拆分，按业务领域拆分服务
• 读写分离，减轻主库压力
• 分库分表，水平扩展存储能力
• CDN加速静态资源

2. 缓存层面：

• 多级缓存：本地缓存+分布式缓存
• 缓存预热，提前加载热点数据
• 处理缓存穿透、击穿、雪崩

3. 异步处理：

• 使用消息队列异步处理耗时操作
• CompletableFuture实现本地异步
• 事件驱动架构解耦

4. 数据库优化：

• 读写分离，主库写从库读
• 分库分表，水平扩展
• 索引优化，合理使用索引
• 连接池调优，HikariCP

5. 保护机制：

• 限流：Sentinel QPS限流
• 降级：非核心功能降级
• 熔断：防止雪崩

6. 监控告警：

• 监控系统指标：QPS、RT、错误率
• 监控资源指标：CPU、内存、磁盘、网络
• 设置告警阈值，及时发现问题

通过多维度优化，系统可支撑10万+ QPS。"

面试题2：如何处理热点数据？

回答模板：

"热点数据处理策略：

1. 识别热点：

• 监控各key的访问频率
• 采样分析找出热点key

2. 缓存热点：

• 本地缓存热点数据，减少Redis压力
• 设置更长的过期时间

3. 多级缓存：

• 应用层本地缓存
• 分布式缓存
• 数据库

4. 动态限流：

• 热点key单独限流
• 非热点key正常限流

5. 数据迁移：

• 热点数据可以单独存储
• 使用更高性能的存储

实际项目中，我们通过监控识别热点key，动态调整缓存策略和限流规则。"

面试题3：消息队列如何保证顺序性？

回答模板：

"消息顺序性保证：

1. 全局有序：

• 使用单分区Topic，所有消息按顺序写入
• 单消费者消费

2. 分区有序：

• 相同key的消息发到同一分区
• Kafka通过partitionKey保证
• 每个分区内部有序

3. 消费端保证：

• 单线程消费保证顺序
• 多线程消费需要额外设计

实际项目中，大多数场景只需要分区有序，通过设计合理的分区键实现。全局有序会牺牲吞吐量，谨慎使用。"

面试题4：分布式锁怎么实现？

回答模板：

"分布式锁实现方案：

方案1：Redis SETNX + 过期时间（常用）

boolean locked = redisTemplate.opsForValue()
    .setIfAbsent(lockKey, value, timeout, TimeUnit.SECONDS);

方案2：Redisson（推荐，自动续约）

RLock lock = redissonClient.getLock(lockKey);
lock.lock();  // 自动续约（看门狗）
try {
    // 业务逻辑
} finally {
    lock.unlock();
}

方案3：Zookeeper临时节点

• 创建临时有序节点
• 最小节点获取锁

Redis方案性能好，Zookeeper方案可靠性高。实际项目中我们使用Redisson，它实现了Redis分布式锁，支持可重入、自动续约、公平锁。"

面试题5：如何保证消息不丢失？

回答模板：

"消息不丢失从三个环节保证：

1. 生产端：

• 开启confirm机制，等待Broker确认
• 发送失败重试，最多3次
• 持久化到本地，发送成功后删除

2. 服务端：

• 消息持久化到磁盘
• 集群部署，多副本
• 同步复制副本再返回确认

3. 消费端：

• 手动ACK确认
• 消费失败不ACK，重新投递
• 业务逻辑在事务内执行

通过全链路保障，可以实现消息不丢失。对于金融等关键业务，可以配合数据库事务，实现最终一致性。"

面试题6：如何设计一个短链接系统？

回答模板：

"短链接系统设计：

1. 短链接生成算法：

• 自增ID转62进制：a-z, A-Z, 0-9
• 哈希+冲突检测
• 雪花算法

2. 存储设计：

CREATE TABLE short_urls (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    short_code VARCHAR(10) UNIQUE,
    long_url VARCHAR(2048),
    create_time DATETIME,
    expire_time DATETIME
);

3. 缓存优化：

• Redis缓存长短映射
• LRU淘汰冷数据

4. 防刷机制：

• 限流防止单用户刷
• BloomFilter过滤已访问

5. 分布式ID：

• 使用雪花算法生成唯一ID

实际项目中，我们使用Redis自增ID+62进制生成短链接，Redis缓存热点映射，MySQL持久化存储。通过限流和BloomFilter防止恶意访问。"

面试题7：如何设计一个限流系统？

回答模板：

"限流系统设计：

1. 限流维度：

• 全局限流：保护系统整体
• 用户限流：防止单用户刷
• IP限流：防止恶意攻击

2. 限流算法：

• 固定窗口：简单但边界问题
• 滑动窗口：精确但内存占用大
• 令牌桶：适合流量整形
• 漏桶：适合恒定输出

3. 实现方案：

• Sentinel：功能全面，支持多种规则
• Guava RateLimiter：令牌桶实现
• Redis+Lua：分布式限流

4. 配置管理：

• 通过配置中心动态调整
• 按接口配置不同规则

实际项目中我们使用Sentinel，支持QPS限流、并发限流、Warm Up预热，通过Nacos动态配置规则。"

面试题8：如何优化慢查询？

回答模板：

"慢查询优化步骤：

1. 识别慢查询：

• 开启慢查询日志
• 使用explain分析执行计划

2. 优化索引：

• 为WHERE、JOIN、ORDER BY列建索引
• 使用覆盖索引减少回表
• 遵循最左前缀原则

3. 优化SQL：

• 避免SELECT *，只查需要的列
• 避免子查询，改用JOIN
• 分页查询优化，使用游标

4. 架构优化：

• 读写分离
• 分库分表
• 使用缓存

5. 代码优化：

• 批量操作减少数据库调用
• 使用连接池复用连接

通过多维度优化，慢查询从秒级降低到毫秒级。"

面试题9：如何实现分布式事务？

回答模板：

"分布式事务实现方案：

方案1：2PC/XA（强一致性）

• 两阶段提交：准备阶段+提交阶段
• 缺点：性能差，单点故障

方案2：TCC（补偿机制）

• Try：预留资源
• Confirm：确认提交
• Cancel：取消补偿
• 适用：业务逻辑可控的场景

方案3：Saga（长事务）

• 每步都有补偿操作
• 失败时反向补偿
• 适用：长流程业务

方案4：本地消息表（最终一致性）

• 业务+消息在同一事务
• 定时任务投递消息
• 适用：对实时性要求不高的场景

实际项目中，我们根据业务场景选择：强一致性用TCC，最终一致性用本地消息表。"

面试题10：如何设计一个即时通讯系统？

回答模板：

"即时通讯系统设计：

1. 消息推送：

• 长连接：WebSocket
• 推送服务：Netty
• 心跳保活

2. 消息存储：

• 聊天记录：MySQL持久化
• 未读消息：Redis计数
• 消息索引：ES搜索

3. 用户在线状态：

• Redis Set存储在线用户
• 过期时间+心跳更新

4. 离线消息：

• 离线时消息存储到Redis
• 上线后推送

5. 群聊优化：

• 使用消息队列异步处理
• 批量推送

实际项目中，我们使用WebSocket建立长连接，Netty作为服务端，消息存储MySQL，未读状态Redis，离线消息MQ异步推送。通过分片策略支持千万级在线用户。"

---

第九部分：快速记忆卡片

核心概念速记

【高并发三大瓶颈】
CPU瓶颈、I/O瓶颈、内存瓶颈

【缓存三剑客】
缓存穿透（布隆过滤器）、缓存击穿（互斥锁）、缓存雪崩（随机过期）

【缓存更新策略】
Cache Aside（最常用）、Write Through、Write Behind

【限流算法】
固定窗口、滑动窗口、令牌桶、漏桶

【系统保护三剑客】
限流、降级、熔断

【QPS计算】
QPS = 并发数 / 平均响应时间

技术选型决策树

需要缓存？
├─ 单机 → Caffeine/Guava
└─ 分布式 → Redis

需要消息队列？
├─ 高吞吐 → Kafka
├─ 功能丰富 → RabbitMQ
└─ 阿里系 → RocketMQ

需要限流？
├─ Java应用 → Sentinel
├─ 网关层 → Nginx
└─ 基础设施 → Redis

需要分布式锁？
├─ 性能优先 → Redis
└─ 可靠性优先 → Zookeeper

---

第十部分：学习路线图

阶段1：基础理解（1周）

• [ ] 理解并发核心概念（QPS、TPS、并发数）
• [ ] 理解三大瓶颈（CPU、I/O、内存）
• [ ] 掌握性能测试方法
• [ ] 理解多线程基础

阶段2：缓存技术（2周）

• [ ] 本地缓存（Caffeine、Guava）
• [ ] 分布式缓存（Redis）
• [ ] 缓存穿透、击穿、雪崩处理
• [ ] 缓存更新策略

阶段3：消息队列（1周）

• [ ] RabbitMQ基础
• [ ] Kafka基础
• [ ] 消息可靠性保证
• [ ] 消息堆积处理

阶段4：数据库优化（1周）

• [ ] 读写分离
• [ ] 分库分表
• [ ] 索引优化
• [ ] 连接池优化

阶段5：系统保护（1周）

• [ ] 限流（Sentinel）
• [ ] 降级
• [ ] 熔断（Resilience4j）
• [ ] 异步处理

阶段6：实战项目（2周）

• [ ] 秒杀系统设计
• [ ] 评论系统设计
• [ ] 点赞系统设计
• [ ] 排行榜系统设计

阶段7：面试准备（1周）

• [ ] 背诵10个核心面试题回答
• [ ] 练习设计3个完整方案
• [ ] 模拟面试

---

结语

记住几个关键点：

1. 核心思想： 异步化、缓存、解耦
2. 架构原则： 分层、拆分、冗余
3. 优化思路： 找瓶颈、定策略、监控效果
4. 保护机制： 限流、降级、熔断

面试时，先说架构图，再说具体实现，最后说优化效果。你不需要做过10万QPS的系统，只要理解设计思路和原理，就能给出令人信服的答案！

最重要的是：自信！ 高并发系统设计的核心是理解瓶颈并合理运用各种技术解决，这不是神秘的东西，而是工程经验的总结。