MySQL高并发处理策略：如何应对流量洪峰避免数据库崩溃与性能瓶颈的实战技巧

引言：理解高并发对MySQL的挑战

在当今互联网应用中，高并发场景已经成为常态。无论是电商秒杀、社交平台热点事件，还是金融交易系统，都可能面临瞬时流量洪峰的考验。MySQL作为最流行的关系型数据库，在高并发环境下容易出现性能瓶颈甚至崩溃。本文将深入探讨MySQL高并发处理的实战策略，帮助您构建稳定可靠的数据库系统。

高并发对MySQL的主要挑战包括：

• 连接数耗尽：大量并发请求导致连接池耗尽，新请求无法获取连接
• CPU瓶颈：复杂查询、排序、聚合操作消耗大量CPU资源
• I/O瓶颈：频繁的磁盘读写导致I/O等待，影响响应时间
• 锁竞争：行锁、表锁、元数据锁等导致事务阻塞
• 内存压力：缓冲池不足，频繁的磁盘I/O操作

一、架构层面的优化策略

1.1 读写分离架构

读写分离是应对高并发的最基础策略。通过将读请求和写请求分离到不同的数据库实例，可以显著降低主库压力。

实现方案：

• 主库（Master）：处理所有写操作（INSERT/UPDATE/DELETE）
• 从库（Slave）：处理所有读操作（SELECT）
• 中间件：使用ProxySQL、MyCat或应用层路由

代码示例（Spring Boot配置多数据源）：

@Configuration
public class DataSourceConfig {
    
    // 主数据源（写操作）
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.master")
    public DataSource masterDataSource() {
        return DataSourceBuilder.create().build();
    }
    
    // 从数据源（读操作）
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.slave")
    public DataSource slaveDataSource() {
        return DataSourceBuilder.create().build();
    }
    
    // 动态数据源路由
    @Bean
    public DataSource routingDataSource() {
        RoutingDataSource routingDataSource = new RoutingDataSource();
        Map<Object, Object> targetDataSources = new HashMap<>();
        targetDataSources.put("master", masterDataSource());
        targetDataSources.put("slave", slaveDataSource());
        routingDataSource.setTargetDataSources(targetDataSources);
        routingDataSource.setDefaultTargetDataSource(masterDataSource());
        return routingDataSource;
    }
}

// 自定义路由数据源
public class RoutingDataSource extends AbstractRoutingDataSource {
    @Override
    protected Object determineCurrentLookupKey() {
        return TransactionSynchronizationManager.isCurrentTransactionReadOnly() ? "slave" : "master";
    }
}

ProxySQL配置示例：

-- 添加主库
INSERT INTO mysql_servers (hostgroup_id, hostname, port, weight) 
VALUES (10, '192.168.1.100', 3306, 100);

-- 添加从库
INSERT INTO mysql_servers (hostgroup_id, hostname, port, weight) 
VALUES (20, '192.168.1.101', 3306, 100);

-- 配置读写分离规则
INSERT INTO mysql_query_rules (rule_id, active, match_digest, destination_hostgroup, apply) 
VALUES (1, 1, '^SELECT.*FOR UPDATE', 10, 1);

INSERT INTO mysql_query_rules (rule_id, active, match_digest, destination_hostgroup, apply) 
VALUES (2, 1, '^SELECT', 20, 1);

1.2 分库分表策略

当单表数据量超过千万级或并发量极高时，需要考虑分库分表。

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

• 用户库：user_db（用户表、用户扩展表）
• 订单库：order_db（订单表、订单详情表）
• 商品库：product_db（商品表、库存表）

水平分表： 将大表按规则拆分到多个表中

ShardingSphere-JDBC配置示例：

# application.yml
spring:
  shardingsphere:
    datasource:
      names: ds0, ds1
      ds0:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://192.168.1.100:3306/order_db
        username: root
        password: password
      ds1:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://192.168.1.101:3306/order_db
        username: root
        password: password
    
    rules:
      sharding:
        tables:
          orders:
            actual-data-nodes: ds${0..1}.order_${0..3}  # 2个库，每个库4张表
            table-strategy:
              standard:
                sharding-column: user_id
                sharding-algorithm-name: user_id_mod
            database-strategy:
              standard:
                sharding-column: order_id
                sharding-algorithm-name: order_id_mod
        
        sharding-algorithms:
          user_id_mod:
            type: MOD
            props:
              sharding-count: 4
          order_id_mod:
            type: MOD
            props:
              sharding-count: 2

1.3 缓存层策略

在高并发场景下，引入缓存层是减轻数据库压力的最有效手段。

Redis缓存策略：

@Service
public class ProductService {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private ProductMapper productMapper;
    
    private static final String PRODUCT_CACHE_PREFIX = "product:";
    private static final long CACHE_TTL = 300; // 5分钟
    
    // 查询商品详情（缓存穿透保护）
    public Product getProductById(Long id) {
        String cacheKey = PRODUCT_CACHE_PREFIX + id;
        
        // 1. 先查缓存
        Object cached = redisTemplate.opsForValue().get(cacheKey);
        if (cached != null) {
            return (Product) cached;
        }
        
        // 2. 缓存未命中，查询数据库
        Product product = productMapper.selectById(id);
        
        if (product != null) {
            // 3. 写入缓存
            redisTemplate.opsForValue().set(cacheKey, product, CACHE_TTL, TimeUnit.SECONDS);
        } else {
            // 4. 缓存空值，防止缓存穿透
            redisTemplate.opsForValue().set(cacheKey, "", 60, TimeUnit.SECONDS);
        }
        
        return product;
    }
    
    // 更新商品信息（缓存更新策略）
    @Transactional
    public void updateProduct(Product product) {
        // 1. 更新数据库
        productMapper.updateById(product);
        
        // 2. 删除缓存（Cache Aside模式）
        String cacheKey = PRODUCT_CACHE_PREFIX + product.getId();
        redisTemplate.delete(cacheKey);
    }
}

缓存预热： 在系统启动或低峰期预热热点数据

@Component
public class CacheWarmup {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private ProductMapper productMapper;
    
    @EventListener(ApplicationReadyEvent.class)
    public void warmupHotProducts() {
        // 从数据库加载热点商品数据
        List<Product> hotProducts = productMapper.selectHotProducts();
        
        for (Product product : hotProducts) {
            String cacheKey = "product:" + product.getId();
            redisTemplate.opsForValue().set(cacheKey, product, 300, TimeUnit.SECONDS);
        }
    }
}

二、MySQL配置优化

2.1 连接池优化

连接池配置直接影响高并发下的数据库连接管理。

HikariCP配置示例：

spring:
  datasource:
    hikari:
      # 连接池名称
      pool-name: HikariCP-Pool
      # 最小空闲连接数
      minimum-idle: 20
      # 最大连接数（根据并发量调整）
      maximum-pool-size: 100
      # 连接超时时间（毫秒）
      connection-timeout: 30000
      # 空闲连接超时时间（毫秒）
      idle-timeout: 600000
      # 连接最大生命周期（毫秒）
      max-lifetime: 1800000
      # 连接测试查询
      connection-test-query: SELECT 1
      # 是否缓存预编译语句
      cache-prep-statements: true
      # 预编译语句缓存大小
      prep-statement-cache-size: 250
      # 预编译语句缓存上限
      prep-statement-cache-sql-limit: 2048

MySQL服务器连接配置：

-- 查看当前连接数
SHOW STATUS LIKE 'Threads_connected';

-- 查看最大连接数
SHOW VARIABLES LIKE 'max_connections';

-- 修改最大连接数（临时生效）
SET GLOBAL max_connections = 1000;

-- 永久修改（my.cnf）
[mysqld]
max_connections = 1000
back_log = 500
wait_timeout = 600
interactive_timeout = 600

2.2 InnoDB缓冲池优化

InnoDB缓冲池是MySQL性能的关键，决定了数据页在内存中的缓存效率。

配置参数：

-- 查看缓冲池大小
SHOW VARIABLES LIKE 'innodb_buffer_pool_size';

-- 查看缓冲池使用情况
SHOW ENGINE INNODB STATUS\G

-- 在my.cnf中配置
[mysqld]
# 设置为物理内存的50%-70%
innodb_buffer_pool_size = 8G
# 缓冲池实例数（建议与CPU核心数相同）
innodb_buffer_pool_instances = 8
# 缓冲池刷新策略
innodb_buffer_pool_flush_at_neighbors = 1
# 页大小
innodb_page_size = 16384

监控缓冲池命中率：

-- 计算缓冲池命中率
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';

-- 理想命中率应 > 99%，如果低于95%需要增加缓冲池大小

2.3 查询缓存与日志优化

查询缓存配置（MySQL 8.0已移除，适用于5.7及以下）：

-- 查看查询缓存状态
SHOW VARIABLES LIKE 'query_cache%';

-- 配置查询缓存（my.cnf）
[mysqld]
query_cache_type = 1
query_cache_size = 64M
query_cache_limit = 2M

日志优化配置：

[mysqld]
# 事务日志（redo log）
innodb_log_file_size = 2G
innodb_log_buffer_size = 64M
innodb_log_files_in_group = 3

# 二进制日志（用于复制和恢复）
log_bin = mysql-bin
binlog_format = ROW
expire_logs_days = 7
max_binlog_size = 500M

# 慢查询日志（用于性能分析）
slow_query_log = 1
slow_query_log_file = /var/log/mysql/slow.log
long_query_time = 2
log_queries_not_using_indexes = 1

三、SQL语句优化技巧

3.1 索引优化策略

索引设计原则：

• 选择性高的列适合建索引
• 联合索引遵循最左前缀原则
• 避免过多索引（影响写性能）
• 定期分析索引使用情况

创建高效索引的示例：

-- 原始低效查询
SELECT * FROM orders WHERE user_id = 123 AND status = 'PAID' AND created_at > '2024-01-01';

-- 优化方案1：创建复合索引
CREATE INDEX idx_user_status_created ON orders(user_id, status, created_at);

-- 优化方案2：覆盖索引（避免回表）
CREATE INDEX idx_user_status_created_cover ON orders(user_id, status, created_at, order_id, amount);

-- 查看索引使用情况
EXPLAIN SELECT * FROM orders WHERE user_id = 123 AND status = 'PAID' AND created_at > '2024-01-01';

索引维护：

-- 查看表索引
SHOW INDEX FROM orders;

-- 分析索引使用情况
SELECT * FROM sys.schema_index_statistics WHERE table_schema = 'your_database';

-- 删除未使用的索引（需要先开启统计信息收集）
ANALYZE TABLE orders;
-- 然后通过performance_schema查看索引使用频率

3.2 避免全表扫描

反例（导致全表扫描）：

-- 错误：在索引列上使用函数
SELECT * FROM users WHERE DATE(created_at) = '2024-01-01';

-- 错误：使用 != 或 <> 
SELECT * FROM products WHERE status != 'DELETED';

-- 错误：使用 OR 连接不同列
SELECT * FROM orders WHERE user_id = 123 OR amount > 1000;

正例（使用索引）：

-- 正确：直接比较日期范围
SELECT * FROM users WHERE created_at >= '2024-01-01' AND created_at < '2024-01-02';

-- 正确：使用 IN 代替 !=
SELECT * FROM products WHERE status IN ('ACTIVE', 'PENDING');

-- 正确：使用 UNION ALL 代替 OR
SELECT * FROM orders WHERE user_id = 123
UNION ALL
SELECT * FROM orders WHERE amount > 1000;

3.3 分页优化

传统分页的性能问题：

-- 低效：深度分页时性能极差
SELECT * FROM orders ORDER BY id LIMIT 1000000, 20;

优化方案1：延迟关联

-- 先查出主键，再关联详情
SELECT o.* 
FROM orders o
INNER JOIN (
    SELECT id 
    FROM orders 
    ORDER BY id 
    LIMIT 1000000, 20
) AS tmp ON o.id = tmp.id;

优化方案2：位置记录法

-- 记录上一页最后一条记录的ID
SELECT * FROM orders WHERE id > 1000000 ORDER BY id LIMIT 20;

优化方案3：ES分页（适用于超大数据量）

// 使用Elasticsearch进行分页查询
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
sourceBuilder.from(1000000);
sourceBuilder.size(20);
sourceBuilder.sort("id", SortOrder.ASC);

3.4 批量操作优化

批量插入优化：

// 低效：逐条插入
for (Order order : orders) {
    orderMapper.insert(order);
}

// 高效：批量插入
@InsertProvider(type = OrderSqlProvider.class, method = "batchInsert")
void batchInsert(@Param("orders") List<Order> orders);

// SQL Provider
public class OrderSqlProvider {
    public String batchInsert(Map<String, Object> param) {
        List<Order> orders = (List<Order>) param.get("orders");
        StringBuilder sql = new StringBuilder();
        sql.append("INSERT INTO orders (user_id, amount, status, created_at) VALUES ");
        for (int i = 0; i < orders.size(); i++) {
            if (i > 0) sql.append(",");
            sql.append("(#{orders[" + i + "].userId}, #{orders[" + i + "].amount}, " +
                     "#{orders[" + i + "].status}, #{orders[" + i + "].createdAt})");
        }
        return sql.toString();
    }
}

批量更新优化：

-- 低效：逐条更新
UPDATE orders SET status = 'PAID' WHERE id = 1;
UPDATE orders SET status = 'PAID' WHERE id = 2;
UPDATE orders SET status = 'PAID' WHERE id = 3;

-- 高效：批量更新
UPDATE orders SET status = 'PAID' WHERE id IN (1, 2, 3);

-- 或使用CASE WHEN（适用于不同值）
UPDATE orders 
SET status = CASE id 
    WHEN 1 THEN 'PAID'
    WHEN 2 THEN 'SHIPPED'
    WHEN 3 THEN 'CANCELLED'
END
WHERE id IN (1, 2, 3);

四、事务与锁优化

4.1 事务隔离级别选择

不同隔离级别的适用场景：

-- 查看当前隔离级别
SELECT @@transaction_isolation;

-- 设置隔离级别
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;

-- 在my.cnf中全局设置
[mysqld]
transaction-isolation = READ-COMMITTED

Spring Boot中配置事务隔离级别：

@Service
public class OrderService {
    
    // 读已提交级别，避免脏读，性能较好
    @Transactional(isolation = Isolation.READ_COMMITTED)
    public void createOrder(Order order) {
        // 业务逻辑
    }
    
    // 可重复读（默认级别），适用于需要一致性读的场景
    @Transactional(isolation = Isolation.REPEATABLE_READ)
    public void updateProductStock(Long productId, Integer quantity) {
        // 库存扣减逻辑
    }
}

4.2 锁优化策略

减少锁粒度：

-- 使用行锁代替表锁
-- 错误：表锁
LOCK TABLES orders WRITE;

-- 正确：行锁（InnoDB默认）
UPDATE orders SET status = 'PAID' WHERE id = 123;

-- 使用乐观锁（适用于读多写少）
ALTER TABLE orders ADD COLUMN version INT DEFAULT 0;

-- 乐观锁更新
UPDATE orders 
SET status = 'PAID', version = version + 1 
WHERE id = 123 AND version = 1;

避免死锁：

// 固定加锁顺序
public void transferMoney(Long fromAccountId, Long toAccountId, BigDecimal amount) {
    // 确保所有事务按相同顺序加锁
    Long lock1 = Math.min(fromAccountId, toAccountId);
    Long lock2 = Math.max(fromAccountId, toAccountId);
    
    // 先锁lock1，再锁lock2
    synchronized (lock1.toString().intern()) {
        synchronized (lock2.toString().intern()) {
            // 转账逻辑
        }
    }
}

4.3 长事务优化

识别长事务：

-- 查看当前运行的长事务
SELECT * FROM information_schema.processlist 
WHERE time > 60 AND command != 'Sleep';

-- 查看InnoDB事务状态
SHOW ENGINE INNODB STATUS\G

优化长事务：

// 将大事务拆分为小事务
@Transactional
public void processBatch(List<Order> orders) {
    for (Order order : orders) {
        // 每次处理一条记录就提交事务
        processSingleOrder(order);
        // 手动提交，避免事务过大
        TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
    }
}

// 或者使用REQUIRES_NEW传播行为
@Transactional(propagation = Propagation.REQUIRES_NEW)
public void processSingleOrder(Order order) {
    orderMapper.updateById(order);
}

五、监控与诊断工具

5.1 性能监控指标

关键监控指标：

-- QPS（每秒查询数）
SHOW GLOBAL STATUS LIKE 'Queries';
SHOW GLOBAL STATUS LIKE 'Uptime';

-- TPS（每秒事务数）
SHOW GLOBAL STATUS LIKE 'Com_commit';
SHOW GLOBAL STATUS LIKE 'Com_rollback';

-- 连接数监控
SHOW GLOBAL STATUS LIKE 'Threads_connected';
SHOW GLOBAL STATUS LIKE 'Threads_running';

-- 缓冲池命中率
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';

慢查询日志分析：

# 使用mysqldumpslow分析慢查询日志
mysqldumpslow -s t -t 10 /var/log/mysql/slow.log

# 使用pt-query-digest（更详细）
pt-query-digest /var/log/mysql/slow.log > slow_report.txt

5.2 实时诊断工具

Performance Schema查询：

-- 查看当前最耗时的查询
SELECT 
    THREAD_ID,
    EVENT_NAME,
    SUM_TIMER_WAIT/1000000000000 AS wait_seconds,
    COUNT_STAR
FROM performance_schema.events_statements_summary_by_thread
ORDER BY wait_seconds DESC
LIMIT 10;

-- 查看表I/O统计
SELECT 
    OBJECT_SCHEMA,
    OBJECT_NAME,
    SUM_TIMER_READ/1000000000000 AS read_seconds,
    SUM_TIMER_WRITE/1000000000000 AS write_seconds
FROM performance_schema.table_io_waits_summary_by_table
ORDER BY read_seconds + write_seconds DESC
LIMIT 10;

sys schema视图：

-- 查看最耗资源的查询
SELECT * FROM sys.statement_analysis ORDER BY avg_time DESC LIMIT 10;

-- 查看未使用索引的表
SELECT * FROM sys.schema_unused_indexes;

-- 查看重复索引
SELECT * FROM sys.schema_redundant_indexes;

5.3 应用层监控

Micrometer + Prometheus监控：

@Component
public class DatabaseMetrics {
    
    private final MeterRegistry registry;
    
    public DatabaseMetrics(MeterRegistry registry) {
        this.registry = registry;
    }
    
    // 记录查询耗时
    public <T> T recordQueryTime(String query, Supplier<T> queryFunction) {
        Timer.Sample sample = Timer.start(registry);
        try {
            return queryFunction.get();
        } finally {
            sample.stop(registry.timer("db.query.time", "query", query));
        }
    }
    
    // 记录连接获取时间
    public void recordConnectionAcquireTime(long timeMs) {
        registry.timer("db.connection.acquire").record(timeMs, TimeUnit.MILLISECONDS);
    }
}

六、高并发场景实战案例

6.1 秒杀系统设计

秒杀场景特点：

• 瞬时高并发（QPS可达10万+）
• 库存扣减（需要原子性）
• 请求过滤（大量无效请求）

完整解决方案：

@Service
public class SeckillService {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private OrderMapper orderMapper;
    
    @Autowired
    private ProductMapper productMapper;
    
    private static final String STOCK_KEY = "seckill:stock:";
    private static final String ORDER_KEY = "seckill:order:";
    private static final String USER_KEY = "seckill:user:";
    
    /**
     * 秒杀流程：
     * 1. 前置校验（Redis预减库存）
     * 2. 生成订单（数据库）
     * 3. 异步处理（消息队列）
     */
    public SeckillResult seckill(Long productId, Long userId) {
        String stockKey = STOCK_KEY + productId;
        String userKey = USER_KEY + productId;
        
        // 1. 重复购买校验
        if (redisTemplate.opsForSet().isMember(userKey, userId)) {
            return SeckillResult.fail("您已购买过该商品");
        }
        
        // 2. Redis预减库存（原子操作）
        Long stock = redisTemplate.opsForValue().decrement(stockKey);
        if (stock < 0) {
            redisTemplate.opsForValue().increment(stockKey); // 回滚
            return SeckillResult.fail("库存不足");
        }
        
        // 3. 发送消息到队列（异步创建订单）
        SeckillMessage message = new SeckillMessage(productId, userId);
        rabbitTemplate.convertAndSend("seckill.exchange", "seckill.key", message);
        
        return SeckillResult.success("秒杀成功，订单处理中");
    }
    
    /**
     * 异步创建订单（消费者）
     */
    @RabbitListener(queues = "seckill.queue")
    public void createOrder(SeckillMessage message) {
        try {
            // 1. 检查数据库库存
            Product product = productMapper.selectById(message.getProductId());
            if (product.getStock() <= 0) {
                return;
            }
            
            // 2. 数据库扣减库存（乐观锁）
            int updated = productMapper.decreaseStock(message.getProductId());
            if (updated == 0) {
                return; // 库存不足，回滚Redis
            }
            
            // 3. 创建订单
            Order order = new Order();
            order.setUserId(message.getUserId());
            order.setProductId(message.getProductId());
            order.setAmount(product.getPrice());
            order.setStatus("CREATED");
            orderMapper.insert(order);
            
            // 4. 记录已购买用户
            String userKey = USER_KEY + message.getProductId();
            redisTemplate.opsForSet().add(userKey, message.getUserId());
            
        } catch (Exception e) {
            // 异常处理：回滚Redis库存
            redisTemplate.opsForValue().increment(STOCK_KEY + message.getProductId());
        }
    }
}

数据库表设计：

-- 商品表（库存字段单独索引）
CREATE TABLE products (
    id BIGINT PRIMARY KEY,
    name VARCHAR(255),
    price DECIMAL(10,2),
    stock INT,
    version INT DEFAULT 0,  -- 乐观锁版本号
    INDEX idx_stock (stock)  -- 库存索引
);

-- 订单表（分表字段）
CREATE TABLE orders (
    id BIGINT PRIMARY KEY,
    user_id BIGINT,
    product_id BIGINT,
    amount DECIMAL(10,2),
    status VARCHAR(50),
    created_at TIMESTAMP,
    INDEX idx_user_id (user_id),
    INDEX idx_product_id (product_id)
);

6.2 热点数据更新冲突

场景： 热点商品库存扣减导致的锁竞争

解决方案：

@Service
public class HotProductService {
    
    /**
     * 优化前：直接更新（高并发下锁竞争严重）
     */
    @Transactional
    public boolean deductStock(Long productId, Integer quantity) {
        // 高并发下，大量事务在此等待锁
        Product product = productMapper.selectById(productId);
        if (product.getStock() < quantity) {
            return false;
        }
        productMapper.updateStock(productId, product.getStock() - quantity);
        return true;
    }
    
    /**
     * 优化后：分段锁 + 缓存
     */
    public boolean deductStockOptimized(Long productId, Integer quantity) {
        // 1. 分段锁（将库存拆分为多个段）
        int segment = productId % 10; // 10个分段
        String lockKey = "stock:lock:" + segment;
        
        // 2. 获取分布式锁（短暂锁定）
        RLock lock = redissonClient.getLock(lockKey);
        try {
            if (lock.tryLock(100, 10, TimeUnit.MILLISECONDS)) {
                // 3. 检查缓存库存
                String stockKey = "product:stock:" + productId;
                Integer stock = (Integer) redisTemplate.opsForValue().get(stockKey);
                
                if (stock == null) {
                    // 4. 缓存未命中，查询数据库
                    Product product = productMapper.selectById(productId);
                    stock = product.getStock();
                    redisTemplate.opsForValue().set(stockKey, stock, 60, TimeUnit.SECONDS);
                }
                
                if (stock < quantity) {
                    return false;
                }
                
                // 5. 扣减缓存库存
                redisTemplate.opsForValue().decrement(stockKey, quantity);
                
                // 6. 发送异步消息更新数据库
                sendUpdateMessage(productId, quantity);
                
                return true;
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
        
        return false;
    }
    
    private void sendUpdateMessage(Long productId, Integer quantity) {
        // 异步更新数据库，避免阻塞
        UpdateMessage msg = new UpdateMessage(productId, quantity);
        rabbitTemplate.convertAndSend("stock.update.exchange", "stock.update.key", msg);
    }
}

6.3 数据库连接池耗尽

问题现象： 应用启动时报 Communications link failure 或 Too many connections

诊断步骤：

-- 1. 查看当前连接数
SHOW STATUS LIKE 'Threads_connected';

-- 2. 查看连接数限制
SHOW VARIABLES LIKE 'max_connections';

-- 3. 查看当前连接详情
SELECT 
    id,
    user,
    host,
    db,
    command,
    time,
    state,
    info
FROM information_schema.processlist
WHERE command != 'Sleep'
ORDER BY time DESC;

-- 4. 查看连接数历史峰值
SELECT * FROM performance_schema.global_status 
WHERE VARIABLE_NAME = 'Threads_connected'
ORDER BY VARIABLE_VALUE DESC
LIMIT 10;

解决方案：

# 1. 调整应用连接池
spring:
  datasource:
    hikari:
      maximum-pool-size: 200  # 根据并发量调整
      minimum-idle: 50
      connection-timeout: 30000
      idle-timeout: 600000
      max-lifetime: 1800000

# 2. 调整MySQL配置
[mysqld]
max_connections = 1000
back_log = 500
wait_timeout = 600
interactive_timeout = 600

# 3. 代码层面优化
# - 使用连接池监控
# - 及时释放连接
# - 避免长事务

七、总结与最佳实践

7.1 高并发处理 checklist

架构层面：

• [ ] 实现读写分离
• [ ] 引入Redis缓存
• [ ] 考虑分库分表
• [ ] 使用消息队列削峰

MySQL配置：

• [ ] 连接池大小合理
• [ ] 缓冲池设置为内存的50-70%
• [ ] 慢查询日志开启
• [ ] 事务日志大小适当

SQL优化：

• [ ] 关键字段都有索引
• [ ] 避免全表扫描
• [ ] 批量操作代替单条操作
• [ ] 分页优化

监控告警：

• [ ] QPS/TPS监控
• [ ] 连接数告警
• [ ] 慢查询告警
• [ ] 缓冲池命中率监控

7.2 性能优化黄金法则

1. 先测量，再优化：使用监控工具定位真实瓶颈
2. 缓存为王：90%的性能问题可以通过缓存解决
3. 异步化：将非核心逻辑异步处理
4. 分而治之：通过拆分降低单点压力
5. 降级熔断：保证核心服务可用

7.3 持续优化建议

• 定期review慢查询日志：每周分析一次，优化TOP10慢查询
• 定期分析索引使用情况：删除冗余索引，补充缺失索引
• 压力测试：定期进行全链路压测，发现潜在问题
• 容量规划：根据业务增长预测，提前扩容

通过以上策略的综合运用，可以有效应对高并发场景下的MySQL性能挑战，确保系统在流量洪峰下依然稳定可靠。记住，没有银弹，需要根据具体业务场景选择合适的组合方案。