正文

MySQL高并发处理策略电商大促与支付场景百万级请求下的数据库性能优化实战方案

/0 浏览量
0908

MySQL高并发处理策略电商大促与大促与支付场景百万级请求下的数据库性能优化实战方案

去年双11，我们团队的MySQL数据库扛不住流量，从凌晨三点开始报错率飙升，那一刻我才真正意识到”高并发”四个字背后藏着多少血泪。今天把我这几年踩过的坑、调过的参数、重构过的架构，毫无保留地分享给你。

先说说问题有多严重

做电商和支付系统，最怕的就是促销节点。我们平台曾经在大促期间，QPS从平时的5000一下飙到50万，峰值持续时间长达6小时。那会儿我们的数据库CPU直接打满，连接池爆掉，用户付款页面卡在支付按钮上转圈，客服电话被打爆。

这种场景下，MySQL单库单表根本扛不住，必须有一整套组合拳。

第一道防线：缓存层绝对不能省

很多人一上来就想优化SQL，其实真正该先优化的是缓存架构。在MySQL前面加一层缓存，能挡住80%以上的读请求。

# 用Redis做多级缓存的典型架构
import redis
import json
import hashlib
from datetime import datetime

class CacheLayer:
    def __init__(self):
        # 本地缓存，毫秒级响应
        self.local_cache = {}  
        # Redis缓存，集群模式
        self.redis_client = redis.RedisCluster(
            host='10.0.0.100',
            port=7000,
            password='your_password',
            max_connections=500,
            decode_responses=True
        )
        # TTL设置，避免热点数据永不过期
        self.TTL = 300  # 5分钟
        
    def get_product(self, product_id: str) -> dict:
        # 第一级：本地缓存
        if product_id in self.local_cache:
            cached = self.local_cache[product_id]
            if datetime.now() < cached['expire_time']:
                return cached['data']
        
        # 第二级：Redis缓存
        cache_key = f"product:{product_id}"
        data = self.redis_client.get(cache_key)
        if data:
            parsed = json.loads(data)
            # 回填本地缓存，设置较短TTL
            self.local_cache[product_id] = {
                'data': parsed,
                'expire_time': datetime.now() + timedelta(seconds=30)
            }
            return parsed
        
        # 第三级：MySQL数据库
        result = self.query_db(product_id)
        if result:
            # 写入Redis，带过期时间，防止缓存穿透
            self.redis_client.setex(
                cache_key,
                self.TTL,
                json.dumps(result)
            )
            return result
        
        # 缓存空值，防止缓存穿透
        null_key = f"product:{product_id}:null"
        self.redis_client.setex(null_key, 60, 'null')
        return None

这里有个关键细节：缓存穿透和缓存击穿要分开处理。很多团队只做了缓存，没考虑这两种极端情况，结果大促一来，大量恶意请求或热点商品过期瞬间，直接打穿到MySQL，雪崩就开始了。

读写分离不是简单的复制配置

读写分离是基础操作，但很多人配错了。我们之前就是把主库的binlog同步延迟当儿戏，结果促销时用户刚下单，查订单信息却读到了旧数据。

import pymysql
from pymysql.cursors import DictCursor
import time

class DatabaseRouter:
    def __init__(self):
        # 主库连接池 - 写操作专用
        self.write_pool = pymysql.connections.Connection(
            host='10.0.0.10',
            port=3306,
            user='app_writer',
            password='your_password',
            database='ecommerce',
            max_connections=100,
            connect_timeout=3,
            read_timeout=5,
            write_timeout=5,
            cursorclass=DictCursor
        )
        
        # 从库连接池 - 读操作专用，多实例负载均衡
        self.read_pools = [
            pymysql.connections.Connection(
                host=f'10.0.0.{20+i}',
                port=3306,
                user='app_reader',
                password='your_password',
                database='ecommerce',
                max_connections=200,
                connect_timeout=3,
                cursorclass=DictCursor
            )
            for i in range(3)  # 3个从库
        ]
        self.read_index = 0
        
    def write(self, sql: str, params: tuple = None):
        """写操作走主库"""
        conn = self.write_pool
        try:
            with conn.cursor() as cursor:
                cursor.execute(sql, params or ())
                conn.commit()
                return cursor.lastrowid
        except Exception as e:
            conn.rollback()
            # 写失败要告警，不是简单吞掉
            self.send_alert(f"写操作失败: {e}")
            raise
    
    def read(self, sql: str, params: tuple = None, need_strong_consistency: bool = False):
        """读操作走从库，但支付相关需要强一致性时走主库"""
        if need_strong_consistency:
            conn = self.write_pool
        else:
            # 轮询从库，避免单个从库被打爆
            conn = self.read_pools[self.read_index % len(self.read_pools)]
            self.read_index += 1
            
        try:
            with conn.cursor() as cursor:
                cursor.execute(sql, params or ())
                return cursor.fetchall()
        except Exception as e:
            # 从库故障自动降级到主库
            if not need_strong_consistency:
                return self.read(sql, params, need_strong_consistency=True)
            raise

注意代码里的need_strong_consistency参数。支付查询、订单状态变更这类操作，必须走主库，宁可慢一点也不能读错数据。普通商品浏览、历史记录查询这些可以容忍延迟的，才走从库。

分库分表：电商系统的架构基石

单库单表到百万级并发就是天花板了。我们的订单表在高峰期一天产生300万条数据，一个月就是近1亿条，再不分库分表就等着被拖死吧。

-- 分库策略：按用户ID哈希分16个库
CREATE DATABASE order_db_00;
CREATE DATABASE order_db_01;
...
CREATE DATABASE order_db_15;

-- 分表策略：每个库内按订单ID哈希分32张表
-- 以order_db_00为例
USE order_db_00;

CREATE TABLE orders_00 (
    order_id BIGINT NOT NULL COMMENT '订单ID',
    user_id INT NOT NULL COMMENT '用户ID',
    product_id INT NOT NULL COMMENT '商品ID',
    amount DECIMAL(10,2) NOT NULL COMMENT '订单金额',
    status TINYINT NOT NULL DEFAULT 0 COMMENT '订单状态',
    pay_time DATETIME COMMENT '支付时间',
    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    PRIMARY KEY (order_id, id),
    INDEX idx_user_id (user_id),
    INDEX idx_status (status),
    INDEX idx_create_time (create_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 分表01到31同样结构...

# 路由逻辑：计算数据应该落在哪个库哪张表
class ShardingRouter:
    def __init__(self, db_count=16, table_count=32):
        self.db_count = db_count
        self.table_count = table_count
        
    def get_shard(self, user_id: int, order_id: int) -> tuple:
        """根据用户ID确定库，根据订单ID确定表"""
        db_index = user_id % self.db_count
        table_index = order_id % self.table_count
        return (
            f"order_db_{db_index:02d}",
            f"orders_{table_index:02d}"
        )
    
    def get_route_info(self, entity_id: int, entity_type: str) -> dict:
        """统一路由入口"""
        if entity_type == 'user':
            db_index = entity_id % self.db_count
            return {"db": f"order_db_{db_index:02d}", "table": None}
        elif entity_type == 'order':
            db_index = (entity_id // self.table_count) % self.db_count
            table_index = entity_id % self.table_count
            return {
                "db": f"order_db_{db_index:02d}",
                "table": f"orders_{table_index:02d}"
            }

分库分表之后，跨库查询就成了难题。我们的解决思路是：能避免跨库查询就避免，必须跨库的就用异步补偿。比如用户查订单列表，先查Redis缓存，缓存没有再分片查询然后合并，合并过程异步进行，不阻塞主流程。

支付场景的特殊处理

支付是电商系统最敏感的部分，对一致性和性能的要求近乎苛刻。这里分享几个我们实际用过的方案。

1. 支付状态机设计

-- 支付状态机表
CREATE TABLE payment_status_log (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    payment_id BIGINT NOT NULL COMMENT '支付单ID',
    from_status VARCHAR(20) NOT NULL COMMENT '原状态',
    to_status VARCHAR(20) NOT NULL COMMENT '新状态',
    operator VARCHAR(50) NOT NULL COMMENT '操作者',
    operator_id BIGINT COMMENT '操作者ID',
    remark VARCHAR(255) COMMENT '备注',
    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_payment_id (payment_id),
    INDEX idx_create_time (create_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 支付单表
CREATE TABLE payment_order (
    payment_id BIGINT NOT NULL COMMENT '支付单ID',
    order_id BIGINT NOT NULL COMMENT '关联订单ID',
    user_id INT NOT NULL COMMENT '用户ID',
    amount DECIMAL(10,2) NOT NULL COMMENT '支付金额',
    currency VARCHAR(3) NOT NULL DEFAULT 'CNY',
    status TINYINT NOT NULL DEFAULT 0 COMMENT '0:待支付 1:支付中 2:支付成功 3:支付失败 4:已关闭',
    pay_channel VARCHAR(50) COMMENT '支付渠道',
    transaction_no VARCHAR(100) COMMENT '第三方交易号',
    success_time DATETIME COMMENT '支付成功时间',
    expire_time DATETIME NOT NULL COMMENT '支付过期时间',
    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    UNIQUE KEY uk_payment_id (payment_id),
    UNIQUE KEY uk_order_id (order_id),
    INDEX idx_user_id (user_id),
    INDEX idx_status (status),
    INDEX idx_expire_time (expire_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

2. 分布式锁防重复支付

import redis
import uuid
import json

class PaymentDeduplication:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.lock_timeout = 10  # 锁持有时间，秒
        self.dedup_ttl = 300    # 防重标记过期时间，5分钟
        
    def try_acquire(self, payment_id: str, user_id: int) -> bool:
        """尝试获取支付锁，返回是否成功"""
        lock_key = f"payment_lock:{payment_id}"
        lock_value = f"{user_id}:{uuid.uuid4()}"
        
        # SET NX PX 原子操作
        result = self.redis.set(
            lock_key,
            lock_value,
            nx=True,
            px=self.lock_timeout * 1000
        )
        
        if result:
            # 设置防重标记，防止相同请求重复进入
            dedup_key = f"payment_dedup:{payment_id}"
            self.redis.setex(dedup_key, self.dedup_ttl, '1')
            return True
        return False
    
    def release(self, payment_id: str, user_id: int, lock_value: str):
        """释放支付锁"""
        lock_key = f"payment_lock:{payment_id}"
        # 只有持有锁的人才能释放
        current = self.redis.get(lock_key)
        if current == lock_value.encode():
            self.redis.delete(lock_key)

3. 支付超时自动关单

import schedule
import pymysql
from datetime import datetime, timedelta

def auto_close_expired_payments():
    """定时任务：关闭超时未支付的订单"""
    expire_time = datetime.now()
    
    conn = get_write_connection()
    try:
        with conn.cursor() as cursor:
            # 查询已过期但未支付的订单
            sql = """
                SELECT payment_id, order_id, user_id, amount
                FROM payment_order
                WHERE status = 0
                AND expire_time < %s
                LIMIT 100
            """
            cursor.execute(sql, (expire_time,))
            expired_payments = cursor.fetchall()
            
            for payment in expired_payments:
                # 1. 更新支付状态为已关闭
                cursor.execute("""
                    UPDATE payment_order
                    SET status = 4,
                        update_time = NOW()
                    WHERE payment_id = %s
                    AND status = 0
                """, (payment['payment_id'],))
                
                # 2. 更新订单状态为已取消
                cursor.execute("""
                    UPDATE orders
                    SET status = 3,  -- 3表示已取消
                        update_time = NOW()
                    WHERE order_id = %s
                    AND status = 1  -- 只更新待支付状态的订单
                """, (payment['order_id'],))
                
                # 3. 记录日志
                cursor.execute("""
                    INSERT INTO payment_status_log
                    (payment_id, from_status, to_status, operator, remark)
                    VALUES (%s, 'pending', 'closed', 'SYSTEM', '支付超时自动关闭')
                """, (payment['payment_id'],))
                
                # 4. 发送消息给队列，通知下游系统
                send_message_to_queue('payment_expired', {
                    'payment_id': payment['payment_id'],
                    'order_id': payment['order_id'],
                    'user_id': payment['user_id'],
                    'amount': payment['amount']
                })
            
            conn.commit()
            
    except Exception as e:
        conn.rollback()
        send_alert(f"自动关单任务失败: {e}")
    finally:
        conn.close()

# 每分钟执行一次
schedule.every().minute.do(auto_close_expired_payments)

连接池的精细化调优

很多团队连接池配的是默认值，这在平时可能还行，大促一来直接被打爆。我们的优化经验是：

# 连接池配置调优
POOL_CONFIG = {
    'max_connections': 500,        # 根据MySQL max_connections调整
    'min_connections': 50,         # 预热连接数
    'max_idle_time': 3600,         # 连接最大空闲时间
    'max_lifetime': 7200,          # 连接最大生命周期
    'connection_timeout': 3,       # 获取连接超时
    'idle_timeout': 600,           # 空闲连接回收时间
    'max_overflow': 100,           # 超出池大小的临时连接数
    'pool_recycle': 1800,          # 连接回收时间，防止MySQL这边超时断开
}

# 分场景配置不同连接池
POOL_CONFIGS = {
    'read': {
        'max_connections': 300,
        'min_connections': 100,
        'connection_timeout': 5,
    },
    'write': {
        'max_connections': 100,
        'min_connections': 30,
        'connection_timeout': 3,
    },
    'payment': {
        'max_connections': 50,
        'min_connections': 20,
        'connection_timeout': 2,
        # 支付操作单独一个池，避免被其他操作拖死
    }
}

关键参数解释一下：

  • max_connections：不是越大越好，要配合MySQL的max_connections和服务器内存
  • min_connections：预热连接数，大促前预先建立连接，避免临时创建连接的性能损耗
  • max_overflow：允许临时超出的连接数，但设置不宜过大，防止连接数爆炸
  • pool_recycle：一定要设置，否则MySQL服务端可能已经断开了连接而池子里还不知道

SQL层面能做的优化

架构优化是基础，但SQL本身写得烂，什么架构都救不了。

1. 避免深分页

-- 错误的深分页写法（性能极差）
SELECT * FROM orders WHERE user_id = 12345 ORDER BY create_time DESC LIMIT 100000, 20;

-- 优化方案1：延迟关联
SELECT o.* FROM orders o
INNER JOIN (
    SELECT order_id FROM orders 
    WHERE user_id = 12345 
    ORDER BY create_time DESC 
    LIMIT 100000, 20
) AS tmp ON o.order_id = tmp.order_id;

-- 优化方案2：游标分页（推荐）
SELECT * FROM orders 
WHERE user_id = 12345 
AND create_time < '2024-11-10 23:59:59'  -- 上一页最后一条记录的时间
ORDER BY create_time DESC 
LIMIT 20;

2. 覆盖索引减少回表

-- 给高频查询加覆盖索引
-- 场景：根据用户ID和订单状态查询订单列表
ALTER TABLE orders ADD INDEX idx_user_status_create (user_id, status, create_time);

-- 这样查询只需要扫描索引就能返回所有需要的字段，不需要回表
SELECT order_id, amount, status, create_time 
FROM orders 
WHERE user_id = 12345 AND status = 1 
ORDER BY create_time DESC;

3. 大事务拆小

# 支付成功后的处理，切忌用一个事务完成所有操作
def handle_payment_success(payment_id: int):
    # 错误做法：一个大事务包含所有后续操作
    # with transaction.atomic():
    #     update_payment_status(...)
    #     update_order_status(...)
    #     deduct_inventory(...)
    #     add_score(...)
    #     send_notification(...)
    
    # 正确做法：分步处理，每步独立事务或异步
    # 第一步：更新支付状态（核心操作，必须事务保证）
    update_payment_status(payment_id)
    
    # 第二步：更新订单状态（异步，允许短暂延迟）
    async_update_order_status(payment_id)
    
    # 第三步：扣减库存（异步，允许短暂延迟）
    async_deduct_inventory(payment_id)
    
    # 第四步：发送通知（完全异步）
    async_send_notification(payment_id)

大事务不仅占用锁时间长，还会阻塞其他操作，在大促期间尤其危险。能异步的异步，能最终一致的不要强一致。

压测和监控：没有数据就没有发言权

优化之前一定要压测，优化之后要持续监控。我们团队在大促前都会做一次全链路压测。

# 使用Locust进行MySQL压力测试
from locust import HttpUser, task, between
import mysql.connector
import random

class MySQLPerformanceTest(HttpUser):
    wait_time = between(0.1, 0.5)
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.db = mysql.connector.connect(
            host='10.0.0.10',
            user='test_user',
            password='test_password',
            database='ecommerce_test'
        )
        
    @task(3)
    def query_product(self):
        """模拟商品查询"""
        product_id = random.randint(1, 100000)
        cursor = self.db.cursor()
        cursor.execute(
            "SELECT * FROM products WHERE id = %s", 
            (product_id,)
        )
        cursor.fetchone()
        cursor.close()
        
    @task(2)
    def query_order(self):
        """模拟订单查询"""
        user_id = random.randint(1, 50000)
        cursor = self.db.cursor()
        cursor.execute(
            "SELECT * FROM orders WHERE user_id = %s ORDER BY create_time DESC LIMIT 10",
            (user_id,)
        )
        cursor.fetchall()
        cursor.close()
        
    @task(1)
    def create_order(self):
        """模拟下单"""
        user_id = random.randint(1, 50000)
        product_id = random.randint(1, 100000)
        cursor = self.db.cursor()
        cursor.execute(
            "INSERT INTO orders (user_id, product_id, amount, status) VALUES (%s, %s, %s, 0)",
            (user_id, product_id, round(random.uniform(10, 1000), 2))
        )
        self.db.commit()
        cursor.close()

监控方面，重点关注这几个指标：

  • QPS/TPS：每秒查询/事务数
  • 连接数：当前活跃连接数和最大连接数
  • 慢查询数：超过设定阈值的查询数量
  • InnoDB缓冲池命中率：应该保持在95%以上
  • 锁等待时间：出现锁等待要立即告警
  • 主从延迟：超过1秒就要警惕

大促前的检查清单

每年大促前，我们会对照这份清单逐项检查：

  1. 架构层面

    • [ ] 缓存层是否已预热热点数据
    • [ ] 读写分离是否正常，从库延迟是否在可控范围
    • [ ] 分库分表路由是否正确
    • [ ] 连接池配置是否已调整到最大容量

  2. 数据库层面

    • [ ] 慢查询是否已优化或迁移
    • [ ] 索引是否经过EXPLAIN验证
    • [ ] 表统计信息是否已更新
    • [ ] 大表是否已做归档或分区

  3. 监控层面

    • [ ] 告警规则是否已配置并测试
    • [ ] 监控面板是否已搭建
    • [ ] 值班人员是否已到位

  4. 应急层面

    • [ ] 降级方案是否已制定
    • [ ] 回滚方案是否已准备
    • [ ] 应急预案是否已演练

最后说几句

MySQL高并发优化不是靠一个参数、一条SQL就能解决的，它需要从缓存架构、数据库设计、连接管理、SQL优化、监控告警等多个层面综合考虑。我们这套方案是从无数次故障和加班中总结出来的，每一个决策背后都有血泪教训。

如果你正在为高并发数据库问题头疼，记住一个原则：能挡在数据库前面的操作，全部挡在前面。缓存、消息队列、异步处理、读写分离，层层设防，让数据库只做它最擅长的事——可靠地存储和检索数据。

-- 展开阅读全文 --