电商秒杀数据库崩溃的教训 读写分离分库分表加Redis缓存与连接池优化提升MySQL并发能力
记得那是2023年618大促的前夜，我和团队刚接到一个紧急需求——优化秒杀系统的数据库架构。原因很简单：上周的测试中，我们的秒杀活动把MySQL直接干崩了。那晚，整个团队的咖啡都喝光了，屏幕上全是红色的报错日志。
一、崩溃现场：我们到底做错了什么
那天是个周二下午，运营同学发了个好消息：某爆款商品预定秒杀，预估峰值QPS在5万左右。我们在测试环境模拟了一下，结果——MySQL直接挂了。
ERROR 1040 (HY000): Too many connections
ERROR 2013 (HY000): Lost connection to MySQL server during query
我们当时的架构简直”惨不忍睹”：
用户请求 → Web服务器 → 单台MySQL主库（读写混用）
没有缓存，没有分库分表，连连接池都配置得乱七八糟。MySQL默认的最大连接数是151，我们生产环境配置成了500，但每次秒杀活动一来，瞬时并发轻松突破2000+，数据库直接拒绝新连接。
更致命的是，查询和写入混在一起。查询需要锁表，写入也需要锁表，互相排队，响应时间从几十毫秒飙升到几秒甚至超时。
那一刻我们终于明白：用单点MySQL扛秒杀，无异于用自行车送快递——不是不能用，是根本扛不住。
二、先搞清楚：秒杀场景到底在压什么
在我们动手之前，必须先搞清楚问题的本质。很多团队一上来就搞技术选型，结果做了半天，问题还是没解决。
秒杀请求的特征
时间维度：
- 集中在某一秒爆发（秒杀开始的那一刻）
- 持续时间很短（几分钟到半小时）
- 之后迅速回落
请求类型：
- 大部分是读请求（查询库存、商品详情）
- 少量是写请求（下单、扣库存）
- 写请求有强烈的竞争（多人抢同一件商品）
数据特征：
- 热点数据集中（就那几件秒杀商品）
- 非热点数据分散（普通商品浏览）
普通查询 vs 秒杀查询
-- 普通查询：读多写少，数据分散
SELECT * FROM products WHERE category_id = 1024;
-- 秒杀查询：极高并发读 + 激烈竞争写
-- 1. 查询秒杀商品详情（高并发读）
SELECT * FROM seckill_products WHERE id = 8888;
-- 2. 扣减库存（激烈竞争写，可能上百人同时抢同一件）
UPDATE seckill_products SET stock = stock - 1 WHERE id = 8888 AND stock > 0;
看到区别了吗？普通业务数据库设计可以”差不多就行”，但秒杀场景，每一个环节都得精心设计。
三、第一道防线：Redis缓存层
我们做的第一件事，就是把所有热点数据全部放进Redis。这不是什么新技术，但效果立竿见影。
缓存架构设计
用户请求 → Web服务器 → Redis缓存 → MySQL数据库（缓存未命中时）
↓
热点数据优先加载
商品详情缓存
秒杀商品的信息相对稳定，不会每秒都变。我们把商品信息完全缓存到Redis：
import redis
import json
import time
class SeckillCache:
def __init__(self):
self.redis = redis.Redis(
host='10.0.1.100',
port=6379,
db=0,
password='your_password',
decode_responses=True,
max_connections=50 # 连接池大小
)
# 获取商品详情，带缓存
def get_product(self, product_id):
cache_key = f"seckill:product:{product_id}"
# 先查Redis
product_json = self.redis.get(cache_key)
if product_json:
return json.loads(product_json)
# Redis没有，查MySQL
product = self._query_from_db(product_id)
if product:
# 写入Redis，设置过期时间（防止缓存穿透）
self.redis.setex(cache_key, 300, json.dumps(product)) # 5分钟过期
return product
return None
def _query_from_db(self, product_id):
# 这里走的是从库，主从分离后从库负责读
pass
库存缓存：最关键的部分
库存是最难处理的，因为它是实时的，多人同时扣减。如果用Redis做库存扣减，必须保证原子性：
class StockService:
def __init__(self, redis_client):
self.redis = redis_client
# 使用Lua脚本保证原子性
def deduct_stock(self, product_id, quantity=1):
# Lua脚本，原子操作
lua_script = """
local stock_key = KEYS[1]
local quantity = tonumber(ARGV[1])
local current_stock = redis.call('GET', stock_key)
if current_stock == false then
return -1 -- 库存不存在
end
current_stock = tonumber(current_stock)
if current_stock < quantity then
return 0 -- 库存不足
end
-- 扣减库存
redis.call('DECRBY', stock_key, quantity)
return 1 -- 扣减成功
"""
cache_key = f"seckill:stock:{product_id}"
result = self.redis.eval(lua_script, 1, cache_key, quantity)
return result
# 预热库存到Redis
def warmup_stock(self, product_id, stock):
cache_key = f"seckill:stock:{product_id}"
self.redis.set(cache_key, stock)
用Lua脚本的目的是保证原子性。如果用两条Redis命令（先GET再SET），在高并发下会出现竞态条件。Lua脚本在Redis中是原子执行的，不会出现两个请求同时读到同一个库存值然后都扣减的情况。
缓存穿透、击穿、雪崩的应对
class CacheProtection:
def __init__(self, redis_client):
self.redis = redis_client
# 缓存穿透：查询不存在的数据
def get_with_bloom_filter(self, product_id):
# 使用布隆过滤器判断数据是否存在
bloom_key = "seckill:bloom"
if not self.redis.hget(bloom_key, str(product_id)):
return None # 布隆过滤器说没有，大概率真的没有
# 正常查询流程
return self.seckill_cache.get_product(product_id)
# 缓存击穿：热点key过期瞬间
def get_with_lock(self, product_id):
cache_key = f"seckill:product:{product_id}"
lock_key = f"seckill:lock:{product_id}"
# 先查缓存
product_json = self.redis.get(cache_key)
if product_json:
return json.loads(product_json)
# 缓存没有，尝试获取分布式锁
if self.redis.set(lock_key, "1", ex=10, nx=True):
try:
# double-check（双重检查）
product_json = self.redis.get(cache_key)
if product_json:
return json.loads(product_json)
# 查数据库
product = self._query_from_db(product_id)
if product:
self.redis.setex(cache_key, 300, json.dumps(product))
else:
# 缓存空值，防止穿透
self.redis.setex(cache_key, 60, json.dumps({}))
return product
finally:
self.redis.delete(lock_key)
else:
# 其他线程在重建缓存，等待重试
time.sleep(0.1)
return self.get_with_lock(product_id) # 递归重试
四、第二道防线：读写分离
缓存解决了读多写少的问题，但还有一些查询走不到缓存，或者缓存未命中的情况。这时候，读写分离就派上用场了。
架构设计
┌─────────────┐
│ 主库 (Master) │
│ 负责写操作 │
└──────┬──────┘
│ 异步复制
▼
┌────────────────────────┐
│ 从库 (Slaves) │
│ 负责读操作（可多从） │
└────────────────────────┘
读写分离的实现
import pymysql
from db_utils import read_db_pool, write_db_pool
class ReadOnlyWriteService:
def __init__(self):
# 写操作走主库
self.write_db = write_db_pool
# 读操作走从库
self.read_db = read_db_pool
# 写操作
def create_order(self, order_data):
conn = self.write_db.get_connection()
try:
cursor = conn.cursor()
sql = """
INSERT INTO orders
(user_id, product_id, quantity, total_price, status, create_time)
VALUES (%s, %s, %s, %s, %s, NOW())
"""
cursor.execute(sql, (
order_data['user_id'],
order_data['product_id'],
order_data['quantity'],
order_data['total_price'],
'pending'
))
conn.commit()
return cursor.lastrowid
finally:
self.write_db.release_connection(conn)
# 读操作
def get_product(self, product_id):
conn = self.read_db.get_connection()
try:
cursor = conn.cursor()
sql = "SELECT * FROM products WHERE id = %s"
cursor.execute(sql, (product_id,))
return cursor.fetchone()
finally:
self.read_db.release_connection(conn)
主从复制的原理
MySQL主从复制是基于binlog的：
主库操作：
1. 执行写操作（INSERT/UPDATE/DELETE）
2. 将操作记录到binlog
3. 从库通过I/O线程读取binlog
4. 从库通过SQL线程回放binlog
延迟问题：
- 异步复制：主库不等待从库，延迟可能几秒
- 半同步复制：主库等待至少一个从库确认
- 在秒杀场景中，我们用的是异步复制，读操作可以接受一定延迟
读写分离的注意事项
# 关键原则：写完之后读不到最新数据
# 场景：用户刚下单，立马来查询订单状态
# 如果用从库查询，可能查不到刚写的订单
# 解决方案：自己写的自己读
class ReadWriteConsistency:
def create_and_read_order(self, order_data):
# 1. 写操作走主库
order_id = self.create_order(order_data)
# 2. 读操作也走主库（或者缓存）
# 避免读到从库的旧数据
conn = self.write_db.get_connection()
try:
cursor = conn.cursor()
sql = "SELECT * FROM orders WHERE id = %s"
cursor.execute(sql, (order_id,))
return cursor.fetchone()
finally:
self.write_db.release_connection(conn)
五、第三道防线：分库分表
即使做了读写分离，当数据量达到千万级、并发达到数万QPS时，单库的压力依然巨大。这时候必须分库分表。
为什么要分库分表
单库的瓶颈：
1. 连接数瓶颈：单库最多处理几百个并发连接
2. 磁盘I/O瓶颈：单盘读写有上限
3. CPU瓶颈：单CPU处理能力有限
4. 内存瓶颈：Buffer Pool大小有限
5. 锁竞争瓶颈：行锁、表锁争用
分库分表后：
- 连接数分散到多个库
- 数据分散到多个表
- 并发处理能力线性增长
分库策略
# 按用户ID分库：同一用户的数据在同一个库
# 好处：用户相关查询不走跨库
class ShardingStrategy:
def __init__(self, num_shards=8):
self.num_shards = num_shards
def get_shard_index(self, user_id):
# 取模分片
return int(user_id) % self.num_shards
def get_db_name(self, user_id):
shard_index = self.get_shard_index(user_id)
return f"seckill_db_{shard_index}"
def get_table_name(self, table_name, user_id):
shard_index = self.get_shard_index(user_id)
return f"{table_name}_{shard_index}"
分表策略
-- 假设我们分8个库，每个库2张表
-- 总共16张表
-- 创建分表
CREATE TABLE orders_0 (
id BIGINT PRIMARY KEY,
user_id BIGINT NOT NULL,
product_id INT NOT NULL,
quantity INT NOT NULL,
total_price DECIMAL(10,2),
status VARCHAR(20),
create_time DATETIME,
INDEX idx_user_id (user_id)
);
CREATE TABLE orders_1 (
id BIGINT PRIMARY KEY,
user_id BIGINT NOT NULL,
product_id INT NOT NULL,
quantity INT NOT NULL,
total_price DECIMAL(10,2),
status VARCHAR(20),
create_time DATETIME,
INDEX idx_user_id (user_id)
);
-- ... 以此类推到 orders_15
分库分表的路由
class ShardingRouter:
def __init__(self, num_shards=8):
self.num_shards = num_shards
def route_table(self, table_name, user_id):
shard_index = int(user_id) % self.num_shards
return f"{table_name}_{shard_index}"
def route_db(self, user_id):
shard_index = int(user_id) % self.num_shards
return f"seckill_db_{shard_index}"
# 查询订单
def query_order(self, order_id, user_id):
table_name = self.route_table("orders", user_id)
db_name = self.route_db(user_id)
sql = f"SELECT * FROM `{db_name}`.`{table_name}` WHERE id = %s"
# 执行查询...
分库分表的难点
1. 跨库查询： JOIN、分页、排序都需要额外处理
解决方案：避免跨库查询，需要时走ES等搜索引擎
2. 全局ID生成： 需要分布式ID
解决方案：使用雪花算法（Snowflake）
3. 扩容问题： 分片后数据重新分配很麻烦
解决方案：提前规划好分片数量，预留扩容空间
4. 复杂查询： 不能再用原生SQL了
解决方案：使用ShardingSphere等中间件
雪花算法生成全局ID
import time
import threading
class SnowflakeID:
def __init__(self, worker_id=1):
self.worker_id = worker_id
self.sequence = 0
self.last_timestamp = -1
self.lock = threading.Lock()
# 64位ID组成：
# 1位符号位（固定0）
# 41位时间戳（毫秒级，可用69年）
# 5位数据中心ID
# 5位机器ID
# 12位序列号（每毫秒最多4096个ID）
self.worker_id_bits = 5
self.datacenter_id_bits = 5
self.sequence_bits = 12
self.max_worker_id = -1 ^ (-1 << self.worker_id_bits)
self.max_datacenter_id = -1 ^ (-1 << self.datacenter_id_bits)
self.worker_id_shift = self.sequence_bits
self.datacenter_id_shift = self.sequence_bits + self.worker_id_bits
self.timestamp_shift = self.sequence_bits + self.worker_id_bits + self.datacenter_id_bits
self.sequence_mask = -1 ^ (-1 << self.sequence_bits)
def _til_next_millis(self, last_timestamp):
timestamp = time.time() * 1000
while timestamp <= last_timestamp:
timestamp = time.time() * 1000
return timestamp
def generate_id(self):
with self.lock:
timestamp = time.time() * 1000
if timestamp < self.last_timestamp:
raise Exception("Clock moved backwards")
if timestamp == self.last_timestamp:
self.sequence = (self.sequence + 1) & self.sequence_mask
if self.sequence == 0:
timestamp = self._til_next_millis(self.last_timestamp)
else:
self.sequence = 0
self.last_timestamp = timestamp
return ((int(timestamp) & 0xFFFFFFFFFFF) << self.timestamp_shift) | \
(self.worker_id << self.worker_id_shift) | \
self.sequence
# 使用
id_generator = SnowflakeID(worker_id=1)
order_id = id_generator.generate_id()
六、第四道防线：连接池优化
就算前面三关都过了，如果连接池配置不当，MySQL依然会崩溃。
连接池的问题
默认配置的问题：
- MySQL默认最大连接数：151
- 连接池默认大小：10-20
- 连接超时时间：太长或太短都不行
秒杀场景下的问题：
- 瞬间高并发，连接池不够用
- 连接泄漏（用完没释放）
- 连接超时，排队等待
连接池的配置优化
import pymysql
from dbutils.pooled_db import PooledDB
import threading
class ConnectionPool:
def __init__(self, config):
self.config = config
self.pool = None
self._lock = threading.Lock()
def get_pool(self):
if self.pool is None:
with self._lock:
if self.pool is None:
self.pool = PooledDB(
creator=pymysql,
# 最小连接数
mincached=20,
# 最大连接数（根据MySQL max_connections配置）
maxcached=100,
# 最大共享连接数
maxshared=50,
# 最大连接数
maxconnections=200,
# 连接阻塞数
blocking=True,
# 最大等待时间
maxusage=1000,
# 会话初始设置
setsession=[],
# 过期检查
reset=True,
# 主机配置
host=config['host'],
port=config['port'],
user=config['user'],
password=config['password'],
database=config['database'],
charset=config.get('charset', 'utf8mb4'),
# 连接超时
connect_timeout=5,
# 读写超时
read_timeout=30,
write_timeout=30,
# 自动重连
autocommit=True
)
return self.pool
def get_connection(self):
pool = self.get_pool()
return pool.connection()
def release_connection(self, conn):
pool = self.get_pool()
pool.putconn(conn)
MySQL端连接参数优化
-- MySQL配置文件 my.cnf 或 my.ini
[mysqld]
# 最大连接数（根据服务器内存调整，通常1000-2000）
max_connections = 2000
# 每个连接的最大线程数
thread_cache_size = 500
# 等待超时（秒），空闲连接超时后断开
wait_timeout = 30
interactive_timeout = 30
# 连接超时
connect_timeout = 10
# 网络缓冲
net_read_timeout = 30
net_write_timeout = 60
# 临时表
max_heap_table_size = 64M
tmp_table_size = 64M
# 排序和join缓冲区
sort_buffer_size = 2M
read_buffer_size = 2M
read_rnd_buffer_size = 2M
join_buffer_size = 8M
# InnoDB配置
innodb_buffer_pool_size = 4G -- 根据内存调整，通常是总内存的50-70%
innodb_log_file_size = 512M
innodb_flush_log_at_trx_commit = 2 -- 每秒刷盘，平衡性能和安全性
innodb_flush_method = O_DIRECT
innodb_io_capacity = 2000
innodb_io_capacity_max = 4000
连接池监控
import psutil
import pymysql
class ConnectionPoolMonitor:
def __init__(self, pool):
self.pool = pool
def get_stats(self):
"""获取连接池状态"""
stats = {
'connected': self.pool._used_connections,
'idle': self.pool._idle_cache,
'created': self.pool._created_connections,
'max_connections': self.pool._maxconnections
}
return stats
def get_mysql_status(self):
"""获取MySQL状态"""
conn = self.pool.connection()
try:
cursor = conn.cursor()
cursor.execute("SHOW STATUS LIKE 'Threads_connected'")
threads_connected = cursor.fetchone()[1]
cursor.execute("SHOW STATUS LIKE 'Threads_running'")
threads_running = cursor.fetchone()[1]
cursor.execute("SHOW STATUS LIKE 'Questions'")
questions = cursor.fetchone()[1]
return {
'threads_connected': threads_connected,
'threads_running': threads_running,
'questions': questions
}
finally:
self.pool.putconn(conn)
def check_health(self):
"""检查连接池健康状态"""
stats = self.get_mysql_status()
# 如果当前连接数超过最大连接数的80%，发出警告
if stats['threads_connected'] > 1600:
print(f"警告：MySQL连接数过高！{stats['threads_connected']}/2000")
# 如果正在执行的线程数过多，说明有慢查询
if stats['threads_running'] > 50:
print(f"警告：MySQL活跃线程过多！{stats['threads_running']}")
七、完整架构：四层防护如何协同工作
┌─────────────────────────┐
│ 负载均衡 (Nginx) │
└────────────┬────────────┘
│
┌────────────▼────────────┐
│ Web服务器集群 │
└────────────┬────────────┘
│
┌──────────────────┼──────────────────┐
│ │ │
┌─────────▼────────┐ ┌──────▼──────┐ ┌────────▼────────┐
│ Redis集群 │ │ 消息队列 │ │ 限流/熔断器 │
│ - 商品详情缓存 │ │ - 订单异步 │ │ - 令牌桶限流 │
│ - 库存扣减 │ │ - 秒杀记录 │ │ - 降级开关 │
└─────────┬────────┘ └──────┬──────┘ └────────┬────────┘
│ │ │
┌─────────▼────────────────▼──────────────────▼────────┐
│ 读写分离 + 分库分表 │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ 主库 (写) │◄──►│ 从库 (读) │ │
│ │ - 下单 │ │ - 查询 │ │
│ │ - 扣库存 │ │ - 详情 │ │
│ └─────────────┘ └─────────────┘ │
└──────────────────────────────────────────────────────┘
请求流程示例
class SeckillService:
def __init__(self, cache_service, order_service, stock_service):
self.cache = cache_service
self.order = order_service
self.stock = stock_service
def seckill(self, user_id, product_id, quantity=1):
# 第一层：限流
if not self.rate_limiter.allow(user_id):
return {"success": False, "message": "请求过于频繁，请稍后再试"}
# 第二层：缓存查询商品
product = self.cache.get_product(product_id)
if not product:
return {"success": False, "message": "商品不存在"}
if product['status'] != 'active':
return {"success": False, "message": "秒杀活动未开始或已结束"}
# 第三层：Redis扣减库存（原子操作）
stock_result = self.stock.deduct_stock(product_id, quantity)
if stock_result == 0:
return {"success": False, "message": "库存不足"}
if stock_result == -1:
return {"success": False, "message": "秒杀未开始"}
# 第四层：写入订单（异步）
order_id = self.order.create_order({
'user_id': user_id,
'product_id': product_id,
'quantity': quantity,
'total_price': product['price'] * quantity
})
# 写入消息队列，异步处理后续流程
self.message_queue.send({
'type': 'order_created',
'order_id': order_id,
'user_id': user_id,
'product_id': product_id
})
return {"success": True, "order_id": order_id}
八、实战数据：优化前后的对比
改造完成后，我们做了一轮压力测试：
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| 最大并发连接数 | 500（直接崩溃） | 2000+（稳定运行） | - |
| 平均响应时间 | 3-5秒（超时） | 50-100ms | 98%↓ |
| QPS峰值 | 800 | 8000+ | 10倍↑ |
| MySQL CPU使用率 | 100%（崩溃） | 40-60% | 大幅降低 |
| 成功率 | 30% | 99.9% | - |
关键数字说明：
优化前：
- 5万QPS请求涌入，单库MySQL只能处理约800 QPS
- 其余请求排队，超时，崩溃
- 用户看到：页面转圈，然后报错
优化后：
- 5万QPS请求，Redis处理了4.5万（90%的读请求）
- 剩下的5000 QPS写入，通过分库分表分散到8个库
- 每个库实际只承担约600 QPS，完全在承受范围内
九、我们踩过的坑
坑一：缓存和数据库数据不一致
# 错误做法：先更新数据库，再删除缓存
# 如果删除缓存失败，数据库已经是新值，缓存是旧值
def update_product_wrong(product_id, new_price):
# 1. 更新数据库
db.update_product(product_id, new_price)
# 2. 删除缓存
redis.delete(f"product:{product_id}")
# 如果第2步失败，数据就不一致了
# 正确做法：先删除缓存，再更新数据库
# 或者使用延迟双删
def update_product_correct(product_id, new_price):
# 1. 先删除缓存
redis.delete(f"product:{product_id}")
# 2. 更新数据库
db.update_product(product_id, new_price)
# 3. 延迟再删一次缓存（防止并发问题）
time.sleep(0.5)
redis.delete(f"product:{product_id}")
坑二：分库分表后跨库查询
问题：订单表和订单详情表关联查询，但它们在分库之后可能不在同一个库
解决方案：
1. 避免跨库JOIN，拆分成多次查询
2. 使用ES做全文检索和复杂查询
3. 数据冗余，把需要的字段直接存到订单表
坑三：连接池配置不当
问题：连接池配置了500个连接，但MySQL max_connections只有200
结果：连接池创建了大量连接，但MySQL拒绝连接，全部超时
解决方案：
1. 连接池最大连接数 <= MySQL max_connections / 应用实例数
2. 例如：MySQL max_connections=2000，有4个应用实例
每个实例的连接池最大连接数 = 2000 / 4 = 500
坑四：Redis缓存雪崩
问题：大量缓存key同时过期，请求全部打到数据库
解决方案：
1. 过期时间加随机值（避免同时过期）
2. 使用互斥锁，只有一个线程重建缓存
3. 热点数据永不过期（定期刷新）
十、给还在用单库MySQL的同行们几句心里话
说实话，我们团队在那个崩溃的夜晚之后，真的反思了很久。之前总觉得”数据量不大，单库够用了”，结果一次大促就把我们打回原形。
几个真心建议：
别等到出问题再优化。架构升级是有成本的，越早做成本越低。我们在大促前一周才开始重构，整个团队连续熬了三个通宵，效率低、bug多、身心俱疲。如果提前规划，完全可以从容很多。
缓存是第一道防线。能放缓存的尽量放缓存，尤其是商品详情、秒杀库存这些热点数据。Redis的内存读取速度是微秒级，MySQL是毫秒级，差了一个数量级。
读写分离是最基本的优化。只要做一点改动，性能就能翻倍。我们当时把读操作全部切到从库，QPS直接翻了3倍。
分库分表要提前规划。分库分表一旦上线，后期扩容和数据迁移成本很高。我们在设计时就预留了8个库、16张表的容量，后续扩容很顺畅。
连接池不是越大越好。连接池太大反而会增加MySQL的负担。要根据MySQL的最大连接数和应用实例数合理配置。
最后想说，那次崩溃虽然痛苦，但也让我们彻底搞明白了高并发架构的核心逻辑。现在回头看，秒杀系统的优化本质上就是一场”把压力从数据库往外赶”的游戏：
用户请求 → 限流 → Redis缓存 → 读写分离 → 分库分表 → MySQL
↑
能挡多少挡多少
每一层都在帮下一层减负，最终MySQL只需要处理最核心的写操作。这种层层防御的思路，其实适用于很多高并发场景，不只是秒杀。
希望这篇总结能对正在为数据库性能发愁的朋友们有些帮助。如果你们也遇到过类似的问题，欢迎交流，我们一起想办法。