电商大促秒杀MySQL数据库崩溃怎么办 高并发优化从缓存架构到读写分离完整策略
凌晨三点的电话
去年双11，我接到一个电话，对方是某头部电商的技术负责人，声音都带颤：”MySQL扛不住了，QPS直接打满，连接数爆掉，现在用户下单直接超时，怎么办？”
我花了30分钟给他梳理了一整套方案，事后复盘，这套思路其实适用于几乎所有高并发秒杀场景。今天就把这套完整策略讲透，从架构设计到代码落地，一步一步来。
先搞清楚：秒杀为什么会把MySQL打崩
很多人第一反应是”加服务器”，但这只是治标。你得先明白问题出在哪。
秒杀的核心特征
- 瞬时流量爆炸：平时QPS 500，秒杀开始直接飙到 50,000+，100倍的增长不是线性叠加，是指数级的。
- 热点数据集中：所有用户都在抢同一批商品，缓存穿透几乎不存在，但对数据库的压力是结构性的。
- 写操作密集：下单、扣库存、生成订单，每一步都在写库。
- 事务阻塞：MySQL的行锁、表锁在高并发下成为瓶颈，连接数被瞬间耗尽。
真实案例：某生鲜电商的教训
2022年春节档促销，某生鲜平台秒杀活动中，活动开始第3分钟，MySQL错误日志出现大量：
Aborted connection 284712 to db: 'prod_order' user: 'app_read'
host: '10.0.12.45' (Got an error reading communication packets)
同时慢查询日志里一堆：
-- Time: 2022-02-10T10:00:03.128472Z
-- User@Host: app_write[app_write] @ 10.0.15.22 [10.0.15.22]
-- Query_time: 2.847123 Lock_time: 0.000123 Rows_sent: 0 Rows_examined: 154231
SET timestamp=1644475203;
UPDATE inventory SET stock = stock - 1 WHERE item_id = 8849237 AND stock > 0;
这条UPDATE看着简单，但秒杀场景下，同一行数据被上万并发线程同时争抢，InnoDB的行锁竞争直接让线程排队等锁，MySQL的线程池被撑爆，连接数达到1000上限后，新连接直接被拒绝，用户看到的就是”系统繁忙”。
第一道防线：缓存架构——把流量挡在数据库外面
为什么必须先做缓存
核心思路：让缓存扛住99%的请求，数据库只处理真实下单的少数请求。
Redis缓存架构设计
1. 预热：把商品数据提前加载到缓存
秒杀开始前，就必须把商品库存、商品信息全部预热到Redis，不能等到流量来了再查库。
import redis
import json
from redis.client import Redis
from datetime import datetime
class SeckillCachePreload:
"""秒杀缓存预热组件"""
def __init__(self, redis_host='10.0.12.100', redis_port=6379, db=0):
self.redis = Redis(host=redis_host, port=redis_port, db=db,
decode_responses=True, socket_timeout=2)
def preload_item(self, item_id: int, item_info: dict, stock: int):
"""预热单个商品"""
# 商品信息缓存
self.redis.hset(f"seckill:item:{item_id}", mapping=item_info)
# 库存缓存，使用STRING类型方便原子操作
self.redis.set(f"seckill:stock:{item_id}", stock)
# 设置过期时间，防止数据脏读
self.redis.expire(f"seckill:stock:{item_id}", 86400)
# 记录预热时间
self.redis.set(f"seckill:preload_time:{item_id}", datetime.now().isoformat())
def preload_batch(self, items: list):
"""批量预热，使用Pipeline大幅提升性能"""
pipe = self.redis.pipeline(transaction=False)
for item in items:
pipe.hset(f"seckill:item:{item['id']}", mapping=item['info'])
pipe.set(f"seckill:stock:{item['id']}", item['stock'])
pipe.expire(f"seckill:stock:{item['id']}", 86400)
pipe.execute()
print(f"批量预热完成，共{len(items)}个商品")
2. 库存扣减：用Redis做原子扣减，而非直接操作数据库
这是最关键的一步。秒杀下单的核心逻辑是”扣库存”，如果每次都去MySQL执行UPDATE inventory SET stock=stock-1，数据库必死无疑。
import redis
import uuid
import json
from redis.clients import Redis
class SeckillStockService:
"""秒杀库存扣减服务"""
def __init__(self, redis_host='10.0.12.100', redis_port=6379, db=0):
self.redis = Redis(host=redis_host, port=redis_port, db=db,
decode_responses=True, socket_timeout=2,
max_connections=200)
def try_deduct_stock(self, item_id: int, user_id: str) -> dict:
"""
原子扣减库存，返回结果
返回格式: {'success': bool, 'order_no': str, 'msg': str}
"""
stock_key = f"seckill:stock:{item_id}"
user_set_key = f"seckill:user_buy:{item_id}"
# 使用Lua脚本保证原子性，这是核心
# Lua脚本在Redis中是原子执行的，不会有并发问题
lua_script = """
local stock_key = KEYS[1]
local user_set_key = KEYS[2]
local user_id = ARGV[1]
local order_no = ARGV[2]
-- 检查用户是否已经购买过
if redis.call('sismember', user_set_key, user_id) == 1 then
return {-1, '重复购买，每个用户限购一件'}
end
-- 尝试原子扣减库存
local stock = redis.call('decr', stock_key)
if stock < 0 then
-- 库存扣减失败，恢复并返回
redis.call('incr', stock_key)
return {-2, '库存不足'}
end
-- 扣减成功，记录用户购买信息
redis.call('sadd', user_set_key, user_id)
-- 设置用户购买记录的过期时间（活动结束后清理）
redis.call('expire', user_set_key, 86400)
return {1, order_no}
"""
# 生成唯一订单号
order_no = self._generate_order_no(item_id, user_id)
# 执行Lua脚本
result = self.redis.eval(
lua_script,
2, # 2个KEYS参数
stock_key,
user_set_key,
user_id,
order_no
)
status = int(result[0])
if status == 1:
return {'success': True, 'order_no': result[1], 'msg': '下单成功'}
elif status == -1:
return {'success': False, 'order_no': '', 'msg': result[1]}
elif status == -2:
return {'success': False, 'order_no': '', 'msg': result[1]}
else:
return {'success': False, 'order_no': '', 'msg': '系统异常'}
def _generate_order_no(self, item_id: int, user_id: str) -> str:
"""生成唯一订单号"""
timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
unique_part = uuid.uuid4().hex[:8]
return f"SK{timestamp}{unique_part}{item_id}{user_id[-4:]}"
为什么要用Lua脚本？
如果不使用Lua，而是分两步操作（先decr再sadd），在高并发下会出现两个线程同时通过库存检查，导致超卖。Lua脚本在Redis中是原子执行的，从开始到结束没有其他命令能插进来，这才是正确的做法。
3. 限流：Redis计数器实现接口限流
class RateLimiter:
"""基于Redis的滑动窗口限流器"""
def __init__(self, redis_host='10.0.12.100', redis_port=6379, db=0):
self.redis = Redis(host=redis_host, port=redis_port, db=db,
decode_responses=True)
def is_allowed(self, user_id: str, item_id: int, max_requests: int = 3, window_seconds: int = 10) -> bool:
"""
检查用户是否在限流范围内
每个用户每秒最多请求N次
"""
key = f"rate_limit:{user_id}:{item_id}"
current = self.redis.incr(key)
if current == 1:
# 第一次请求，设置过期时间
self.redis.expire(key, window_seconds)
return current <= max_requests
第二道防线：消息队列——削峰填谷，异步处理
为什么需要MQ
即使Redis扛住了瞬时流量，最终还是要落库。把下单请求打到MySQL还是太猛，需要一个”缓冲池”来消化流量。
完整的秒杀下单流程
用户请求 → 网关限流 → Redis预扣库存 → MQ异步下单 → MySQL最终落库
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
@Service
public class SeckillOrderService {
@Autowired
private StringRedisTemplate redisTemplate;
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
private static final String SECKILL_STOCK_KEY_PREFIX = "seckill:stock:";
private static final String SECKILL_USER_KEY_PREFIX = "seckill:user_buy:";
private static final String SECKILL_ORDER_QUEUE = "seckill_order_queue";
/**
* 秒杀下单主入口
*/
public SeckillResult seckillOrder(String userId, Long itemId) {
String stockKey = SECKILL_STOCK_KEY_PREFIX + itemId;
String userSetKey = SECKILL_USER_KEY_PREFIX + itemId;
// 1. 检查Redis中是否有该商品库存预热
if (!redisTemplate.hasKey(stockKey)) {
return SeckillResult.error("活动未开始或商品不存在");
}
// 2. 执行Lua脚本原子扣减库存
String luaScript =
"if redis.call('sismember', KEYS[2], ARGV[1]) == 1 then " +
" return {-1, '重复购买'} " +
"end " +
"local stock = redis.call('decr', KEYS[1]) " +
"if stock < 0 then " +
" redis.call('incr', KEYS[1]) " +
" return {-2, '库存不足'} " +
"end " +
"redis.call('sadd', KEYS[2], ARGV[1]) " +
"redis.call('expire', KEYS[2], 86400) " +
"return {1, ARGV[2]}";
DefaultRedisScript<List<Long>> script = new DefaultRedisScript<>();
script.setScriptText(luaScript);
script.setResultType(List.class);
String orderNo = generateOrderNo(userId, itemId);
List<Long> result = redisTemplate.execute(
script,
Arrays.asList(stockKey, userSetKey),
userId, orderNo
);
long status = result.get(0);
if (status == -1) {
return SeckillResult.error("重复购买");
}
if (status == -2) {
return SeckillResult.error("库存不足");
}
// 3. 扣减成功，发送MQ消息异步落库
SeckillOrderMessage message = new SeckillOrderMessage();
message.setOrderNo(orderNo);
message.setUserId(userId);
message.setItemId(itemId);
message.setCreateTime(System.currentTimeMillis());
kafkaTemplate.send(SECKILL_ORDER_QUEUE, JSON.toJSONString(message));
return SeckillResult.success(orderNo, "下单成功，请等待支付");
}
/**
* Kafka消费者：将订单最终写入MySQL
*/
@KafkaListener(topics = SECKILL_ORDER_QUEUE, groupId = "seckill-order-group")
public void handleOrderMessage(String messageJson) {
try {
SeckillOrderMessage message = JSON.parseObject(messageJson, SeckillOrderMessage.class);
// 4. 写入MySQL订单表
Order order = new Order();
order.setOrderNo(message.getOrderNo());
order.setUserId(message.getUserId());
order.setItemId(message.getItemId());
order.setStatus(OrderStatus.PENDING_PAYMENT);
order.setCreateTime(new Date());
orderMapper.insert(order);
// 5. 扣减MySQL库存（此时库存已在Redis扣减，这里是最终一致性保障）
orderMapper.deductStock(message.getItemId());
log.info("订单落库成功: orderNo={}, itemId={}, userId={}",
message.getOrderNo(), message.getItemId(), message.getUserId());
} catch (Exception e) {
log.error("订单落库失败，消息: {}", messageJson, e);
// 实际生产环境需要重试机制和死信队列
// 这里简化处理，后续可接入延迟重试
}
}
private String generateOrderNo(String userId, Long itemId) {
return "SK" + System.currentTimeMillis() + UUID.randomUUID().toString().substring(0, 8)
+ itemId + userId.substring(userId.length() - 4);
}
}
MQ消费者侧的优化
消费者从MQ拉取消息后写库，这里也要注意不能直接把消息全部打到MySQL。需要做：
- 批量写入：攒一批消息（比如100条）一次性批量INSERT，比单条INSERT性能高一个数量级
- 分库分表：订单表按user_id分片，避免单表过大
- 重试机制：失败的消息进入死信队列，延迟重试
@KafkaListener(topics = SECKILL_ORDER_QUEUE, groupId = "seckill-order-batch-group")
public void handleBatchOrderMessage(List<String> messages) {
// 批量处理，提升写入性能
List<Order> orders = new ArrayList<>();
for (String messageJson : messages) {
try {
SeckillOrderMessage message = JSON.parseObject(messageJson, SeckillOrderMessage.class);
Order order = new Order();
order.setOrderNo(message.getOrderNo());
order.setUserId(message.getUserId());
order.setItemId(message.getItemId());
order.setStatus(OrderStatus.PENDING_PAYMENT);
order.setCreateTime(new Date());
orders.add(order);
} catch (Exception e) {
log.error("解析消息失败: {}", messageJson, e);
}
}
if (!orders.isEmpty()) {
// 批量插入，大幅提升性能
orderMapper.batchInsert(orders);
log.info("批量落库成功，本次处理{}条订单", orders.size());
}
}
对应的Mapper：
<!-- MyBatis批量插入 -->
<insert id="batchInsert" parameterType="java.util.List">
INSERT INTO t_order (order_no, user_id, item_id, status, create_time)
VALUES
<foreach collection="list" item="item" separator=",">
(#{item.orderNo}, #{item.userId}, #{item.itemId}, #{item.status}, #{item.createTime})
</foreach>
</insert>
<!-- 扣减库存，使用行锁保证安全 -->
<update id="deductStock">
UPDATE t_inventory
SET stock = stock - 1
WHERE item_id = #{itemId} AND stock > 0
</update>
第三道防线：MySQL读写分离——让读操作不再影响写性能
为什么读写分离重要
秒杀场景中，查询类请求（查看商品详情、库存余量）远多于写请求。如果所有请求都打到主库，主库资源被读操作占用，写操作的延迟会急剧上升。
架构设计
主库（Master） ← 写操作（下单、扣库存）
|
└── 异步复制 ──→ 从库（Slave1/Slave2/Slave3）← 读操作（查商品、查库存）
ShardingSphere实现读写分离
# application-sharding.yml
spring:
shardingsphere:
datasource:
names: master,slave1,slave2,slave3
# 主库配置
master:
type: com.zaxxer.hikari.HikariDataSource
driver-class-name: com.mysql.cj.jdbc.Driver
jdbc-url: jdbc:mysql://10.0.12.200:3306/seckill_db?useSSL=false&serverTimezone=Asia/Shanghai
username: seckill_write
password: ${DB_WRITE_PASSWORD}
hikari:
maximum-pool-size: 50
minimum-idle: 10
connection-timeout: 30000
# 从库配置
slave1:
type: com.zaxxer.hikari.HikariDataSource
driver-class-name: com.mysql.cj.jdbc.Driver
jdbc-url: jdbc:mysql://10.0.12.201:3306/seckill_db?useSSL=false&serverTimezone=Asia/Shanghai
username: seckill_read
password: ${DB_READ_PASSWORD}
hikari:
maximum-pool-size: 100
minimum-idle: 20
slave2:
type: com.zaxxer.HikariDataSource
driver-class-name: com.mysql.cj.jdbc.Driver
jdbc-url: jdbc:mysql://10.0.12.202:3306/seckill_db?useSSL=false&serverTimezone=Asia/Shanghai
username: seckill_read
password: ${DB_READ_PASSWORD}
hikari:
maximum-pool-size: 100
minimum-idle: 20
slave3:
type: com.zaxxer.hikari.HikariDataSource
driver-class-name: com.mysql.cj.jdbc.Driver
jdbc-url: jdbc:mysql://10.0.12.203:3306/seckill_db?useSSL=false&serverTimezone=Asia/Shanghai
username: seckill_read
password: ${DB_READ_PASSWORD}
hikari:
maximum-pool-size: 100
minimum-idle: 20
rules:
# 读写分离规则
readwrite-splitting:
data-sources:
ds:
write-data-source-name: master
read-data-source-names: [slave1, slave2, slave3]
load-balancer-name: round_robin # 轮询负载均衡
props:
write-data-source-name: master
read-data-source-names: "${read-data-source-names}"
# 负载均衡规则
load-balancers:
round_robin:
type: ROUND_ROBIN
props:
sql-show: true # 打印SQL，方便调试
代码层的使用
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@Service
public class SeckillItemService {
// 写库JdbcTemplate（必须使用主库）
@Resource(name = "writeJdbcTemplate")
private JdbcTemplate writeJdbcTemplate;
// 读库JdbcTemplate（走从库）
@Resource(name = "readJdbcTemplate")
private JdbcTemplate readJdbcTemplate;
/**
* 查询商品秒杀信息 —— 走读库
*/
public SeckillItemDTO getSeckillItem(Long itemId) {
String sql = "SELECT i.id, i.name, i.cover, i.seckill_price, " +
"i.seckill_start_time, i.seckill_end_time, " +
"i.total_stock, i sold_stock " +
"FROM t_seckill_item i " +
"WHERE i.id = ? AND i.status = 1";
return readJdbcTemplate.queryForObject(sql, new Object[]{itemId},
(rs, num) -> {
SeckillItemDTO dto = new SeckillItemDTO();
dto.setId(rs.getLong("id"));
dto.setName(rs.getString("name"));
dto.setCover(rs.getString("cover"));
dto.setSeckillPrice(rs.getBigDecimal("seckill_price"));
dto.setTotalStock(rs.getInt("total_stock"));
dto.setSoldStock(rs.getInt("sold_stock"));
dto.setStartTime(rs.getTimestamp("seckill_start_time").getTime());
dto.setEndTime(rs.getTimestamp("seckill_end_time").getTime());
return dto;
});
}
/**
* 秒杀下单 —— 必须走写库，且需要事务
*/
@Transactional(rollbackFor = Exception.class)
public void createSeckillOrder(String orderNo, Long itemId, String userId) {
// 1. 插入订单
String insertOrderSql = "INSERT INTO t_order (order_no, user_id, item_id, " +
"amount, status, create_time) " +
"VALUES (?, ?, ?, (SELECT seckill_price FROM t_seckill_item WHERE id=?), " +
"'PENDING_PAYMENT', NOW())";
writeJdbcTemplate.update(insertOrderSql, orderNo, userId, itemId, itemId);
// 2. 更新已售数量
String updateSoldSql = "UPDATE t_seckill_item SET sold_stock = sold_stock + 1 WHERE id = ?";
writeJdbcTemplate.update(updateSoldSql, itemId);
}
}
从库延迟问题怎么解决
读写分离最大的坑是主从延迟。Redis扣了库存，但读从库查到的库存还是旧的，导致超卖。
解决方案：
- 关键查询强制走主库：库存查询、订单状态查询这类涉及一致性的，用
@Master注解或直接使用写库JdbcTemplate - Redis作为最终数据源：库存查询优先查Redis，不查MySQL
- 监控延迟：建立主从延迟监控，延迟超过阈值时自动切回读主库
/**
* 库存查询 —— 优先Redis，查不到再查主库（不走从库）
*/
public int getStockFromRedisOrMaster(Long itemId) {
String stockKey = "seckill:stock:" + itemId;
// 先查Redis
String stockStr = redisTemplate.opsForValue().get(stockKey);
if (stockStr != null) {
return Integer.parseInt(stockStr);
}
// Redis没有，查主库（注意：不走从库）
String sql = "SELECT stock FROM t_inventory WHERE item_id = ? FOR UPDATE";
Integer stock = writeJdbcTemplate.queryForObject(sql, Integer.class, itemId);
if (stock != null) {
// 回写Redis，设置短期过期
redisTemplate.opsForValue().set(stockKey, String.valueOf(stock), 300, TimeUnit.SECONDS);
}
return stock != null ? stock : 0;
}
第四道防线：数据库本身优化
即使有了上述架构，MySQL本身的优化也不能忽视。
1. 表结构优化
-- 核心表结构示例
-- 订单表：按user_id分片时不需要，但字段要精简
CREATE TABLE t_order (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
order_no VARCHAR(32) NOT NULL UNIQUE COMMENT '订单号',
user_id BIGINT UNSIGNED NOT NULL COMMENT '用户ID',
item_id BIGINT UNSIGNED NOT NULL COMMENT '商品ID',
amount DECIMAL(10,2) NOT NULL COMMENT '订单金额',
status TINYINT NOT NULL DEFAULT 0 COMMENT '0待支付 1已支付 2已取消 3已过期',
create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
pay_time DATETIME,
INDEX idx_user_id (user_id),
INDEX idx_item_id (item_id),
INDEX idx_status_create (status, create_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='秒杀订单表';
-- 库存表：只保留必要字段，避免大字段
CREATE TABLE t_inventory (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
item_id BIGINT UNSIGNED NOT NULL UNIQUE COMMENT '商品ID',
total_stock INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '总库存',
stock INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '剩余库存',
sold_stock INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '已售数量',
version INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '乐观锁版本号',
INDEX idx_item_id (item_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='秒杀库存表';
-- 活动表：秒杀活动配置
CREATE TABLE t_seckill_activity (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
activity_no VARCHAR(32) NOT NULL UNIQUE,
title VARCHAR(128) NOT NULL,
start_time DATETIME NOT NULL,
end_time DATETIME NOT NULL,
status TINYINT NOT NULL DEFAULT 0 COMMENT '0未开始 1进行中 2已结束',
create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
INDEX idx_status_time (status, start_time, end_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='秒杀活动表';
2. MySQL参数调优
# my.cnf 关键配置（针对高并发秒杀场景）
[mysqld]
# 连接数：根据业务调整，秒杀场景可以设高一些
max_connections = 2000
# 线程缓存，减少创建/销毁线程的开销
thread_cache_size = 512
# InnoDB缓冲池大小，建议设为物理内存的50%-70%
innodb_buffer_pool_size = 8G
innodb_buffer_pool_instances = 8
# 日志相关：刷盘策略影响性能和持久性
innodb_flush_log_at_trx_commit = 2 # 每秒刷盘，性能更好
innodb_log_buffer_size = 16M
innodb_log_file_size = 512M
# 并发控制
innodb_thread_concurrency = 0 # 0表示不限制
innodb_write_io_threads = 8
innodb_read_io_threads = 8
# 关闭同步binlog，提升写性能（主从架构下）
sync_binlog = 0
# 减少日志写入频率
innodb_flush_log_at_timeout = 3
# 调整排序缓冲区
sort_buffer_size = 2M
read_buffer_size = 2M
read_rnd_buffer_size = 2M
3. SQL优化原则
-- ❌ 错误写法：SELECT *，查询了不需要的字段
SELECT * FROM t_seckill_item WHERE id = 12345;
-- ✅ 正确写法：只查需要的字段
SELECT id, name, cover, seckill_price, total_stock, sold_stock
FROM t_seckill_item WHERE id = 12345;
-- ❌ 错误写法：对大字段做函数操作，无法走索引
SELECT * FROM t_order WHERE DATE(create_time) = '2024-11-11';
-- ✅ 正确写法：范围查询走索引
SELECT * FROM t_order WHERE create_time >= '2024-11-11 00:00:00'
AND create_time < '2024-11-12 00:00:00';
-- ❌ 错误写法：隐式类型转换，索引失效
SELECT * FROM t_order WHERE order_no = 12345; -- order_no是字符串，12345是数字
-- ✅ 正确写法：类型一致
SELECT * FROM t_order WHERE order_no = '12345';
-- ❌ 错误写法：回表查询过多
SELECT id, name, cover, detail, specs, params, price, ...
FROM t_seckill_item WHERE id IN (1,2,3,4,5,6,7,8,9,10);
-- 如果detail是TEXT大字段，每条都要回表
-- ✅ 正确写法：分页或拆表，避免大字段查询
SELECT id, name, cover, price FROM t_seckill_item WHERE id IN (...);
-- 详情单独查
4. 连接池配置
秒杀场景下，连接池配置直接影响数据库承载能力：
# HikariCP配置（Spring Boot默认连接池）
spring:
datasource:
hikari:
maximum-pool-size: 100 # 最大连接数，根据MySQL max_connections调整
minimum-idle: 20 # 最小空闲连接
idle-timeout: 30000 # 空闲连接超时
max-lifetime: 1800000 # 连接最大生命周期
connection-timeout: 5000 # 获取连接超时时间（秒杀场景适当缩短）
leak-detection-threshold: 30000 # 连接泄露检测
第五道防线：架构层面的兜底策略
全链路架构总览
┌─────────────────────┐
│ 用户端/APP │
└──────────┬──────────┘
│
┌──────────▼──────────┐
│ CDN + 静态资源 │
│ (商品图片/页面缓存) │
└──────────┬──────────┘
│
┌──────────▼──────────┐
│ API网关层 │
│ 限流/鉴权/路由 │
│ (Gateway RateLimit) │
└──────────┬──────────┘
│
┌──────────▼──────────┐
│ 服务层 (Spring) │
│ ┌───────────────┐ │
│ │ Redis缓存层 │ │ ← 拦截99%请求
│ │ (预扣库存) │ │
│ └───────┬───────┘ │
│ ┌───────▼───────┐ │
│ │ MQ消息队列 │ │ ← 削峰填谷
│ │ (Kafka/RocketMQ)│
│ └───────┬───────┘ │
│ ┌───────▼───────┐ │
│ │ MySQL主从 │ │ ← 最终持久化
│ │ (读写分离) │ │
│ └───────────────┘ │
└─────────────────────┘
熔断降级：扛不住时的最后手段
当Redis和MySQL都快要撑不住时，需要熔断保护，避免雪崩：
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
import io.github.resilience4j.ratelimiter.RateLimiter;
import io.github.resilience4j.ratelimiter.RateLimiterConfig;
import java.time.Duration;
import java.util.concurrent.Callable;
@Service
public class SeckillCircuitBreakerService {
// 熔断器：当错误率超过阈值时自动熔断，保护下游
private final CircuitBreaker seckillCircuitBreaker;
// 限流器：限制请求速率
private final RateLimiter seckillRateLimiter;
public SeckillCircuitBreakerService() {
// 熔断器配置
CircuitBreakerConfig config = CircuitBreakerConfig.custom()
.failureRateThreshold(50) // 错误率超过50%熔断
.waitDurationInOpenState(Duration.ofSeconds(30)) // 熔断后30秒尝试恢复
.slidingWindowType(CircuitBreakerConfig.SlidingWindowType.COUNT_BASED)
.slidingWindowSize(100) // 最近100次请求
.minimumNumberOfCalls(20) // 至少20次才触发
.enableAutomaticTransitionFromOpenToHalfOpen()
.build();
this.seckillCircuitBreaker = CircuitBreakerRegistry.of(config)
.circuitBreaker("seckill-cb");
// 限流器配置
RateLimiterConfig rateLimiterConfig = RateLimiterConfig.custom()
.limitRefreshPeriod(Duration.ofSeconds(1))
.limitForPeriod(1000) // 每秒最多1000个请求通过
.timeoutDuration(Duration.ofSeconds(0))
.build();
this.seckillRateLimiter = RateLimiterRegistry.of(rateLimiterConfig)
.rateLimiter("seckill-rl");
}
/**
* 带熔断和限流的秒杀下单
*/
public SeckillResult seckillWithProtection(String userId, Long itemId) {
// 1. 先过限流
if (!seckillRateLimiter.tryAcquirePermission()) {
return SeckillResult.error("请求过于频繁，请稍后再试");
}
// 2. 带熔断的执行
Callable<SeckillResult> callable = () -> {
return seckillOrderService.seckillOrder(userId, itemId);
};
try {
return seckillCircuitBreaker.executeCallable(callable);
} catch (Exception e) {
if (e.getCause() instanceof CircuitBreakerOpenException) {
return SeckillResult.error("系统繁忙，请稍后再试");
}
log.error("秒杀下单异常", e);
return SeckillResult.error("系统异常，请稍后再试");
}
}
}
降级策略：当Redis不可用时
@Service
public class SeckillFallbackService {
@Autowired
private SeckillOrderService seckillOrderService;
/**
* 当Redis不可用时的降级方案
* 直接进入MySQL层，但需要更严格的限流
*/
public SeckillResult seckillFallback(String userId, Long itemId) {
// 降级时限流更严格，每秒只处理10个请求
if (!FallbackRateLimiter.tryAcquire()) {
return SeckillResult.error("系统繁忙，请稍后再试");
}
try {
// 直接使用MySQL乐观锁扣库存
return seckillOrderService.seckillDirectToDb(userId, itemId);
} catch (Exception e) {
log.error("降级方案执行失败", e);
return SeckillResult.error("系统异常");
}
}
}
监控与告警：不能盲人摸象
关键监控指标
# Prometheus监控配置示例
scrape_configs:
- job_name: 'seckill-mysql'
static_configs:
- targets: ['10.0.12.200:9104'] # MySQL exporter
metrics_path: /metrics
- job_name: 'seckill-redis'
static_configs:
- targets: ['10.0.12.100:9121'] # Redis exporter
- job_name: 'seckill-app'
static_configs:
- targets: ['10.0.15.10:8080', '10.0.15.11:8080'] # Spring Boot Actuator
关键告警规则
# AlertManager告警规则
groups:
- name: seckill_critical
rules:
# MySQL连接数超过80%告警
- alert: MySQLConnectionHigh
expr: mysql_global_status_threads_connected / mysql_global_variables_max_connections > 0.8
for: 1m
labels:
severity: warning
annotations:
summary: "MySQL连接数过高"
description: "当前连接数 {{ $value | humanizePercentage }}，请及时处理"
# Redis内存使用超过80%
- alert: RedisMemoryHigh
expr: redis_memory_used_bytes / redis_memory_max_bytes > 0.8
for: 2m
labels:
severity: warning
annotations:
summary: "Redis内存使用率过高"
# 秒杀下单失败率超过10%
- alert: SeckillOrderFailRateHigh
expr: rate(seckill_order_fail_total[5m]) / rate(seckill_order_total[5m]) > 0.1
for: 1m
labels:
severity: critical
annotations:
summary: "秒杀下单失败率过高"
# 主从延迟超过5秒
- alert: MySQLReplicationLag
expr: mysql_slave_status_seconds_behind_master > 5
for: 1m
labels:
severity: warning
annotations:
summary: "MySQL主从延迟过高"
完整流程回顾
把整套方案串起来，一个典型的秒杀请求是这样的：
1. 用户点击"抢购"
↓
2. 请求到达API网关，网关做基础限流（每秒10万请求）
↓
3. 请求到达服务层，先检查熔断器状态
↓
4. 进入Redis预扣库存（Lua脚本原子操作）
- 库存不足 → 直接返回"库存不足"，结束
- 重复购买 → 直接返回"重复购买"，结束
- 扣减成功 → 继续
↓
5. 发送MQ消息，立即返回"下单成功"给用户
↓
6. MQ消费者异步处理：
- 批量写入MySQL订单表
- 扣减MySQL库存
- 记录操作日志
↓
7. 用户进入支付页面，等待支付
关键数字
| 组件 | 预估处理能力 |
|---|---|
| API网关（限流） | 10万 QPS |
| Redis（Lua脚本） | 5万+ QPS |
| Kafka（消息积压） | 10万+ 消息/秒 |
| MySQL（批量写入） | 5000+ TPS |
| 整体系统 | 3万+ QPS秒杀 |
实战中的几个”坑”
坑1：Redis和MySQL库存不一致
这是最常见的问题。Redis扣了库存，但MySQL写失败，导致实际库存对不上。
解决方案：
- 采用最终一致性策略：Redis作为主要数据源，MySQL作为备份
- 定时任务：每5分钟比对Redis和MySQL库存，发现差异时以Redis为准修正MySQL
- 对账：活动结束后进行全面对账
@Component
public class StockReconcileTask {
@Autowired
private StringRedisTemplate redisTemplate;
@Autowired
private InventoryMapper inventoryMapper;
/**
* 每5分钟执行一次库存对账
*/
@Scheduled(cron = "0 */5 * * * *")
public void reconcileStock() {
List<Long> itemIds = inventoryMapper.selectAllSeckillItemIds();
for (Long itemId : itemIds) {
String redisStock = redisTemplate.opsForValue().get("seckill:stock:" + itemId);
Inventory dbInventory = inventoryMapper.selectById(itemId);
int redisVal = redisStock != null ? Integer.parseInt(redisStock) : 0;
int dbVal = dbInventory != null ? dbInventory.getStock() : 0;
if (redisVal != dbVal) {
log.warn("库存不一致，itemId={}, redis={}, db={}", itemId, redisVal, dbVal);
// 以Redis为准，修正MySQL
inventoryMapper.updateStock(itemId, redisVal);
}
}
}
}
坑2：缓存穿透
秒杀场景下，非秒杀商品也被大量请求查询，而这些商品不存在，每次都会打到MySQL。
解决方案：
- 对不存在的商品，也缓存一个”空值”（比如-1），设置较短过期时间
- 在Redis层做布隆过滤器，快速判断商品是否存在
public SeckillItemDTO getItemSafe(Long itemId) {
String key = "seckill:item:" + itemId;
// 先查缓存
String cached = redisTemplate.opsForValue().get(key);
if (cached != null) {
if ("NULL".equals(cached)) {
return null; // 商品不存在，直接返回，不再查库
}
return JSON.parseObject(cached, SeckillItemDTO.class);
}
// 缓存没有，查数据库
SeckillItemDTO item = itemMapper.selectSeckillItem(itemId);
if (item == null) {
// 缓存空值，防止穿透，过期时间短一些
redisTemplate.opsForValue().set(key, "NULL", 60, TimeUnit.SECONDS);
return null;
}
// 缓存商品数据，过期时间2小时
redisTemplate.opsForValue().set(key, JSON.toJSONString(item), 7200, TimeUnit.SECONDS);
return item;
}
坑3：热点KEY问题
当某个商品极其热门，所有请求都打到Redis的同一个key上，Redis单热点也会扛不住。
解决方案：
- 本地缓存：在应用服务器本地（Caffeine/Guava）缓存热点数据
- 分片缓存：将库存分片到多个key，分散压力
/**
* 本地缓存 + Redis二级缓存
*/
@Service
public class HotItemCacheService {
// 本地缓存：容量1000，过期时间30秒
private final Cache<Long, SeckillItemDTO> localCache = Caffeine.newBuilder()
.maximumSize(1000)
.expireAfterWrite(30, TimeUnit.SECONDS)
.build();
@Autowired
private StringRedisTemplate redisTemplate;
public SeckillItemDTO getHotItem(Long itemId) {
// 1. 先查本地缓存
SeckillItemDTO cached = localCache.getIfPresent(itemId);
if (cached != null) {
return cached;
}
// 2. 再查Redis
String key = "seckill:item:" + itemId;
String redisVal = redisTemplate.opsForValue().get(key);
if (redisVal != null) {
SeckillItemDTO item = JSON.parseObject(redisVal, SeckillItemDTO.class);
localCache.put(itemId, item); // 回写本地缓存
return item;
}
// 3. 最后查数据库
SeckillItemDTO item = itemMapper.selectSeckillItem(itemId);
if (item != null) {
localCache.put(itemId, item);
redisTemplate.opsForValue().set(key, JSON.toJSONString(item), 3600, TimeUnit.SECONDS);
}
return item;
}
}
秒杀前的 checklist
每次大促前，按这个清单过一遍：
- [ ] 压测：在预发环境用生产1.5倍的流量压测，确认各环节能承受
- [ ] 预热：商品数据、库存全部预热到Redis，检查预热是否完整
- [ ] 容量评估：确认MySQL、Redis、MQ的容量是否足够，连接池配置是否合理
- [ ] 监控告警：确认所有关键指标都有监控，告警通道畅通（电话/钉钉/短信）
- [ ] 降级预案：确认熔断、降级策略已配置，知道何时触发、如何手动触发
- [ ] 回滚方案：如果出现问题，知道如何快速回滚到上一个稳定版本
- [ ] 值班安排：确认DBA、Redis运维、应用开发都在岗，联系方式畅通
写在最后
这套方案不是银弹，真正的秒杀系统需要根据业务规模、团队能力、预算来做取舍。但核心思想是一致的：能挡在数据库外面的请求，坚决不让它进去。缓存扛流量、MQ削峰值、读写分离提升吞吐、熔断降级保护系统——这四个层次层层递进，构成了完整的防护体系。
我见过太多团队只做了其中一两层，然后在大促当天手忙脚乱。真正的高手，是把每一个环节都做到位，让系统在任何压力下都能优雅地降级，而不是直接崩掉。
如果你对某个环节有疑问，比如Redis集群的搭建、Kafka的配置、ShardingSphere的调优，我们可以继续深入聊。