MySQL高并发处理策略：如何避免数据库崩溃并提升系统性能的实战指南

引言：理解高并发场景下的MySQL挑战

在现代互联网应用中，高并发访问已经成为常态。无论是电商平台的秒杀活动、社交媒体的热点事件，还是金融系统的交易高峰，MySQL数据库都面临着前所未有的压力。高并发场景下，数据库容易出现连接耗尽、响应延迟飙升、甚至服务崩溃等问题。本文将深入探讨MySQL高并发处理的核心策略，从架构设计到参数调优，从代码实现到监控预警，提供一套完整的实战指南，帮助您构建稳定高效的数据库系统。

一、高并发对MySQL的核心影响

1.1 连接层压力

当并发请求激增时，MySQL的连接数会快速上升。默认的最大连接数（max_connections=151）在高并发场景下很快就会被耗尽，导致新连接被拒绝。同时，大量的连接会消耗服务器内存，每个连接都需要分配thread_stack（默认256KB）和read_buffer等内存结构。

1.2 锁竞争加剧

InnoDB引擎的行锁、间隙锁在高并发写操作下会产生激烈的锁竞争。特别是热点数据更新时，大量事务会排队等待锁释放，导致响应时间线性增长。严重的锁竞争还会引发死锁，增加事务回滚率。

1.3 I/O瓶颈

高并发下的大量读写操作会迅速耗尽磁盘I/O带宽。特别是未优化的查询，如全表扫描、大量随机读等，会使磁盘I/O成为系统瓶颈。同时，redo log、binlog的频繁写入也会加剧I/O压力。

1.4 CPU和内存资源争用

复杂的查询、排序、分组操作会消耗大量CPU资源。高并发下的临时表、排序缓冲区等内存结构会频繁申请释放，增加CPU的上下文切换开销。内存不足时还会触发swap，导致性能急剧下降。

二、连接层优化策略

2.1 合理配置连接数参数

max_connections：这是MySQL允许的最大并发连接数。设置过高会耗尽系统内存，过低则会拒绝服务。建议根据服务器内存和业务特点计算：

理论最大连接数 = 可用内存 / (每个连接的内存开销)
每个连接的内存开销 ≈ 1MB (包括thread_stack、read_buffer等)

例如，32GB内存的服务器，建议设置max_connections=2000-3000。但实际还需要考虑其他内存消耗。

关键配置示例：

[mysqld]
max_connections = 2000
thread_cache_size = 100  # 线程缓存，减少线程创建销毁开销
back_log = 500  # TCP连接队列长度，应对瞬时高峰

2.2 连接池技术应用

应用层必须使用连接池，避免频繁创建销毁连接。主流连接池配置要点：

HikariCP（Java）配置示例：

HikariConfig config = new HikariConfig();
config.setJdbcUrl("jdbc:mysql://localhost:3306/mydb");
config.setUsername("user");
config.setPassword("password");
config.setMaximumPoolSize(100);  // 池大小，根据业务调整
config.setMinimumIdle(10);       // 最小空闲连接
config.setConnectionTimeout(30000); // 获取连接超时时间
config.setIdleTimeout(600000);    // 空闲连接超时
config.setMaxLifetime(1800000);   // 连接最大存活时间
config.setLeakDetectionThreshold(60000); // 连接泄漏检测
config.addDataSourceProperty("cachePrepStmts", "true");
config.addDataSourceProperty("prepStmtCacheSize", "250");
config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
HikariDataSource dataSource = new HikariDataSource(config);

连接池大小计算公式：

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

对于读密集型应用可以适当增大，写密集型应保守设置。

2.3 读写分离架构

通过主从复制实现读写分离，将读请求分发到从库，减轻主库压力：

// 简单的读写分离数据源路由
public class DataSourceRouter extends AbstractRoutingDataSource {
    private static final ThreadLocal<String> CONTEXT_HOLDER = new ThreadLocal<>();
    
    @Override
    protected Object determineCurrentLookupKey() {
        return CONTEXT_HOLDER.get();
    }
    
    public static void setMaster() {
        CONTEXT_HOLDER.set("master");
    }
    
    public static void setSlave() {
        CONTEXT_HOLDER.set("slave");
    }
    
    public static void clear() {
        CONTEXT_HOLDER.remove();
    }
}

// 使用AOP进行路由
@Aspect
@Component
public class DataSourceAspect {
    @Before("@annotation(master)")
    public void setMaster(JoinPoint jp, Master master) {
        DataSourceRouter.setMaster();
    }
    
    @Before("@annotation(slave)")
    public void setSlave(JoinPoint jp, Slave slave) {
        DataSourceRouter.setSlave();
    }
    
    @After("@annotation(master) || @annotation(slave)")
    public void clear(JoinPoint jp) {
        DataSourceRouter.clear();
    }
}

三、SQL与索引优化

3.1 执行计划分析

使用EXPLAIN分析SQL执行计划，重点关注type、key、rows、Extra字段：

-- 示例：分析一个慢查询
EXPLAIN SELECT o.order_id, c.name, o.amount 
FROM orders o 
JOIN customers c ON o.customer_id = c.id 
WHERE o.create_time >= '2024-01-01' 
AND o.status = 'PAID' 
ORDER BY o.amount DESC 
LIMIT 100;

-- 关键指标解读：
-- type: ALL（全表扫描）→ 需要优化
-- key: NULL（未使用索引）→ 需要添加索引
-- rows: 1000000（扫描行数过多）→ 索引选择性差
-- Extra: Using filesort（文件排序）→ 需要优化索引或查询

3.2 索引设计原则

最左前缀原则：复合索引必须遵循最左前缀匹配，否则索引失效。

-- 创建复合索引
ALTER TABLE orders ADD INDEX idx_customer_status_time (customer_id, status, create_time);

-- 有效使用索引的查询：
SELECT * FROM orders WHERE customer_id = 123; -- ✅ 使用索引第一列
SELECT * FROM orders WHERE customer_id = 123 AND status = 'PAID'; -- ✅ 使用前两列
SELECT * FROM orders WHERE customer_id = 123 AND status = 'PAID' AND create_time >= '2024-01-01'; -- ✅ 完全匹配

-- 无效使用索引的查询：
SELECT * FROM orders WHERE status = 'PAID'; -- ❌ 未使用最左列
SELECT * FROM orders WHERE create_time >= '2024-01-01'; -- ❌ 未使用最左列

索引选择性：选择性高的列适合建索引。选择性 = 不重复值数量 / 总行数。

-- 计算列的选择性
SELECT COUNT(DISTINCT status) / COUNT(*) AS selectivity FROM orders;
-- 结果 > 0.3 适合建索引，< 0.1 不适合

3.3 避免索引失效的常见场景

函数操作：WHERE YEAR(create_time) = 2024 → 改为 create_time BETWEEN ‘2024-01-01’ AND ‘2024-12-31’
隐式类型转换：WHERE phone = 13800138000 (phone是varchar) → 改为 WHERE phone = ‘13800138000’
LIKE前缀模糊查询：WHERE name LIKE ‘%张%’ → 改为 WHERE name LIKE ‘张%’
OR条件：WHERE id = 1 OR id = 2 → 改为 WHERE id IN (1,2)
负向查询：!=、NOT IN、NOT LIKE 通常无法使用索引

3.4 覆盖索引与延迟关联

覆盖索引：查询的列全部在索引中，避免回表操作。

-- 原始查询（需要回表）
SELECT order_id, customer_id, amount, status FROM orders WHERE customer_id = 123;

-- 创建覆盖索引
ALTER TABLE orders ADD INDEX idx_customer_cover (customer_id, order_id, amount, status);

-- 优化后查询（无需回表）
SELECT order_id, customer_id, amount, status FROM orders WHERE customer_id = 123;

延迟关联：对于需要返回大量列的查询，先通过索引获取主键，再关联获取其他列。

-- 低效查询
SELECT * FROM orders WHERE customer_id = 123 AND amount > 1000;

-- 优化：延迟关联
SELECT t1.* 
FROM orders t1 
JOIN (SELECT order_id FROM orders WHERE customer_id = 123 AND amount > 1000) t2 
ON t1.order_id = t2.order_id;

四、事务与锁优化

4.1 事务设计原则

短事务原则：事务应尽可能短，减少锁持有时间。

// ❌ 错误示例：长事务
@Transactional
public void processOrder(Long orderId) {
    // 1. 查询订单（获取共享锁）
    Order order = orderMapper.selectById(orderId);
    
    // 2. 调用外部API（耗时操作，事务未提交，锁未释放）
    paymentService.verifyPayment(order);
    
    // 3. 更新状态
    order.setStatus("PAID");
    orderMapper.update(order);
}

// ✅ 正确示例：短事务
public void processOrder(Long orderId) {
    // 1. 事务外执行耗时操作
    paymentService.verifyPayment(orderId);
    
    // 2. 缩短事务范围
    try {
        transactionTemplate.execute(status -> {
            Order order = orderMapper.selectById(orderId);
            order.setStatus("PAID");
            orderMapper.update(order);
            return null;
        });
    } catch (Exception e) {
        // 处理异常
    }
}

4.2 乐观锁与悲观锁选择

乐观锁：适合读多写少场景，通过版本号避免更新丢失。

-- 表结构
CREATE TABLE product (
    id BIGINT PRIMARY KEY,
    name VARCHAR(100),
    stock INT,
    version INT DEFAULT 0
);

-- 更新操作
UPDATE product 
SET stock = stock - 1, version = version + 1 
WHERE id = 123 AND version = 2; -- 版本号匹配

-- Java代码实现
public boolean deductStock(Long productId, int quantity) {
    int retry = 3;
    while (retry-- > 0) {
        Product product = productMapper.selectById(productId);
        int newVersion = product.getVersion() + 1;
        int affected = productMapper.updateStock(productId, quantity, product.getVersion(), newVersion);
        if (affected > 0) {
            return true;
        }
        // 短暂休眠后重试
        Thread.sleep(50);
    }
    return false;
}

悲观锁：适合写多读少、强一致性场景。

-- 显式悲观锁
BEGIN;
SELECT * FROM inventory WHERE product_id = 123 FOR UPDATE; -- 排他锁
UPDATE inventory SET stock = stock - 1 WHERE product_id = 123;
COMMIT;

-- 避免间隙锁：使用等值查询
-- ❌ 范围查询会产生间隙锁
SELECT * FROM inventory WHERE product_id BETWEEN 100 AND 200 FOR UPDATE;
-- ✅ 等值查询减少锁范围
SELECT * FROM inventory WHERE product_id = 123 FOR UPDATE;

4.3 死锁预防与处理

死锁常见场景：

交叉更新：A更新1→2，B更新2→1
索引顺序不一致：不同事务按不同顺序访问数据
间隙锁冲突：范围更新导致的死锁

预防策略：

-- 1. 固定访问顺序
-- 事务1和事务2都按相同顺序访问：先orders后order_items

-- 2. 使用相同索引
-- 所有事务都使用主键或相同索引进行更新

-- 3. 减少间隙锁
-- 使用等值查询代替范围查询
-- 降低事务隔离级别（RC级别间隙锁较少）
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;

死锁监控与处理：

-- 查看最近死锁信息
SHOW ENGINE INNODB STATUS\G

-- 在MySQL 8.0+中启用死锁监控
SET GLOBAL innodb_print_all_deadlocks = ON;

-- 应用层重试机制
public <T> T executeWithRetry(Supplier<T> operation) {
    int maxRetries = 3;
    for (int i = 0; i < maxRetries; i++) {
        try {
            return operation.get();
        } catch (DeadlockException e) {
            if (i == maxRetries - 1) throw e;
            // 指数退避
            try {
                Thread.sleep((long) Math.pow(2, i) * 100);
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
            }
        }
    }
    throw new RuntimeException("Max retries exceeded");
}

五、查询缓存与缓冲区优化

5.1 查询缓存（Query Cache）

注意：MySQL 8.0已移除查询缓存功能，但在5.7及以下版本仍有价值。对于读多写少的静态数据，查询缓存能显著提升性能。

-- MySQL 5.7 配置
[mysqld]
query_cache_type = 1  # 0:OFF, 1:ON, 2:DEMAND
query_cache_size = 128M
query_cache_limit = 2M  # 单个查询最大缓存结果集

-- 在SQL中显式使用缓存
SELECT SQL_CACHE * FROM static_table WHERE id = 1;
SELECT SQL_NO_CACHE * FROM dynamic_table WHERE id = 1;

5.2 InnoDB缓冲池（Buffer Pool）

核心参数：

[mysqld]
innodb_buffer_pool_size = 24G  # 70-80%物理内存
innodb_buffer_pool_instances = 8  # 多实例减少竞争
innodb_buffer_pool_dump_at_shutdown = 1  # 关闭时dump热数据
innodb_buffer_pool_load_at_startup = 1    # 启动时加载热数据

监控缓冲池命中率：

-- 缓冲池命中率应 > 99%
SELECT 
    (1 - (SUM(VARIABLE_VALUE) / @@innodb_buffer_pool_size)) * 100 AS hit_rate
FROM performance_schema.global_status 
WHERE VARIABLE_NAME = 'Innodb_buffer_pool_reads';

-- 查看缓冲池使用情况
SHOW ENGINE INNODB STATUS\G
-- 关注 BUFFER POOL AND MEMORY 部分

5.3 其他关键缓冲区

[mysqld]
# 读缓冲
read_buffer_size = 2M
read_rnd_buffer_size = 4M

# 排序缓冲
sort_buffer_size = 4M

# Join缓冲
join_buffer_size = 4M

# 临时表内存大小
tmp_table_size = 64M
max_heap_table_size = 64M

# 线程栈大小（根据连接数调整）
thread_stack = 256K

六、高并发写入优化

6.1 批量操作

批量插入：减少网络往返和事务开销。

// ❌ 单条插入（N次网络往返）
for (Order order : orderList) {
    orderMapper.insert(order);
}

// ✅ 批量插入（1次网络往返）
public void batchInsert(List<Order> orders) {
    // JDBC批量处理
    String sql = "INSERT INTO orders (order_id, customer_id, amount, status) VALUES (?, ?, ?, ?)";
    try (Connection conn = dataSource.getConnection();
         PreparedStatement ps = conn.prepareStatement(sql)) {
        
        int batchSize = 0;
        for (Order order : orders) {
            ps.setLong(1, order.getOrderId());
            ps.setLong(2, order.getCustomerId());
            ps.setBigDecimal(3, order.getAmount());
            ps.setString(4, order.getStatus());
            ps.addBatch();
            
            // 每1000条提交一次
            if (++batchSize % 1000 == 0) {
                ps.executeBatch();
            }
        }
        ps.executeBatch(); // 提交剩余
    } catch (SQLException e) {
        throw new RuntimeException(e);
    }
}

批量更新：

-- 使用INSERT ON DUPLICATE KEY UPDATE
INSERT INTO orders (order_id, customer_id, amount) 
VALUES (1, 100, 500), (2, 101, 300), (3, 102, 800)
ON DUPLICATE KEY UPDATE 
    amount = VALUES(amount),
    update_time = NOW();

-- 使用CASE WHEN批量更新
UPDATE orders 
SET status = CASE 
    WHEN order_id = 1 THEN 'PAID'
    WHEN order_id = 2 THEN 'SHIPPED'
    WHEN order_id = 3 THEN 'CANCELLED'
END
WHERE order_id IN (1, 2, 3);

6.2 异步写入与队列

消息队列解耦：将写操作异步化，削峰填谷。

// 生产者：将写操作放入队列
public void createOrderAsync(Order order) {
    // 1. 快速返回，不等待数据库写入
    order.setStatus("PENDING");
    orderMapper.insert(order); // 先写入状态
    
    // 2. 发送消息到队列
    rabbitTemplate.convertAndSend("order.create", order);
    
    // 3. 异步处理后续逻辑
    // 消费者会处理支付、库存扣减等
}

// 消费者：批量处理队列消息
@RabbitListener(queues = "order.create")
public void processOrderCreate(Order order) {
    // 批量处理：累积一定数量后批量提交
    List<Order> batch = orderBuffer.getAndAdd(order);
    if (batch.size() >= 100) {
        batchInsert(batch);
        orderBuffer.clear();
    }
}

6.3 分区表（Partitioning）

对于大数据量表，使用分区可以提升查询性能和管理效率。

-- 按时间范围分区（适合日志、订单等时间序列数据）
CREATE TABLE orders (
    order_id BIGINT,
    customer_id BIGINT,
    amount DECIMAL(10,2),
    create_time DATETIME,
    status VARCHAR(20),
    PRIMARY KEY (order_id, create_time) -- 分区键必须包含在主键中
) PARTITION BY RANGE (YEAR(create_time)) (
    PARTITION p2023 VALUES LESS THAN (2024),
    PARTITION p2024 VALUES LESS THAN (2025),
    PARTITION p2025 VALUES LESS THAN (2026),
    PARTITION p_future VALUES LESS THAN MAXVALUE
);

-- 按哈希分区（适合均匀分布）
CREATE TABLE user_log (
    id BIGINT,
    user_id BIGINT,
    log_time DATETIME
) PARTITION BY HASH(user_id) PARTITIONS 16;

-- 查看分区信息
SELECT PARTITION_NAME, TABLE_ROWS 
FROM INFORMATION_SCHEMA.PARTITIONS 
WHERE TABLE_NAME = 'orders';

七、读写分离与分库分表

7.1 主从复制配置

主库配置（my.cnf）：

[mysqld]
server-id = 1
log_bin = mysql-bin
binlog_format = ROW  # 行级复制，减少锁
expire_logs_days = 7
sync_binlog = 1      # 每次事务提交都同步磁盘，保证安全
innodb_flush_log_at_trx_commit = 1  # 保证ACID

从库配置：

[mysqld]
server-id = 2
relay_log = mysql-relay-bin
read_only = 1  # 只读模式，防止误写

创建复制用户：

-- 在主库执行
CREATE USER 'repl'@'%' IDENTIFIED BY 'password';
GRANT REPLICATION SLAVE ON *.* TO 'repl'@'%';
FLUSH PRIVILEGES;

-- 在从库执行
CHANGE MASTER TO
MASTER_HOST='master_ip',
MASTER_USER='repl',
MASTER_PASSWORD='password',
MASTER_LOG_FILE='mysql-bin.000001',
MASTER_LOG_POS=1234;

START SLAVE;
SHOW SLAVE STATUS\G  -- 检查Slave_IO_Running和Slave_SQL_Running为Yes

7.2 读写分离实现

ShardingSphere-JDBC配置：

# application.yml
spring:
  shardingsphere:
    datasource:
      names: master, slave0, slave1
      master:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://master:3306/mydb
        username: root
        password: password
      slave0:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://slave0:3306/mydb
        username: root
        password: password
      slave1:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://slave1:3306/mydb
        username: root
        password: password
    
    rules:
      readwrite-splitting:
        data-sources:
          mydb:
            type: Static
            props:
              write-data-source-name: master
              read-data-source-names: slave0,slave1
              load-balance-type: ROUND_ROBIN

7.3 分库分表（Sharding）

垂直分库：按业务模块拆分数据库。

原库：mydb (user, order, product, payment)
拆分后：
- user_db: user, user_profile
- order_db: order, order_item
- product_db: product, category
- payment_db: payment, refund

水平分表：按数据特征拆分大表。

-- 按用户ID取模分表（16张表）
-- order_0, order_1, ..., order_15

-- ShardingSphere配置
spring:
  shardingsphere:
    rules:
      sharding:
        tables:
          orders:
            actual-data-nodes: mydb.order_$->{0..15}
            table-strategy:
              standard:
                sharding-column: customer_id
                sharding-algorithm-name: mod
        sharding-algorithms:
          mod:
            type: MOD
            props:
              sharding-count: 16

八、高并发配置调优

8.1 InnoDB核心参数

[mysqld]
# 1. 缓冲池（最重要）
innodb_buffer_pool_size = 24G  # 70-80%内存
innodb_buffer_pool_instances = 8  # 多实例减少竞争

# 2. 日志文件
innodb_log_file_size = 2G      # 重做日志大小，建议1-2G
innodb_log_buffer_size = 64M   # 日志缓冲区
innodb_flush_log_at_trx_commit = 1  # 1:安全, 2:性能更好但可能丢数据

# 3. I/O相关
innodb_flush_method = O_DIRECT  # 绕过OS缓存，直接I/O
innodb_io_capacity = 2000       # SSD建议2000-5000
innodb_io_capacity_max = 4000   # 最大IOPS

# 4. 并发控制
innodb_thread_concurrency = 0   # 0:自动管理，建议0
innodb_read_io_threads = 8
innodb_write_io_threads = 8

# 5. 锁相关
innodb_lock_wait_timeout = 50   # 锁等待超时
innodb_rollback_on_timeout = 1  # 超时回滚整个事务

8.2 连接与线程参数

[mysqld]
# 连接数
max_connections = 2000
thread_cache_size = 100
back_log = 500

# 超时设置
wait_timeout = 600  # 非交互连接超时，避免连接泄漏
interactive_timeout = 600

# 内存相关
tmp_table_size = 64M
max_heap_table_size = 64M
sort_buffer_size = 4M
join_buffer_size = 4M
read_buffer_size = 2M
read_rnd_buffer_size = 4M

8.3 日志与监控参数

[mysqld]
# 慢查询日志
slow_query_log = 1
slow_query_log_file = /var/log/mysql/slow.log
long_query_time = 1  # 记录超过1秒的查询
log_queries_not_using_indexes = 1

# 错误日志
log_error = /var/log/mysql/error.log

# 通用查询日志（调试用，生产环境关闭）
general_log = 0
general_log_file = /var/log/mysql/general.log

# 性能监控
performance_schema = ON  # MySQL 5.6+ 默认开启

九、高并发场景下的监控与预警

9.1 关键监控指标

连接数监控：

-- 当前连接数 vs 最大连接数
SHOW STATUS LIKE 'Threads_connected';
SHOW VARIABLES LIKE 'max_connections';

-- 计算使用率
SELECT 
    (VARIABLE_VALUE / @@max_connections) * 100 AS usage_rate
FROM performance_schema.global_status 
WHERE VARIABLE_NAME = 'Threads_connected';

QPS/TPS监控：

-- 计算每秒查询数和事务数
SELECT 
    (VARIABLE_VALUE - @last_queries) / TIMESTAMPDIFF(SECOND, @last_time, NOW()) AS qps,
    (VARIABLE_VALUE - @last_committed) / TIMESTAMPDIFF(SECOND, @last_time, NOW()) AS tps
FROM performance_schema.global_status 
WHERE VARIABLE_NAME IN ('Queries', 'Com_commit');

-- 使用Prometheus + Grafana监控
# mysqld_exporter配置
--collect.global_status
--collect.info_schema.innodb_metrics
--collect.perf_schema.tablelocks
--collect.perf_schema.eventsstatements

慢查询监控：

-- 查看最近慢查询
SELECT 
    DIGEST_TEXT,
    COUNT_STAR,
    AVG_TIMER_WAIT/1000000000000 AS avg_time_sec,
    MAX_TIMER_WAIT/1000000000000 AS max_time_sec
FROM performance_schema.events_statements_summary_by_digest
WHERE AVG_TIMER_WAIT > 1000000000000  -- 1秒
ORDER BY AVG_TIMER_WAIT DESC
LIMIT 10;

9.2 锁监控

-- 查看当前锁等待
SELECT 
    r.trx_id waiting_trx_id,
    r.trx_mysql_thread_id waiting_thread,
    r.trx_query waiting_query,
    b.trx_id blocking_trx_id,
    b.trx_mysql_thread_id blocking_thread,
    b.trx_query blocking_query
FROM information_schema.innodb_lock_waits w
INNER JOIN information_schema.innodb_trx b ON b.trx_id = w.blocking_trx_id
INNER JOIN information_schema.innodb_trx r ON r.trx_id = w.requesting_trx_id;

-- 查看MDL锁（元数据锁）
SELECT 
    ps.id AS processlist_id,
    ps.user,
    ps.host,
    ps.db,
    ps.command,
    ps.time,
    ps.state,
    ps.info
FROM performance_schema.metadata_locks ml
JOIN performance_schema.threads t ON ml.OWNER_THREAD_ID = t.THREAD_ID
JOIN information_schema.processlist ps ON t.PROCESSLIST_ID = ps.id
WHERE ml.OBJECT_SCHEMA = 'mydb' AND ml.OBJECT_NAME = 'orders';

9.3 性能监控工具

Percona Toolkit：

# 安装
sudo apt-get install percona-toolkit

# 分析慢查询日志
pt-query-digest /var/log/mysql/slow.log > slow_report.txt

# 监控复制延迟
pt-heartbeat --check h=master,u=root,p=password --master-server-id=1

# 在线DDL工具
pt-online-schema-change --alter "ADD INDEX idx_status (status)" D=mydb,t=orders --execute

Prometheus + mysqld_exporter：

# docker-compose.yml
version: '3'
services:
  mysqld_exporter:
    image: prom/mysqld-exporter
    environment:
      DATA_SOURCE_NAME: "user:password@(mysql:3306)/"
    ports:
      - "9104:9104"
  
  prometheus:
    image: prom/prometheus
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"

十、实战案例：秒杀系统优化

10.1 问题分析

场景：10000 QPS秒杀，库存100件，热点数据更新。

问题：

库存行锁竞争激烈
大量请求导致连接耗尽
数据库CPU飙升至100%

10.2 优化方案

架构层：

// 1. 缓存预热与库存扣减
public class SeckillService {
    private static final String STOCK_KEY = "seckill:stock:";
    private static final String ORDER_KEY = "seckill:order:";
    
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    @Autowired
    private OrderMapper orderMapper;
    
    // 预热：将库存加载到Redis
    public void preloadStock(Long productId, int stock) {
        redisTemplate.opsForValue().set(STOCK_KEY + productId, String.valueOf(stock));
    }
    
    // 秒杀下单
    public Result seckill(Long productId, Long userId) {
        String lockKey = "seckill:lock:" + productId;
        
        // 1. Redis预扣库存（Lua脚本保证原子性）
        String luaScript = 
            "if redis.call('exists', KEYS[1]) == 1 then " +
            "   local stock = tonumber(redis.call('get', KEYS[1])); " +
            "   if stock > 0 then " +
            "       redis.call('decr', KEYS[1]); " +
            "       return 1; " +
            "   end; " +
            "end; " +
            "return 0;";
        
        Long result = redisTemplate.execute(
            new DefaultRedisScript<>(luaScript, Long.class),
            Collections.singletonList(STOCK_KEY + productId)
        );
        
        if (result == 0) {
            return Result.fail("库存不足");
        }
        
        // 2. 异步创建订单
        String orderId = IdUtil.randomUUID();
        rabbitTemplate.convertAndSend("seckill.order", 
            new SeckillMessage(userId, productId, orderId));
        
        return Result.success(orderId);
    }
    
    // 异步消费：批量写入数据库
    @RabbitListener(queues = "seckill.order")
    public void processSeckillOrder(List<SeckillMessage> batch) {
        // 批量插入订单
        List<Order> orders = batch.stream()
            .map(m -> new Order(m.getOrderId(), m.getUserId(), m.getProductId()))
            .collect(Collectors.toList());
        
        orderMapper.batchInsert(orders);
        
        // 批量更新库存（在业务低峰期）
        if (batch.size() >= 100) {
            updateStockBatch(batch);
        }
    }
}

数据库层：

-- 1. 库存表使用乐观锁
CREATE TABLE stock (
    product_id BIGINT PRIMARY KEY,
    stock INT,
    version INT DEFAULT 0
);

-- 2. 订单表分表（按用户ID取模）
CREATE TABLE order_0 (
    order_id VARCHAR(50) PRIMARY KEY,
    user_id BIGINT,
    product_id BIGINT,
    create_time DATETIME
) PARTITION BY HASH(user_id) PARTITIONS 16;

-- 3. 关闭MySQL查询缓存（写多读少）
SET GLOBAL query_cache_size = 0;

配置优化：

[mysqld]
# 针对秒杀场景优化
innodb_buffer_pool_size = 32G
innodb_log_file_size = 4G
innodb_flush_log_at_trx_commit = 2  # 性能优先，可能丢1秒数据
max_connections = 3000
thread_cache_size = 200

10.3 效果对比

指标	优化前	优化后
平均响应时间	850ms	45ms
成功率	65%	99.5%
数据库CPU	100%	35%
连接数峰值	1500	800

十一、常见误区与最佳实践

11.1 常见误区

盲目增大max_connections：不解决根本问题，反而增加内存消耗
过度使用索引：索引越多，写入越慢，维护成本越高
长事务：持有锁时间过长，阻塞其他操作
*SELECT **：返回不需要的列，增加I/O和网络开销
忽视从库延迟：读写分离时从库数据可能滞后

11.2 最佳实践清单

[ ] 使用连接池，合理设置池大小
[ ] 所有查询必须使用EXPLAIN分析
[ ] 事务尽可能短，避免事务内调用外部服务
[ ] 高并发写入使用批量操作+异步队列
[ ] 热点数据使用Redis缓存
[ ] 监控慢查询，定期优化
[ ] 主从复制开启GTID，便于故障恢复
[ ] 定期备份并测试恢复流程
[ ] 使用Percona Toolkit定期分析
[ ] 建立完善的监控预警体系

十二、总结

MySQL高并发优化是一个系统工程，需要从连接层、SQL层、事务层、架构层多维度入手。核心原则是：减少资源竞争、缩短操作时间、分散系统压力。

关键要点：

连接优化：连接池+读写分离，避免连接耗尽
SQL优化：合理索引+覆盖索引，减少I/O
事务优化：短事务+乐观锁，减少锁竞争
架构优化：缓存+队列+分库分表，分散压力
监控预警：实时监控+快速响应，防患于未然

记住，没有银弹。每个业务场景都有其特殊性，需要根据实际压力测试结果，持续监控和调优，才能构建稳定高效的MySQL高并发系统。# MySQL高并发处理策略：如何避免数据库崩溃并提升系统性能的实战指南