双11秒杀MySQL连接池爆满崩溃后 高并发优化实战指南
那个让人心碎的凌晨三点
我记得很清楚，那是2022年的双11凌晨，我的手机在凌晨两点被连续叫醒。公司的主控大屏上，MySQL的CPU使用率像疯了一样从30%飙到100%，连接数从正常的500条瞬间拉到数据库最大限制的2000条上限，然后——整个数据库拒绝新连接了。
秒杀系统彻底瘫痪。
订单中心告警邮件像雪片一样飞来，运营群里一片死寂，老板的电话紧接着打过来：”什么情况？！双11啊！”
那一刻我真正理解了什么叫”连接池爆满”。不是课本上的概念，是每一秒钟几万笔订单在堆积，数据库的TCP连接被撑爆，应用服务器拿不到连接，请求全部超时，整个链路雪崩。
那晚我们花了整整三天时间才完全恢复，双11当天的GMV直接腰斩。
事后复盘的时候，我把整个优化过程整理成了这篇指南。不是为了炫耀技术，而是真的希望后来的人别再踩同样的坑。
一、先搞清楚：连接池为什么会爆？
在讲优化之前，得先让你明白问题到底出在哪里。不然你修起来也是盲人摸象。
连接池的本质
想象你在餐厅当厨师，连接池就是你的锅。MySQL的数据库连接就是一次炒菜的工作能力。
应用服务器（餐厅厨房）
↓
连接池（锅的数量有限）
↓
MySQL数据库（灶台）
连接池的作用很简单：避免每次请求都去跟数据库重新建立TCP连接（太慢了，三次握手加认证至少几十毫秒），而是预先创建一批连接放着，需要的时候拿来用，用完放回去。
问题出在哪？
很多团队在搭建秒杀系统的时候，连接池配置是这样的：
// 典型的错误配置，见过太多这样的了
@Configuration
public class DataSourceConfig {
@Bean
public DruidDataSource dataSource() {
DruidDataSource dataSource = new DruidDataSource();
dataSource.setUrl("jdbc:mysql://master-db:3306/seckill?useSSL=false");
dataSource.setUsername("root");
dataSource.setPassword("123456");
// 最大连接数设置得非常大——这是最大的误区！
dataSource.setMaxPoolSize(500); // 以为连接越多越好？错！
// 最小连接数也开得很大
dataSource.setMinIdle(200);
// 连接超时时间设置得非常长
dataSource.setMaxWait(30000); // 等30秒？秒杀场景下300毫秒都嫌长！
return dataSource;
}
}
你觉得这样配置很合理？最大500个连接，够用了吧？
不合理的地方有三处：
第一，连接数不是越大越好。
数据库本身有资源限制。每一个连接都要占用内存（每个连接大约1-2MB）、CPU上下文切换开销、TCP连接维护成本。500个连接同时跑，数据库本身的CPU就吃不消了。而且，操作系统层面也有文件描述符限制（ulimit -n），连接数太多会直接报错”Too many open files”。
第二，秒杀场景的流量特点是”瞬间爆发”。
平时每分钟几千请求，双11零点那一秒能到几十万。连接池的初始连接需要时间预热，等流量来了再慢慢建连接，黄花菜都凉了。
第三，最致命的是——连接泄漏。
这是我在复盘中发现的最隐蔽的坑。看这段代码：
// 典型的连接泄漏写法
public Order createOrder(SeckillOrderRequest request) {
Connection conn = dataSource.getConnection(); // 获取连接
try {
// 业务逻辑...
PreparedStatement ps = conn.prepareStatement(
"INSERT INTO seckill_order VALUES (?, ?, ?)"
);
ps.setInt(1, request.getUserId());
ps.setString(2, request.getProductId());
ps.setLong(3, System.currentTimeMillis());
ps.executeUpdate();
// 然后就完了——没有close！
// 如果这里抛异常，连接就永远回不去了！
} catch (SQLException e) {
log.error("创建订单失败", e);
throw new RuntimeException("订单创建失败", e);
// 异常情况下，连接泄漏！
}
// 注意：finally块里居然没有conn.close()！
}
这段代码有3个致命问题：
- 没有
finally块来确保连接归还 - 即使有
finally，close()调用也没有放在try-catch里 - 异常处理逻辑不完整
我翻了我们当时的代码仓库，类似的问题在整个订单模块里到处都是。连接一旦泄漏，就被连接池判定为”使用中”，永远不会归还。连接池慢慢被泄漏的连接占满，新请求拿不到连接，要么阻塞等待（你设置了30秒超时），要么直接超时返回。最终整个连接池被撑爆。
二、连接池调优：从血泪中总结的六个关键参数
基于那次惨痛的教训，我重新设计了整个连接池配置。下面是优化后的版本，每个参数都有明确的理由。
正确的连接池配置
@Configuration
public class DataSourceConfig {
@Bean("seckillDataSource")
@Primary
public DruidDataSource seckillDataSource() {
DruidDataSource dataSource = new DruidDataSource();
// ==================== 核心连接参数 ====================
// 初始连接数：预热到合理水平，避免冷启动时连接创建延迟
// 建议设为 maxPoolSize 的 30%-50%
dataSource.setInitialSize(50);
// 最小空闲连接：保证基础流量下有足够连接可用
dataSource.setMinIdle(30);
// 最大活跃连接：根据数据库服务器实际承受能力设定
// 单节点MySQL在SSD上，建议不超过 200-300
// 连接数 = QPS × 平均响应时间(秒) × 安全系数(1.5)
// 例如：QPS=2000, 响应时间=0.05s → 2000×0.05×1.5=150
dataSource.setMaxActive(150);
// ==================== 超时控制 ====================
// 获取连接的最大等待时间：秒杀场景建议 2-5秒
// 超过这个时间直接失败，避免线程无限等待
dataSource.setMaxWait(3000);
// 连接泄漏检测：超过这个时间的连接会被强制回收
// 这是一个安全网，防止代码遗漏close()导致连接耗尽
dataSource.setRemoveAbandoned(true);
dataSource.setRemoveAbandonedTimeout(60);
dataSource.setLogAbandoned(true); // 记录泄漏日志，方便排查
// ==================== 健康检查 ====================
// 空闲连接存活最长时间：超过这个时间的空闲连接会被回收
dataSource.setMaxIdle(600000); // 10分钟
// 连接有效期：超过这个时间的连接会被强制关闭
// 防止长连接被网络中间设备断开后应用层不知道
dataSource.setMaxLifetime(1800000); // 30分钟
// ==================== 防抖动 ====================
// 空闲连接检测间隔：定期清理无效连接
dataSource.setTestWhileIdle(true);
dataSource.setTestOnBorrow(false); // 借出时不检测（影响性能）
dataSource.setTestOnReturn(false); // 归还时不检测（影响性能）
// 空闲连接检测SQL
dataSource.setValidationQuery("SELECT 1");
dataSource.setTimeBetweenEvictionRunsMillis(30000); // 每30秒检测一次
// ==================== 监控配置 ====================
// 开启Druid监控
dataSource.setUseGlobalDataSourceStat(true);
dataSource.setRemoveAbandoned(true);
return dataSource;
}
}
参数详解
| 参数 | 设置值 | 作用 | 设置依据 |
|---|---|---|---|
initialSize |
50 | 启动时创建50个连接，避免冷启动压力 | 根据峰值QPS预估的并发连接数 |
minIdle |
30 | 保持至少30个空闲连接 | 保证突发流量时有足够缓冲 |
maxActive |
150 | 最大活跃连接数 | QPS × 平均响应时间 × 1.5 |
maxWait |
3000ms | 等待连接的最长时间 | 秒杀场景要求快速失败 |
maxLifetime |
30分钟 | 连接最大存活时间 | 防止长连接被网络中断 |
removeAbandonedTimeout |
60秒 | 泄漏检测超时 | 正常请求一般在1秒内完成 |
关键公式：如何科学计算maxActive
这是我在优化过程中学到的最重要的公式：
maxActive = 峰值QPS × 平均数据库响应时间(秒) × 安全系数
举个例子：
假设秒杀峰值QPS = 10000
假设数据库平均响应时间 = 50ms = 0.05s
安全系数 = 1.5（考虑网络抖动和突发流量）
maxActive = 10000 × 0.05 × 1.5 = 750
但是！单个MySQL实例通常不建议超过300个连接
（因为连接本身的CPU和内存开销）
所以这里应该做分库，而不是无限增大连接池
如果计算出来的连接数超过单库承受能力，不要硬撑，直接分库。
三、读写分离：把压力从主库上分流
连接池调优只是治标，读写分离才是治本的第一步。
为什么需要读写分离？
在秒杀系统中，读操作和写操作的比例通常是 10:1 甚至 100:1。
什么意思？每1个写请求（创建订单），会有100个读请求（查询库存、查询商品信息、查询活动规则）。
如果你让所有的读请求都打到主库上，主库的压力会非常大。而主库的主要任务是处理写操作，保证数据的一致性。
读写分离的思路很简单：
- 主库（Master）：处理所有写操作，保证数据一致性
- 从库（Slave）：处理所有读操作，可以部署多个从库来分散压力
架构设计
┌─────────────────────────────────────┐
│ 应用层（API网关） │
│ 读请求 → 路由到从库 │
│ 写请求 → 路由到主库 │
└──────────────┬──────────────────────┘
│
┌────────────────────┼────────────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ 主库 │ │ 从库1 │ │ 从库2 │
│ (写) │ │ (读) │ │ (读) │
└─────┬────┘ └──────────┘ └──────────┘
│
MySQL主从复制（binlog）
│
▼
┌──────────┐
│ 从库3 │
│ (读) │
└──────────┘
Spring Boot 实现读写分离
/**
* 动态数据源路由配置
* 核心思想：根据当前线程的"读写标记"来决定走哪个数据源
*/
@Configuration
public class DynamicDataSourceConfig {
@Bean
@ConfigurationProperties("spring.datasource.master")
public DataSource masterDataSource() {
return DataSourceBuilder.create().build();
}
@Bean
@ConfigurationProperties("spring.datasource.slave1")
public DataSource slave1DataSource() {
return DataSourceBuilder.create().build();
}
@Bean
@ConfigurationProperties("spring.datasource.slave2")
public DataSource slave2DataSource() {
return DataSourceBuilder.create().build();
}
/**
* 动态数据源：根据上下文决定路由到哪个数据源
*/
@Bean
public DataSource dynamicDataSource(
@Qualifier("masterDataSource") DataSource master,
@Qualifier("slave1DataSource") DataSource slave1,
@Qualifier("slave2DataSource") DataSource slave2) {
Map<Object, Object> targetDataSources = new HashMap<>();
targetDataSources.put(DBTypeEnum.MASTER, master);
targetDataSources.put(DBTypeEnum.SLAVE_1, slave1);
targetDataSources.put(DBTypeEnum.SLAVE_2, slave2);
RoutingDataSource routingDataSource = new RoutingDataSource();
routingDataSource.setTargetDataSources(targetDataSources);
routingDataSource.setDefaultTargetDataSource(master); // 默认走主库
return routingDataSource;
}
/**
* 数据源路由策略：基于ThreadLocal的读写标记
*/
public class RoutingDataSource extends AbstractRoutingDataSource {
@Override
protected Object determineCurrentLookupKey() {
// 从ThreadLocal中获取当前线程的数据源类型
return DataSourceContextHolder.getDataSourceType();
}
}
}
/**
* 数据源类型枚举
*/
public enum DBTypeEnum {
MASTER, SLAVE_1, SLAVE_2
}
/**
* 数据源上下文：基于ThreadLocal实现线程隔离
*/
public class DataSourceContextHolder {
private static final ThreadLocal<DBTypeEnum> CONTEXT = new ThreadLocal<>();
public static void setDataSourceType(DBTypeEnum type) {
CONTEXT.set(type);
}
public static DBTypeEnum getDataSourceType() {
DBTypeEnum type = CONTEXT.get();
return type != null ? type : DBTypeEnum.MASTER;
}
public static void clear() {
CONTEXT.remove();
}
}
/**
* AOP切面：自动根据方法注解路由到对应数据源
*/
@Aspect
@Component
public class DataSourceAop {
/**
* 标记读操作：路由到从库
*/
@Around("@annotation(com.example.annotation.Read)")
public Object aroundRead(ProceedingJoinPoint point) throws Throwable {
try {
DataSourceContextHolder.setDataSourceType(
selectSlave() // 负载均衡选择从库
);
return point.proceed();
} finally {
DataSourceContextHolder.clear();
}
}
/**
* 标记写操作：路由到主库
*/
@Around("@annotation(com.example.annotation.Write)")
public Object aroundWrite(ProceedingJoinPoint point) throws Throwable {
try {
DataSourceContextHolder.setDataSourceType(DBTypeEnum.MASTER);
return point.proceed();
} finally {
DataSourceContextHolder.clear();
}
}
/**
* 默认逻辑：Service层方法无注解时，根据方法名判断
*/
@Around("execution(* com.example.service..*.*(..))")
public Object aroundService(ProceedingJoinPoint point) throws Throwable {
String methodName = point.getSignature().getName();
// 根据方法名前缀判断读写
if (methodName.startsWith("query")
|| methodName.startsWith("get")
|| methodName.startsWith("list")
|| methodName.startsWith("count")) {
DataSourceContextHolder.setDataSourceType(selectSlave());
} else {
DataSourceContextHolder.setDataSourceType(DBTypeEnum.MASTER);
}
try {
return point.proceed();
} finally {
DataSourceContextHolder.clear();
}
}
private DBTypeEnum selectSlave() {
// 简单的轮询负载均衡
int index = Integer.parseInt(
System.currentTimeMillis() % 2 + ""
);
return index == 0 ? DBTypeEnum.SLAVE_1 : DBTypeEnum.SLAVE_2;
}
}
使用示例
@Service
public class SeckillService {
@Resource
private SeckillOrderMapper orderMapper;
/**
* 查询库存 — 自动路由到从库
*/
public int queryStock(Long productId) {
return orderMapper.queryStock(productId);
}
/**
* 创建订单 — 强制路由到主库
*/
@Write
public OrderResult createOrder(SeckillOrderRequest request) {
// 扣减库存（写操作，必须走主库）
int remaining = orderMapper.decreaseStock(request.getProductId(), request.getQuantity());
if (remaining < 0) {
return OrderResult.outOfStock();
}
// 创建订单
SeckillOrder order = new SeckillOrder();
order.setUserId(request.getUserId());
order.setProductId(request.getProductId());
order.setCreateTime(System.currentTimeMillis());
orderMapper.insert(order);
return OrderResult.success(order.getId());
}
}
读写分离的注意事项
读写分离不是银弹，有几个坑必须注意：
1. 主从延迟问题
MySQL的主从复制是异步的。主库写入后，从库需要一定时间（通常几毫秒到几百毫秒）才能同步到数据。
秒杀场景下，如果用户刚下完单，立刻去查询订单状态，有可能查到的是从库的旧数据——订单还没同步过来。
解决方案：
/**
* 强制读主库：适用于需要强一致性的场景
*/
@ReadFromMaster
public OrderResult queryOrderAfterCreate(Long orderId) {
// 强制走主库，确保读到最新数据
return orderMapper.selectById(orderId);
}
2. 从库的读压力
从库也不是万能的。如果读请求太多，从库也会扛不住。这时候需要考虑：
- 增加从库数量
- 对读请求做缓存（下一节讲）
- 对查询做限流
四、分库分表：当单个数据库扛不住的时候
读写分离解决了读压力，但写压力呢？如果秒杀场景下每秒有几万笔订单写入，单个MySQL的写性能还是不够用。
为什么要分库分表？
MySQL的性能瓶颈主要来自三个方面：
1. 单机CPU瓶颈
单个MySQL实例的CPU核数是有限的。并发查询和事务处理会占满CPU。
2. 单机IO瓶颈
磁盘IO是有上限的。即使使用SSD，每秒的随机写入次数也是有限的（通常几万到几十万IOPS）。
3. 单机内存瓶颈
InnoDB的缓冲池（Buffer Pool）需要容纳热数据。数据量太大，Buffer Pool装不下，就会频繁磁盘读写。
水平分表：解决单表数据量过大的问题
先看一个简单的例子。假设我们有一个订单表：
-- 原始的订单表
CREATE TABLE seckill_order (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
user_id BIGINT NOT NULL,
product_id BIGINT NOT NULL,
quantity INT NOT NULL DEFAULT 1,
status TINYINT NOT NULL DEFAULT 0,
create_time DATETIME NOT NULL,
update_time DATETIME NOT NULL,
INDEX idx_user_id (user_id),
INDEX idx_product_id (product_id),
INDEX idx_create_time (create_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
当数据量增长到千万级甚至亿级时，单表的索引会变得非常巨大，查询性能急剧下降。
分表策略：按用户ID取模
-- 分表为16张表
CREATE TABLE seckill_order_00 (LIKE seckill_order) ENGINE=InnoDB;
CREATE TABLE seckill_order_01 (LIKE seckill_order) ENGINE=InnoDB;
-- ... 到 seckill_order_15
Java层的分表逻辑：
/**
* 分表策略：基于用户ID取模
* 将同一用户的所有订单分散到不同的表中
*/
@Component
public class ShardingStrategy {
private static final int TABLE_COUNT = 16;
/**
* 根据用户ID计算表后缀
*/
public String getTableSuffix(Long userId) {
// 使用 hashCode 取模，保证分布均匀
int hash = Math.abs(userId.hashCode());
int index = hash % TABLE_COUNT;
return String.format("%02d", index);
}
/**
* 动态表名生成
*/
public String getTableName(Long userId) {
return "seckill_order_" + getTableSuffix(userId);
}
}
@Service
public class OrderShardingService {
@Resource
private ShardingStrategy shardingStrategy;
/**
* 插入订单：根据用户ID路由到对应分表
*/
public Long createOrder(SeckillOrder order) {
String tableName = shardingStrategy.getTableName(order.getUserId());
// 动态SQL：插入到正确的分表
String sql = "INSERT INTO " + tableName
+ " (user_id, product_id, quantity, status, create_time, update_time)"
+ " VALUES (#userId#, #productId#, #quantity#, #status#, #createTime#, #updateTime#)";
// 执行插入
long orderId = mybatisExecutor.insert(sql, order);
return orderId;
}
/**
* 查询订单：根据用户ID路由到对应分表
*/
public SeckillOrder queryOrder(Long userId, Long orderId) {
String tableName = shardingStrategy.getTableName(userId);
String sql = "SELECT * FROM " + tableName
+ " WHERE id = #orderId# AND user_id = #userId#";
return mybatisExecutor.queryOne(sql, userId, orderId, SeckillOrder.class);
}
}
水平分库：解决单机写瓶颈
分表只能解决单表数据量问题，但写压力还是集中在单机上。要解决写瓶颈，需要分库。
分库策略：按业务模块拆分
┌──────────────────────────────────────────────────────────────┐
│ 业务分库策略 │
├──────────────┬──────────────┬──────────────┬─────────────────┤
│ 订单库 │ 用户库 │ 商品库 │ 库存库 │
│ order_db │ user_db │ product_db │ stock_db │
├──────────────┼──────────────┼──────────────┼─────────────────┤
│ 订单主表 │ 用户信息 │ 商品信息 │ 库存信息 │
│ 订单详情 │ 用户地址 │ 商品分类 │ 库存流水 │
│ 订单支付 │ 用户积分 │ 商品SKU │ 扣减记录 │
│ 订单售后 │ 用户订单 │ 商品评价 │ 预扣库存 │
└──────────────┴──────────────┴──────────────┴─────────────────┘
每个库独立配置连接池，互不干扰：
@Configuration
public class MultiDataSourceConfig {
/**
* 订单库数据源
*/
@Bean("orderDataSource")
@ConfigurationProperties("datasource.order")
public DataSource orderDataSource() {
return DataSourceBuilder.create().build();
}
/**
* 用户库数据源
*/
@Bean("userDataSource")
@ConfigurationProperties("datasource.user")
public DataSource userDataSource() {
return DataSourceBuilder.create().build();
}
/**
* 商品库数据源
*/
@Bean("productDataSource")
@ConfigurationProperties("datasource.product")
public DataSource productDataSource() {
return DataSourceBuilder.create().build();
}
/**
* 库存库数据源
*/
@Bean("stockDataSource")
@ConfigurationProperties("datasource.stock")
public DataSource stockDataSource() {
return DataSourceBuilder.create().build();
}
/**
* 订单库的MyBatis配置
*/
@Bean
public SqlSessionFactory orderSqlSessionFactory(
@Qualifier("orderDataSource") DataSource dataSource) throws Exception {
SqlSessionFactoryBean factory = new SqlSessionFactoryBean();
factory.setDataSource(dataSource);
factory.setMapperLocations(
new PathMatchingResourcePatternResolver()
.getResources("classpath:mapper/order/*.xml")
);
return factory.getObject();
}
// 同理配置 userSqlSessionFactory、productSqlSessionFactory、stockSqlSessionFactory
}
配置文件：
datasource:
order:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://order-db-master:3306/seckill_order?useSSL=false&serverTimezone=Asia/Shanghai
username: order_user
password: ${ORDER_DB_PASSWORD}
druid:
initial-size: 20
min-idle: 10
max-active: 100
max-wait: 2000
user:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://user-db-master:3306/seckill_user?useSSL=false&serverTimezone=Asia/Shanghai
username: user_user
password: ${USER_DB_PASSWORD}
druid:
initial-size: 10
min-idle: 5
max-active: 50
max-wait: 2000
product:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://product-db-master:3306/seckill_product?useSSL=false&serverTimezone=Asia/Shanghai
username: product_user
password: ${PRODUCT_DB_PASSWORD}
druid:
initial-size: 10
min-idle: 5
max-active: 50
max-wait: 2000
stock:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://stock-db-master:3306/seckill_stock?useSSL=false&serverTimezone=Asia/Shanghai
username: stock_user
password: ${STOCK_DB_PASSWORD}
druid:
initial-size: 30
min-idle: 15
max-active: 200 # 库存库是核心，连接数可以设大一些
max-wait: 1000
五、缓存降级：最后一道防线
分库分表解决了数据库层面的压力，但还有最后一步——让请求根本不要打到数据库。这就是缓存的作用。
缓存架构设计
用户请求
│
▼
┌─────────────┐
│ Redis集群 │ ← 第一层：缓存查询结果（库存、商品信息）
│ (毫秒级) │
└──────┬──────┘
│ 缓存未命中
▼
┌─────────────┐
│ 本地缓存 │ ← 第二层：应用层缓存（Guava Cache / Caffeine）
│ (微秒级) │
└──────┬──────┘
│ 本地缓存未命中
▼
┌─────────────┐
│ MySQL │ ← 第三层：数据库（分库分表后的结果）
│ (毫秒级) │
└─────────────┘
多级缓存实现
/**
* 多级缓存管理器
* 优先级：本地缓存 > Redis > 数据库
*/
@Service
public class MultiLevelCacheService {
@Resource
private StringRedisTemplate redisTemplate;
/**
* 本地缓存：使用Caffeine，适合热点数据
* 容量10000，TTL 10秒，过期后自动淘汰
*/
private final Cache<String, String> localCache = Caffeine.newBuilder()
.maximumSize(10000)
.expireAfterWrite(10, TimeUnit.SECONDS)
.recordStats()
.build();
/**
* 获取库存：多级缓存查询
*/
public int getStock(Long productId) {
String cacheKey = "stock:" + productId;
// 第一级：本地缓存
String localValue = localCache.getIfPresent(cacheKey);
if (localValue != null) {
// 命中本地缓存，直接返回
return Integer.parseInt(localValue);
}
// 第二级：Redis缓存
String redisValue = redisTemplate.opsForValue().get(cacheKey);
if (redisValue != null) {
// 命中Redis，回写到本地缓存
localCache.put(cacheKey, redisValue);
return Integer.parseInt(redisValue);
}
// 第三级：数据库
int stock = stockMapper.queryStock(productId);
// 写入两级缓存
redisTemplate.opsForValue().set(cacheKey, String.valueOf(stock), 30, TimeUnit.SECONDS);
localCache.put(cacheKey, String.valueOf(stock));
return stock;
}
/**
* 扣减库存：更新缓存
*/
@Transactional
public OrderResult decreaseStock(Long productId, int quantity) {
String cacheKey = "stock:" + productId;
// 1. 先从缓存检查库存是否充足
String stockStr = localCache.getIfPresent(cacheKey);
if (stockStr == null) {
stockStr = redisTemplate.opsForValue().get(cacheKey);
}
if (stockStr == null || Integer.parseInt(stockStr) < quantity) {
return OrderResult.outOfStock();
}
// 2. 数据库扣减库存（使用分布式锁保证并发安全）
boolean locked = tryDistributedLock(productId);
if (!locked) {
// 获取锁失败，短暂等待后重试
try {
Thread.sleep(50);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return decreaseStock(productId, quantity);
}
try {
int affected = stockMapper.decreaseStock(productId, quantity);
if (affected == 0) {
return OrderResult.outOfStock();
}
// 3. 更新缓存
int newStock = Integer.parseInt(stockStr) - quantity;
redisTemplate.opsForValue().set(cacheKey, String.valueOf(newStock), 30, TimeUnit.SECONDS);
localCache.put(cacheKey, String.valueOf(newStock));
// 4. 创建订单
SeckillOrder order = new SeckillOrder();
order.setProductId(productId);
order.setQuantity(quantity);
order.setCreateTime(System.currentTimeMillis());
order.setStatus(0);
orderMapper.insert(order);
return OrderResult.success(order.getId());
} finally {
releaseDistributedLock(productId);
}
}
}
缓存降级策略
“降级”的意思是：当某个服务不可用时，临时关闭它的缓存，让请求直接打到数据库或者其他服务上。
/**
* 缓存降级管理器
* 支持动态开关缓存，应对突发情况
*/
@Component
public class CacheDowngradeManager {
/**
* 降级标记：key为服务名称，value为是否降级
*/
private final ConcurrentHashMap<String, Boolean> downgradeFlags = new ConcurrentHashMap<>();
/**
* 检查缓存是否可用
*/
public boolean isCacheAvailable(String serviceName) {
// 检查降级标记
Boolean isDowngraded = downgradeFlags.get(serviceName);
if (Boolean.TRUE.equals(isDowngraded)) {
return false;
}
// 检查Redis健康状态
return isRedisHealthy();
}
/**
* 执行降级：关闭指定服务的缓存
*/
public void downgrade(String serviceName) {
downgradeFlags.put(serviceName, true);
log.warn("缓存降级: {} 已关闭", serviceName);
}
/**
* 恢复缓存
*/
public void recover(String serviceName) {
downgradeFlags.remove(serviceName);
log.info("缓存恢复: {} 已恢复", serviceName);
}
/**
* 检查Redis健康状态
*/
private boolean isRedisHealthy() {
try {
return redisTemplate.getConnectionFactory()
.getConnection()
.ping();
} catch (Exception e) {
log.error("Redis健康检查失败", e);
return false;
}
}
/**
* 批量降级：当系统压力过大时，关闭非核心缓存
*/
public void batchDowngrade(String... services) {
for (String service : services) {
downgrade(service);
}
}
}
缓存的常见陷阱
陷阱1：缓存穿透
用户查询一个不存在的数据（比如不存在的商品ID），缓存和数据库都没有，每次请求都会打到数据库。
解决：对不存在的结果也缓存一个空值，设置短TTL。
/**
* 缓存空值，防止穿透
*/
public int getStockWithNullCache(Long productId) {
String cacheKey = "stock:" + productId;
// 查询缓存
String cached = redisTemplate.opsForValue().get(cacheKey);
if (cached != null) {
// 空值标记：-1 表示库存不存在
if ("-1".equals(cached)) {
return 0;
}
return Integer.parseInt(cached);
}
// 查询数据库
int stock = stockMapper.queryStock(productId);
if (stock <= 0) {
// 库存为0，缓存空值3秒
redisTemplate.opsForValue().set(cacheKey, "-1", 3, TimeUnit.SECONDS);
return 0;
}
// 正常缓存，TTL 30秒
redisTemplate.opsForValue().set(cacheKey, String.valueOf(stock), 30, TimeUnit.SECONDS);
return stock;
}
陷阱2：缓存击穿
热点key在过期瞬间，大量请求同时打到数据库。
解决：使用互斥锁，只有一个请求去查数据库，其他请求等待。
/**
* 缓存击穿防护：使用分布式锁
*/
public int getStockWithLock(Long productId) {
String cacheKey = "stock:" + productId;
// 先查缓存
String cached = redisTemplate.opsForValue().get(cacheKey);
if (cached != null) {
return Integer.parseInt(cached);
}
// 缓存未命中，尝试获取锁
String lockKey = "lock:stock:" + productId;
boolean locked = redisTemplate.opsForValue()
.setIfAbsent(lockKey, "1", 10, TimeUnit.SECONDS);
if (locked) {
try {
// 拿到锁，再查一次缓存（可能其他请求已经写入了）
cached = redisTemplate.opsForValue().get(cacheKey);
if (cached != null) {
return Integer.parseInt(cached);
}
// 查数据库
int stock = stockMapper.queryStock(productId);
// 写入缓存
redisTemplate.opsForValue().set(cacheKey, String.valueOf(stock), 30, TimeUnit.SECONDS);
return stock;
} finally {
// 释放锁
redisTemplate.delete(lockKey);
}
} else {
// 没拿到锁，短暂等待后重试
try {
Thread.sleep(50);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return getStockWithLock(productId);
}
}
陷阱3：缓存雪崩
大量缓存同时过期，所有请求打到数据库。
解决：TTL设置随机值，避免集中过期。
/**
* 随机TTL，防止缓存雪崩
*/
public void setWithRandomTTL(String key, String value) {
// 基础TTL 30秒，随机加上0-30秒
int randomTTL = 30 + new Random().nextInt(30);
redisTemplate.opsForValue().set(key, value, randomTTL, TimeUnit.SECONDS);
}
六、全链路优化：从入口到数据库的完整方案
连接池、读写分离、分库分表、缓存降级，这些单独使用都有效果，但组合起来才是真正的高并发方案。
秒杀系统的完整架构
┌─────────────────┐
│ CDN静态资源 │ ← 商品图片、JS、CSS
└────────┬────────┘
│
┌────────▼────────┐
│ 反向代理(Nginx) │ ← 限流、SSL终止
└────────┬────────┘
│
┌────────▼────────┐
│ 应用服务器集群 │ ← 无状态，水平扩展
│ (Spring Cloud) │
└───┬────────┬────┘
│ │
┌─────────────┘ └─────────────┐
▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ Redis集群 │ │ 消息队列 │
│ - 库存预扣减 │ │ - 订单异步落库 │
│ - 热点缓存 │ │ - 积分异步计算 │
│ - 分布式锁 │ │ - 通知异步发送 │
└────────┬────────┘ └────────┬────────┘
│ │
┌────────▼────────┐ ┌────────▼────────┐
│ 本地缓存 │ │ 订单库(分库) │
│ (Caffeine) │ │ order_db_0~15 │
└────────┬────────┘ └────────┬────────┘
│ │
┌────────▼────────┐ ┌────────▼────────┐
│ 商品库(分库) │ │ 用户库(分库) │
│ product_db_0~7 │ │ user_db_0~7 │
└─────────────────┘ └─────────────────┘
关键代码：秒杀核心流程
/**
* 秒杀服务：完整的高并发秒杀流程
*
* 流程：
* 1. 限流检查（Nginx层 + 应用层）
* 2. Redis预扣减库存（原子操作）
* 3. 库存不足 → 直接返回
* 4. 库存充足 → 发送消息到MQ
* 5. 立即返回"排队中"给用户
* 6. MQ消费者异步创建订单
*/
@Service
public class SeckillService {
@Resource
private StringRedisTemplate redisTemplate;
@Resource
private RabbitTemplate rabbitTemplate;
@Resource
private SeckillOrderMapper orderMapper;
/**
* 秒杀入口
*/
public SeckillResult seckill(SeckillRequest request) {
// 1. 限流检查
if (!rateLimiter.tryAcquire(request.getUserId())) {
return SeckillResult TOO_MANY_REQUESTS;
}
// 2. 参数校验
if (!validateRequest(request)) {
return SeckillResult.INVALID_PARAM;
}
// 3. 检查活动状态
SeckillActivity activity = activityCache.getActivity(request.getActivityId());
if (activity == null || activity.getStatus() != 1) {
return SeckillResult.ACTIVITY_NOT_START;
}
// 4. 检查是否已购买
if (orderMapper.existsByUserAndActivity(request.getUserId(), request.getActivityId())) {
return SeckillResult.ALREADY_PURCHASED;
}
// 5. Redis预扣减库存（原子操作）
String stockKey = "seckill:stock:" + request.getActivityId();
Long remaining = redisTemplate.opsForValue().decrement(stockKey);
if (remaining < 0) {
// 库存不足，恢复计数
redisTemplate.opsForValue().increment(stockKey);
return SeckillResult.OUT_OF_STOCK;
}
// 6. 库存充足，发送MQ消息（异步创建订单）
SeckillMessage message = new SeckillMessage();
message.setUserId(request.getUserId());
message.setActivityId(request.getActivityId());
message.setProductId(request.getProductId());
message.setQuantity(request.getQuantity());
message.setTimestamp(System.currentTimeMillis());
rabbitTemplate.convertAndSend(
"seckill.exchange",
"seckill.order",
JSON.toJSONString(message)
);
// 7. 立即返回，告知用户排队中
return SeckillResult.QUEUING;
}
/**
* MQ消费者：异步创建订单
*/
@RabbitListener(queues = "seckill.order.queue")
public void handleSeckillOrder(String messageJson) {
SeckillMessage message = JSON.parseObject(messageJson, SeckillMessage.class);
try {
// 1. 创建订单
SeckillOrder order = new SeckillOrder();
order.setUserId(message.getUserId());
order.setActivityId(message.getActivityId());
order.setProductId(message.getProductId());
order.setQuantity(message.getQuantity());
order.setStatus(0); // 待支付
order.setCreateTime(System.currentTimeMillis());
orderMapper.insert(order);
// 2. 更新本地缓存
String cacheKey = "order:" + message.getUserId() + ":" + message.getActivityId();
redisTemplate.opsForValue().set(
cacheKey,
String.valueOf(order.getId()),
24,
TimeUnit.HOURS
);
// 3. 发送用户通知（异步）
notificationService.sendNotification(
message.getUserId(),
"您的秒杀订单已创建，订单号：" + order.getId()
);
} catch (Exception e) {
log.error("创建秒杀订单失败, message: {}", messageJson, e);
// 失败补偿：恢复Redis库存
String stockKey = "seckill:stock:" + message.getActivityId();
redisTemplate.opsForValue().increment(stockKey);
// 重试机制
retryService.delayedRetry(messageJson, 3);
}
}
/**
* 限流器：基于令牌桶算法
*/
private final RateLimiter rateLimiter = RateLimiter.create(10.0); // 每秒10个请求
private boolean tryAcquire(Long userId) {
// 用户级限流：每个用户每秒最多1次
String userKey = "rate_limit:user:" + userId;
String count = redisTemplate.opsForValue().get(userKey);
if (count != null && Integer.parseInt(count) >= 1) {
return false;
}
// 全局限流
if (!rateLimiter.tryAcquire()) {
return false;
}
// 更新用户计数
if (count == null) {
redisTemplate.opsForValue().set(userKey, "1", 1, TimeUnit.SECONDS);
} else {
redisTemplate.opsForValue().increment(userKey);
}
return true;
}
}
监控与告警
没有监控的优化是盲目的。你需要知道什么时候出问题、问题在哪里。
/**
* 系统监控指标收集
*/
@Component
public class SystemMonitor {
/**
* 连接池监控：每10秒收集一次
*/
@Scheduled(fixedRate = 10000)
public void collectConnectionPoolMetrics() {
// Druid连接池监控
DataSource dataSource = dataSourceBean();
DruidDataSource druidDataSource = (DruidDataSource) dataSource;
// 收集关键指标
int activeCount = druidDataSource.getActiveCount();
int poolSize = druidDataSource.getPoolingCount();
long waitThreadCount = druidDataSource.getWaitThreadCount();
long connectCount = druidDataSource.getConnectCount();
long closeCount = druidDataSource.getCloseCount();
// 上报到监控系统（Prometheus / Grafana）
metricsCollector.gauge("db.pool.active", activeCount);
metricsCollector.gauge("db.pool.idle", poolSize);
metricsCollector.gauge("db.pool.waiting", waitThreadCount);
metricsCollector.counter("db.pool.connections.total", connectCount);
// 告警：如果等待连接的线程数超过阈值
if (waitThreadCount > 50) {
alertService.sendAlert(
"连接池等待线程数过高",
"当前等待线程数: " + waitThreadCount,
AlertLevel.WARN
);
}
// 告警：如果活跃连接数接近上限
if (activeCount > druidDataSource.getMaxActive() * 0.8) {
alertService.sendAlert(
"连接池使用率过高",
"当前活跃连接: " + activeCount + "/" + druidDataSource.getMaxActive(),
AlertLevel.ERROR
);
}
}
/**
* Redis监控
*/
@Scheduled(fixedRate = 5000)
public void collectRedisMetrics() {
RedisClusterInfo info = redisClusterMonitor.getInfo();
metricsCollector.gauge("redis.connected_clients", info.getConnectedClients());
metricsCollector.gauge("redis.used_memory", info.getUsedMemory());
metricsCollector.gauge("redis.keyspace_hits", info.getKeyspaceHits());
metricsCollector.gauge("redis.keyspace_misses", info.getKeyspaceMisses());
// 计算命中率
double hitRate = (double) info.getKeyspaceHits() /
(info.getKeyspaceHits() + info.getKeyspaceMisses());
metricsCollector.gauge("redis.hit_rate", hitRate);
// 告警：命中率过低
if (hitRate < 0.5) {
alertService.sendAlert(
"Redis命中率过低",
"当前命中率: " + String.format("%.2f%%", hitRate * 100),
AlertLevel.WARN
);
}
}
}
七、那次崩溃后，我们的完整复盘
回到最初的问题：双11凌晨连接池为什么爆？
复盘下来，根本原因不是”连接数设置太大”，而是三个问题叠加：
第一，没有缓存层。
所有的请求都直接打到数据库，连接数随着QPS线性增长。峰值时连接数直接打到2000的上限。
第二，没有读写分离。
读请求和写请求混在一起，主库既要处理写操作，又要处理大量的读操作，CPU和IO双重压力。
第三，连接泄漏。
代码里到处都是没有正确关闭连接的代码，连接池里的连接被慢慢耗尽，新请求拿不到连接。
修复方案：
修复前：
用户 → API → 数据库（直接打，无缓存，无分库）
连接池：maxActive=500，泄漏严重
修复后：
用户 → Nginx限流 → API → 本地缓存 → Redis → MySQL（读写分离 + 分库分表）
连接池：maxActive=150，泄漏检测开启，多级缓存
优化后的效果对比
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| 最大QPS | 3,000 | 50,000 | 16倍 |
| 平均响应时间 | 200ms | 15ms | 13倍 |
| 连接池最大使用 | 1,800⁄2000 (90%) | 120⁄150 (80%) | 稳定在安全线 |
| P99延迟 | 2秒 | 50ms | 40倍 |
| 数据库CPU峰值 | 95% | 40% | 大幅下降 |
八、给后来者的建议
如果你正在搭建高并发系统，或者正在经历类似的困境，记住这几点：
1. 连接池不是越大越好。
连接池的配置要基于实际的业务模型来算，不是拍脑袋。用那个公式：maxActive = QPS × 响应时间 × 1.5。
2. 一定要开泄漏检测。
removeAbandoned=true 是救命稻草。即使代码写错了，它也能在连接泄漏后帮你回收，给你争取排查时间。
3. 缓存是第一道防线。
能不进数据库的请求，尽量不进。Redis + 本地缓存的组合，能挡住90%以上的流量。
4. 读写分离是最简单的优化。
只需要一个从库，就能把读压力分流一半以上。这是性价比最高的优化手段。
5. 分库分表是最后的手段。
不要过早优化。先做连接池调优、读写分离、缓存，如果还不够，再考虑分库分表。分库分表带来的复杂度和运维成本很高。
6. 监控和告警不能少。
不知道系统在什么状态下，所有优化都是瞎子摸象。把连接池、Redis、数据库的指标全部接上监控，设置合理的告警阈值。
那次崩溃让我们损失了数十万的GMV，但也让我们建立了现在这套高并发架构。希望这篇指南能帮到你，至少让你少踩几个坑。
如果有具体的问题，欢迎在评论区讨论。记住，高并发的核心不是技术有多复杂，而是对每一个环节的理解有多深。