在当今互联网应用中，高并发场景已成为常态。无论是电商大促、社交网络热点事件，还是金融交易系统，都面临着海量用户同时访问数据库的挑战。MySQL作为最流行的开源关系型数据库，如何在高并发环境下保持稳定、高效的性能，是每个后端工程师必须掌握的核心技能。本文将从数据库优化、应用层优化、架构设计三个维度，结合实战案例，为你提供一套完整的MySQL高并发处理策略。
一、数据库层优化：夯实性能基础
数据库层的优化是高并发处理的基石。如果数据库本身存在性能瓶颈，任何上层优化都将是空中楼阁。
1.1 索引优化：查询性能的加速器
索引是MySQL中最核心的优化手段。一个合理的索引设计能将查询性能提升几个数量级。
索引设计原则：
- 覆盖索引：索引包含查询所需的所有字段，避免回表操作。
- 最左前缀原则：对于复合索引，查询条件必须从最左列开始匹配。
- 避免过度索引：索引会占用存储空间并影响写入性能。
实战案例：订单表查询优化
假设我们有一个订单表orders，包含以下字段：
CREATE TABLE `orders` (
`id` BIGINT PRIMARY KEY AUTO_INCREMENT,
`user_id` BIGINT NOT NULL,
`order_no` VARCHAR(64) NOT NULL,
`status` TINYINT NOT NULL,
`amount` DECIMAL(10,2) NOT NULL,
`create_time` DATETIME NOT NULL,
`update_time` DATETIME NOT NULL,
INDEX `idx_user_status` (`user_id`, `status`, `create_time`), -- 复合索引
INDEX `idx_order_no` (`order_no`), -- 唯一索引
INDEX `idx_create_time` (`create_time`) -- 单列索引
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
场景1：查询用户未完成的订单
-- 优化前：全表扫描
SELECT * FROM orders WHERE user_id = 12345 AND status = 0;
-- 优化后：使用复合索引idx_user_status
-- 索引覆盖了user_id、status、create_time，查询效率极高
SELECT order_no, amount, create_time
FROM orders
WHERE user_id = 12345 AND status = 0
ORDER BY create_time DESC;
场景2：按订单号查询
-- 使用唯一索引idx_order_no，查询速度极快
SELECT * FROM orders WHERE order_no = 'ORD202310010001';
索引优化检查工具：
-- 查看表的索引使用情况
SHOW INDEX FROM orders;
-- 使用EXPLAIN分析查询计划
EXPLAIN SELECT * FROM orders WHERE user_id = 12345 AND status = 0;
1.2 查询优化：避免慢查询
慢查询是高并发场景下的性能杀手。我们需要从SQL编写和执行计划两个层面进行优化。
常见慢查询问题及解决方案：
问题1：SELECT * 导致的回表
-- 问题SQL：查询所有字段，如果表字段多，会消耗大量IO
SELECT * FROM orders WHERE user_id = 12345;
-- 优化方案：只查询需要的字段，最好使用覆盖索引
SELECT id, order_no, amount, create_time
FROM orders
WHERE user_id = 12345;
问题2：大表JOIN导致的性能问题
-- 问题SQL：大表JOIN，没有使用索引
SELECT o.*, u.username
FROM orders o
JOIN users u ON o.user_id = u.id
WHERE o.create_time > '2023-01-01';
-- 优化方案1：确保JOIN字段有索引
ALTER TABLE users ADD INDEX idx_id (id);
ALTER TABLE orders ADD INDEX idx_user_id (user_id);
-- 优化方案2：如果users表很大，考虑使用子查询或分步查询
-- 先查询orders，再批量查询users
SELECT o.* FROM orders o WHERE o.create_time > '2023-01-01';
-- 然后根据user_id列表批量查询users信息
问题3：模糊查询导致的索引失效
-- 问题SQL：前缀模糊查询，索引失效
SELECT * FROM orders WHERE order_no LIKE '%ORD2023%';
-- 优化方案1：使用后缀模糊查询，索引有效
SELECT * FROM orders WHERE order_no LIKE 'ORD2023%';
-- 优化方案2：使用全文索引（MySQL 5.6+）
ALTER TABLE orders ADD FULLTEXT INDEX ft_order_no (order_no);
SELECT * FROM orders WHERE MATCH(order_no) AGAINST('ORD2023');
慢查询日志配置与分析：
# MySQL配置文件my.cnf
[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 # 记录未使用索引的查询
使用pt-query-digest分析慢日志：
# 安装Percona Toolkit
sudo apt-get install percona-toolkit
# 分析慢日志
pt-query-digest /var/log/mysql/slow.log > slow_report.txt
1.3 表结构设计优化
1.3.1 数据类型选择
- 使用最小的数据类型：能用TINYINT就不用INT，能用VARCHAR(50)就不用VARCHAR(255)
- 避免使用TEXT/BLOB：大字段会降低性能，考虑拆分到单独的表
- 使用ENUM代替字符串：状态字段使用ENUM类型，节省空间且查询更快
-- 优化前：使用VARCHAR存储状态
CREATE TABLE `orders` (
`status` VARCHAR(20) NOT NULL -- 'pending', 'paid', 'shipped', 'completed'
);
-- 优化后：使用ENUM类型
CREATE TABLE `orders` (
`status` ENUM('pending', 'paid', 'shipped', 'completed') NOT NULL
);
1.3.2 分区表设计 对于超大表（千万级以上），可以考虑使用分区表来提升查询性能。
-- 按时间范围分区
CREATE TABLE `orders_partitioned` (
`id` BIGINT NOT NULL,
`user_id` BIGINT NOT NULL,
`order_no` VARCHAR(64) NOT NULL,
`status` TINYINT NOT NULL,
`amount` DECIMAL(10,2) NOT NULL,
`create_time` DATETIME NOT NULL,
PRIMARY KEY (`id`, `create_time`)
) PARTITION BY RANGE (YEAR(create_time) * 100 + MONTH(create_time)) (
PARTITION p202301 VALUES LESS THAN (202302),
PARTITION p202302 VALUES LESS THAN (202303),
PARTITION p202303 VALUES LESS THAN (202304),
PARTITION p202304 VALUES LESS THAN (202305),
PARTITION p202305 VALUES LESS THAN (202306),
PARTITION p202306 VALUES LESS THAN (202307),
PARTITION p202307 VALUES LESS THAN (202308),
PARTITION p202308 VALUES LESS THAN (202309),
PARTITION p202309 VALUES LESS THAN (202310),
PARTITION p202310 VALUES LESS THAN (202311),
PARTITION p202311 VALUES LESS THAN (202312),
PARTITION p202312 VALUES LESS THAN (202401),
PARTITION p_future VALUES LESS THAN MAXVALUE
);
1.4 MySQL配置参数优化
关键参数调整：
# my.cnf 配置示例（根据服务器硬件调整）
[mysqld]
# 连接相关
max_connections = 2000
max_connect_errors = 1000
wait_timeout = 600
interactive_timeout = 600
# InnoDB缓冲池（建议设置为物理内存的50-70%）
innodb_buffer_pool_size = 16G
innodb_buffer_pool_instances = 8
# 日志相关
innodb_log_file_size = 2G
innodb_log_buffer_size = 64M
innodb_flush_log_at_trx_commit = 2 # 高并发场景下可设置为2，提升性能
# 事务相关
innodb_flush_method = O_DIRECT
innodb_file_per_table = 1
innodb_read_io_threads = 8
innodb_write_io_threads = 8
# 查询缓存（MySQL 8.0已移除，5.7及以下可考虑）
query_cache_type = 0
query_cache_size = 0
# 临时表
tmp_table_size = 256M
max_heap_table_size = 256M
# 其他
innodb_lock_wait_timeout = 50
innodb_rollback_on_timeout = 0
动态调整参数（无需重启）：
-- 查看当前缓冲池大小
SHOW VARIABLES LIKE 'innodb_buffer_pool_size';
-- 动态调整缓冲池大小（MySQL 5.7+）
SET GLOBAL innodb_buffer_pool_size = 16 * 1024 * 1024 * 1024;
-- 查看当前连接数
SHOW STATUS LIKE 'Threads_connected';
-- 查看慢查询数量
SHOW STATUS LIKE 'Slow_queries';
二、应用层优化：减轻数据库压力
应用层的优化可以有效减少数据库的直接访问压力，提升整体系统吞吐量。
2.1 连接池管理
数据库连接是宝贵的资源，频繁创建和销毁连接会消耗大量CPU和内存。
连接池配置示例（以HikariCP为例）：
// Spring Boot配置
@Configuration
public class DataSourceConfig {
@Bean
@ConfigurationProperties("spring.datasource.hikari")
public DataSource dataSource() {
HikariDataSource dataSource = new HikariDataSource();
// 基本配置
dataSource.setJdbcUrl("jdbc:mysql://localhost:3306/mydb?useSSL=false&serverTimezone=UTC");
dataSource.setUsername("root");
dataSource.setPassword("password");
// 连接池配置
dataSource.setMaximumPoolSize(50); // 最大连接数
dataSource.setMinimumIdle(10); // 最小空闲连接
dataSource.setConnectionTimeout(30000); // 连接超时时间(ms)
dataSource.setIdleTimeout(600000); // 空闲连接超时时间(ms)
dataSource.setMaxLifetime(1800000); // 连接最大生命周期(ms)
dataSource.setConnectionTestQuery("SELECT 1"); // 连接测试查询
// 性能优化
dataSource.setPoolName("MyHikariCP");
dataSource.setAutoCommit(true);
dataSource.setLeakDetectionThreshold(60000); // 连接泄漏检测
return dataSource;
}
}
连接池监控：
// 监控连接池状态
@Component
public class ConnectionPoolMonitor {
@Scheduled(fixedRate = 60000) // 每分钟执行一次
public void monitorConnectionPool() {
HikariDataSource dataSource = (HikariDataSource) this.dataSource;
HikariPoolMXBean poolMXBean = dataSource.getHikariPoolMXBean();
System.out.println("=== 连接池状态监控 ===");
System.out.println("活跃连接数: " + poolMXBean.getActiveConnections());
System.out.println("空闲连接数: " + poolMXBean.getIdleConnections());
System.out.println("总连接数: " + poolMXBean.getTotalConnections());
System.out.println("等待连接的线程数: " + poolMXBean.getThreadsAwaitingConnection());
}
}
2.2 缓存策略
缓存是减少数据库访问的最有效手段。合理的缓存策略可以将数据库QPS降低90%以上。
2.2.1 本地缓存（Caffeine）
// 使用Caffeine实现本地缓存
@Component
public class LocalCacheService {
private final Cache<String, Object> cache = Caffeine.newBuilder()
.maximumSize(10000) // 最大条目数
.expireAfterWrite(10, TimeUnit.MINUTES) // 写入后10分钟过期
.recordStats() // 记录统计信息
.build();
// 获取缓存，如果不存在则从数据库加载
public <T> T get(String key, Supplier<T> loader) {
return (T) cache.get(key, k -> loader.get());
}
// 手动设置缓存
public void put(String key, Object value) {
cache.put(key, value);
}
// 删除缓存
public void invalidate(String key) {
cache.invalidate(key);
}
// 获取缓存统计信息
public void printStats() {
CacheStats stats = cache.stats();
System.out.println("缓存命中率: " + stats.hitRate());
System.out.println("平均加载时间: " + stats.averageLoadPenalty() / 1_000_000 + "ms");
}
}
2.2.2 分布式缓存（Redis）
// Redis缓存配置
@Configuration
@EnableCaching
public class RedisCacheConfig {
@Bean
public RedisCacheManager cacheManager(RedisConnectionFactory connectionFactory) {
RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
.entryTtl(Duration.ofMinutes(10)) // 默认过期时间
.serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()))
.serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()));
return RedisCacheManager.builder(connectionFactory)
.cacheDefaults(config)
.build();
}
}
// 使用缓存注解
@Service
public class OrderService {
@Autowired
private OrderRepository orderRepository;
@Cacheable(value = "orders", key = "#orderId")
public Order getOrderById(Long orderId) {
System.out.println("从数据库查询订单: " + orderId);
return orderRepository.findById(orderId).orElse(null);
}
@CachePut(value = "orders", key = "#order.id")
public Order updateOrder(Order order) {
System.out.println("更新订单: " + order.getId());
return orderRepository.save(order);
}
@CacheEvict(value = "orders", key = "#orderId")
public void deleteOrder(Long orderId) {
System.out.println("删除订单: " + orderId);
orderRepository.deleteById(orderId);
}
}
2.2.3 多级缓存架构
// 多级缓存实现：本地缓存 + Redis缓存 + 数据库
@Component
public class MultiLevelCacheService {
@Autowired
private LocalCacheService localCache;
@Autowired
private RedisTemplate<String, Object> redisTemplate;
@Autowired
private OrderRepository orderRepository;
public Order getOrderWithCache(Long orderId) {
String cacheKey = "order:" + orderId;
// 1. 本地缓存查询
Order order = localCache.get(cacheKey, () -> {
// 2. Redis缓存查询
Object redisData = redisTemplate.opsForValue().get(cacheKey);
if (redisData != null) {
return (Order) redisData;
}
// 3. 数据库查询
Order dbOrder = orderRepository.findById(orderId).orElse(null);
if (dbOrder != null) {
// 回填Redis缓存
redisTemplate.opsForValue().set(cacheKey, dbOrder, Duration.ofMinutes(10));
}
return dbOrder;
});
return order;
}
}
2.3 读写分离
读写分离是提升数据库吞吐量的经典方案。主库负责写操作，从库负责读操作。
2.3.1 基于ShardingSphere的读写分离
# 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-host: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-host: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-host: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-balancer-name: round_robin
load-balancers:
round_robin:
type: ROUND_ROBIN
2.3.2 自定义读写分离实现
// 自定义数据源路由
@Component
public class DynamicDataSource extends AbstractRoutingDataSource {
private static final ThreadLocal<String> CONTEXT_HOLDER = new ThreadLocal<>();
public static void setDataSource(String dataSource) {
CONTEXT_HOLDER.set(dataSource);
}
public static void clearDataSource() {
CONTEXT_HOLDER.remove();
}
@Override
protected Object determineCurrentLookupKey() {
return CONTEXT_HOLDER.get();
}
}
// 切面实现读写分离
@Aspect
@Component
public class ReadWriteSplittingAspect {
@Around("@annotation(readWriteSplitting)")
public Object around(ProceedingJoinPoint joinPoint, ReadWriteSplitting readWriteSplitting) throws Throwable {
try {
// 根据方法注解决定使用主库还是从库
if (readWriteSplitting.readOnly()) {
// 读操作，使用从库（随机选择）
String[] slaves = {"slave0", "slave1"};
String slave = slaves[new Random().nextInt(slaves.length)];
DynamicDataSource.setDataSource(slave);
} else {
// 写操作，使用主库
DynamicDataSource.setDataSource("master");
}
return joinPoint.proceed();
} finally {
DynamicDataSource.clearDataSource();
}
}
}
// 使用示例
@Service
public class OrderService {
@ReadWriteSplitting(readOnly = true)
public Order getOrder(Long id) {
// 这个方法会使用从库
return orderRepository.findById(id).orElse(null);
}
@ReadWriteSplitting(readOnly = false)
public Order createOrder(Order order) {
// 这个方法会使用主库
return orderRepository.save(order);
}
}
2.4 异步处理与消息队列
对于耗时操作，使用异步处理可以避免阻塞主线程，提升系统响应速度。
2.4.1 使用Spring Async实现异步
@Configuration
@EnableAsync
public class AsyncConfig {
@Bean
public TaskExecutor taskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(10);
executor.setMaxPoolSize(50);
executor.setQueueCapacity(100);
executor.setThreadNamePrefix("async-");
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
}
@Service
public class OrderService {
@Async
public CompletableFuture<Order> createOrderAsync(Order order) {
// 异步创建订单，不阻塞主线程
Order savedOrder = orderRepository.save(order);
// 发送消息到消息队列，异步处理后续业务
sendOrderCreatedMessage(savedOrder);
return CompletableFuture.completedFuture(savedOrder);
}
private void sendOrderCreatedMessage(Order order) {
// 发送到消息队列
System.out.println("发送订单创建消息: " + order.getId());
}
}
2.4.2 使用消息队列（RabbitMQ）
// RabbitMQ配置
@Configuration
public class RabbitMQConfig {
@Bean
public Queue orderQueue() {
return new Queue("order.queue", true); // 持久化队列
}
@Bean
public TopicExchange orderExchange() {
return new TopicExchange("order.exchange");
}
@Bean
public Binding binding(Queue queue, TopicExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("order.*");
}
}
// 生产者
@Service
public class OrderProducer {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendOrderCreated(Order order) {
rabbitTemplate.convertAndSend("order.exchange", "order.created", order);
}
}
// 消费者
@Component
public class OrderConsumer {
@RabbitListener(queues = "order.queue")
public void handleOrderCreated(Order order) {
// 异步处理订单创建后的业务逻辑
System.out.println("处理订单创建事件: " + order.getId());
// 发送邮件、更新统计等耗时操作
}
}
三、架构设计：构建可扩展的系统
当单机数据库无法满足高并发需求时，需要从架构层面进行扩展。
3.1 分库分表
分库分表是解决单机数据库性能瓶颈的终极方案。
3.1.1 垂直分库
-- 原始单库设计
CREATE DATABASE mydb;
USE mydb;
CREATE TABLE users (...);
CREATE TABLE orders (...);
CREATE TABLE products (...);
CREATE TABLE payments (...);
-- 垂直分库后
CREATE DATABASE user_db;
CREATE TABLE users (...);
CREATE DATABASE order_db;
CREATE TABLE orders (...);
CREATE DATABASE product_db;
CREATE TABLE products (...);
CREATE DATABASE payment_db;
CREATE TABLE payments (...);
3.1.2 水平分表（按用户ID分片）
// 分片算法实现
@Component
public class OrderShardingAlgorithm implements StandardShardingAlgorithm<Long> {
@Override
public String doSharding(Collection<String> availableTargetNames, ShardingValue<Long> shardingValue) {
Long orderId = shardingValue.getValue();
// 根据订单ID取模分片
int shardIndex = (int) (orderId % 4); // 4个分片
return "orders_" + shardIndex;
}
@Override
public Properties getProps() {
return null;
}
@Override
public void init(Properties properties) {
}
}
// 使用ShardingSphere进行分片配置
@Configuration
public class ShardingConfig {
@Bean
public DataSource dataSource() {
Map<String, DataSource> dataSourceMap = new HashMap<>();
// 配置数据源
HikariDataSource ds0 = new HikariDataSource();
ds0.setJdbcUrl("jdbc:mysql://localhost:3306/order_db_0");
ds0.setUsername("root");
ds0.setPassword("password");
dataSourceMap.put("ds0", ds0);
HikariDataSource ds1 = new HikariDataSource();
ds1.setJdbcUrl("jdbc:mysql://localhost:3306/order_db_1");
ds1.setUsername("root");
ds1.setPassword("password");
dataSourceMap.put("ds1", ds1);
// 分片规则配置
ShardingRuleConfiguration shardingRuleConfig = new ShardingRuleConfiguration();
// 订单表分片规则
TableRuleConfiguration orderTableRuleConfig = new TableRuleConfiguration();
orderTableRuleConfig.setLogicTable("orders");
orderTableRuleConfig.setActualDataNodes("ds${0..1}.orders_${0..3}"); // 2个库，每个库4张表
// 分片算法
StandardShardingAlgorithmConfiguration shardingAlgorithmConfig = new StandardShardingAlgorithmConfiguration();
shardingAlgorithmConfig.setType("INLINE");
Properties props = new Properties();
props.setProperty("algorithm-expression", "ds${order_id % 2}.orders_${order_id % 4}");
shardingAlgorithmConfig.setProps(props);
orderTableRuleConfig.setTableShardingAlgorithmConfig(shardingAlgorithmConfig);
// 主键生成策略
orderTableRuleConfig.setKeyGeneratorConfig(new KeyGeneratorConfiguration("SNOWFLAKE", "order_id", new Properties()));
shardingRuleConfig.getTables().add(orderTableRuleConfig);
// 创建ShardingSphere数据源
return ShardingSphereDataSourceFactory.createDataSource(dataSourceMap,
Arrays.asList(shardingRuleConfig), new Properties());
}
}
3.1.3 分片键选择策略
- 用户ID：适合用户相关的数据，便于按用户查询
- 时间：适合时间序列数据，便于按时间范围查询
- 地理位置：适合地域相关的数据
- 业务ID：适合业务相关的数据
分片键选择示例：
// 订单表按用户ID分片
// 优点：查询用户订单时，只需访问一个分片
// 缺点：单个用户数据量过大时，分片不均匀
// 订单表按时间分片
// 优点：数据分布均匀，便于按时间范围查询
// 缺点：跨分片查询复杂
// 订单表按订单ID分片
// 优点：数据分布均匀
// 缺点：查询用户订单时需要跨多个分片
3.2 数据库中间件
3.2.1 ShardingSphere ShardingSphere是Apache顶级项目，提供数据分片、读写分离、分布式事务等功能。
# ShardingSphere配置示例
spring:
shardingsphere:
datasource:
names: ds0,ds1
ds0:
type: com.zaxxer.hikari.HikariDataSource
driver-class-name: com.mysql.cj.jdbc.Driver
jdbc-url: jdbc:mysql://localhost:3306/order_db_0
username: root
password: password
ds1:
type: com.zaxxer.hikari.HikariDataSource
driver-class-name: com.mysql.cj.jdbc.Driver
jdbc-url: jdbc:mysql://localhost:3306/order_db_1
username: root
password: password
rules:
sharding:
tables:
orders:
actual-data-nodes: ds${0..1}.orders_${0..3}
table-strategy:
standard:
sharding-column: order_id
sharding-algorithm-name: order-table-inline
key-generate-strategy:
column: order_id
key-generator-name: snowflake
sharding-algorithms:
order-table-inline:
type: INLINE
props:
algorithm-expression: ds${order_id % 2}.orders_${order_id % 4}
key-generators:
snowflake:
type: SNOWFLAKE
props:
worker-id: 123
3.2.2 Vitess Vitess是YouTube开源的数据库中间件，专为大规模MySQL集群设计。
# Vitess部署示例
# 1. 启动Vitess集群
vtctld --port=15000 --service_map=grpc-vtctld &
vtgate --port=15001 --service_map=grpc-vtgate &
vttablet --port=15002 --service_map=grpc-vttablet &
# 2. 创建分片
vtctlclient CreateKeyspace --sharding_column=customer_id customer
# 3. 添加分片
vtctlclient AddShard --sharding_column=customer_id customer/0
vtctlclient AddShard --sharding_column=customer_id customer/1
# 4. 启动Vitess客户端
vtgateclient --server=localhost:15001 --keyspace=customer
3.3 数据库集群架构
3.3.1 主从复制集群
-- 主库配置（my.cnf）
[mysqld]
server-id = 1
log-bin = mysql-bin
binlog-format = ROW
binlog-do-db = mydb
sync_binlog = 1
innodb_flush_log_at_trx_commit = 1
-- 从库配置（my.cnf）
[mysqld]
server-id = 2
relay-log = mysql-relay-bin
read-only = 1
3.3.2 MHA（Master High Availability） MHA是MySQL高可用解决方案，能在主库故障时自动切换。
# MHA部署步骤
# 1. 安装MHA
sudo apt-get install mha4mysql-manager mha4mysql-node
# 2. 配置MHA
cat > /etc/mha/app1.cnf << EOF
[server default]
user=root
password=password
ping_interval=3
master_binlog_dir=/var/lib/mysql
[server1]
hostname=master-host
candidate_master=1
[server2]
hostname=slave1-host
candidate_master=1
[server3]
hostname=slave2-host
no_master=1
EOF
# 3. 检查配置
masterha_check_ssh --conf=/etc/mha/app1.cnf
masterha_check_repl --conf=/etc/mha/app1.cnf
# 4. 启动MHA
masterha_manager --conf=/etc/mha/app1.cnf &
3.3.3 Galera Cluster（多主集群） Galera Cluster提供真正的多主复制，所有节点都可读写。
-- Galera Cluster配置（my.cnf）
[mysqld]
wsrep_on=ON
wsrep_provider=/usr/lib/galera/libgalera_smm.so
wsrep_cluster_address=gcomm://node1,node2,node3
wsrep_cluster_name=galera_cluster
wsrep_node_address=node1
wsrep_node_name=node1
wsrep_sst_method=rsync
wsrep_sst_auth=root:password
binlog_format=ROW
default_storage_engine=InnoDB
innodb_autoinc_lock_mode=2
3.4 数据库监控与告警
3.4.1 Prometheus + Grafana监控
# prometheus.yml 配置
global:
scrape_interval: 15s
scrape_configs:
- job_name: 'mysql'
static_configs:
- targets: ['mysql-exporter:9104']
metrics_path: /metrics
params:
collect[]:
- global_status
- global_variables
- slave_status
- innodb_metrics
- performance_schema.tablelocks
- performance_schema.table_io_waits_summary_by_table
- performance_schema.table_io_waits_summary_by_index_usage
- performance_schema.file_summary_by_instance
- performance_schema.events_waits_summary_by_instance
- performance_schema.events_waits_summary_by_thread_by_event_name
- performance_schema.events_waits_summary_global_by_event_name
- performance_schema.file_summary_by_event_name
- performance_schema.table_io_waits_summary_by_table
- performance_schema.table_io_waits_summary_by_index_usage
- performance_schema.table_lock_waits_summary_by_table
- performance_schema.table_lock_waits_summary_by_index_usage
- performance_schema.table_io_waits_summary_by_table
- performance_schema.table_io_waits_summary_by_index_usage
- performance_schema.table_lock_waits_summary_by_table
- performance_schema.table_lock_waits_summary_by_index_usage
- performance_schema.table_io_waits_summary_by_table
- performance_schema.table_io_waits_summary_by_index_usage
- performance_schema.table_lock_waits_summary_by_table
- performance_schema.table_lock_waits_summary_by_index_usage
3.4.2 自定义监控脚本
#!/usr/bin/env python3
# mysql_monitor.py
import pymysql
import time
import json
import requests
class MySQLMonitor:
def __init__(self, host, user, password, port=3306):
self.conn = pymysql.connect(
host=host,
user=user,
password=password,
port=port,
charset='utf8mb4'
)
def get_status(self):
"""获取MySQL状态信息"""
with self.conn.cursor() as cursor:
cursor.execute("SHOW GLOBAL STATUS")
status = dict(cursor.fetchall())
cursor.execute("SHOW GLOBAL VARIABLES")
variables = dict(cursor.fetchall())
cursor.execute("SHOW SLAVE STATUS")
slave_status = cursor.fetchone()
return {
'status': status,
'variables': variables,
'slave_status': slave_status
}
def check_slow_queries(self, threshold=1):
"""检查慢查询"""
with self.conn.cursor() as cursor:
cursor.execute("SHOW GLOBAL STATUS LIKE 'Slow_queries'")
slow_queries = cursor.fetchone()[1]
cursor.execute("SHOW GLOBAL STATUS LIKE 'Uptime'")
uptime = cursor.fetchone()[1]
# 计算每秒慢查询数
slow_per_second = int(slow_queries) / int(uptime)
return {
'slow_queries': int(slow_queries),
'slow_per_second': slow_per_second,
'threshold': threshold,
'alert': slow_per_second > threshold
}
def check_connections(self):
"""检查连接数"""
with self.conn.cursor() as cursor:
cursor.execute("SHOW GLOBAL STATUS LIKE 'Threads_connected'")
connected = cursor.fetchone()[1]
cursor.execute("SHOW GLOBAL VARIABLES LIKE 'max_connections'")
max_connections = cursor.fetchone()[1]
usage = int(connected) / int(max_connections)
return {
'connected': int(connected),
'max_connections': int(max_connections),
'usage': usage,
'alert': usage > 0.8 # 80%使用率告警
}
def send_alert(self, message, level='warning'):
"""发送告警"""
# 这里可以集成钉钉、企业微信、邮件等
alert_data = {
'message': message,
'level': level,
'timestamp': time.time(),
'host': self.conn.host
}
# 示例：发送到Webhook
webhook_url = 'https://your-webhook-url.com/alerts'
try:
requests.post(webhook_url, json=alert_data, timeout=5)
print(f"Alert sent: {message}")
except Exception as e:
print(f"Failed to send alert: {e}")
def main():
monitor = MySQLMonitor(
host='localhost',
user='root',
password='password'
)
while True:
try:
# 检查慢查询
slow_check = monitor.check_slow_queries()
if slow_check['alert']:
monitor.send_alert(
f"慢查询告警: {slow_check['slow_per_second']:.2f} QPS",
'critical'
)
# 检查连接数
conn_check = monitor.check_connections()
if conn_check['alert']:
monitor.send_alert(
f"连接数告警: {conn_check['connected']}/{conn_check['max_connections']} ({conn_check['usage']:.1%})",
'warning'
)
# 获取详细状态
status = monitor.get_status()
# 输出监控信息
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] "
f"慢查询: {slow_check['slow_queries']} "
f"连接数: {conn_check['connected']}/{conn_check['max_connections']} "
f"QPS: {status['status'].get('Queries', 0)}")
time.sleep(60) # 每分钟检查一次
except Exception as e:
print(f"监控异常: {e}")
time.sleep(30)
if __name__ == '__main__':
main()
四、实战案例：电商大促系统设计
4.1 系统架构设计
用户请求 → Nginx负载均衡 → 应用服务器集群 → 缓存层(Redis集群) → 数据库层(MySQL集群)
4.2 关键技术实现
4.2.1 库存扣减（防止超卖）
// 使用Redis Lua脚本保证原子性
@Component
public class InventoryService {
@Autowired
private RedisTemplate<String, String> redisTemplate;
// Lua脚本：原子性扣减库存
private static final String DECREASE_INVENTORY_SCRIPT =
"local stock = redis.call('get', KEYS[1]) " +
"if not stock or tonumber(stock) <= 0 then " +
" return -1 " +
"end " +
"redis.call('decr', KEYS[1]) " +
"return tonumber(stock) - 1";
public boolean decreaseInventory(Long productId, int quantity) {
String key = "inventory:" + productId;
// 执行Lua脚本
Long result = redisTemplate.execute(
new DefaultRedisScript<>(DECREASE_INVENTORY_SCRIPT, Long.class),
Collections.singletonList(key)
);
if (result == null || result < 0) {
return false;
}
// 异步同步到数据库
asyncUpdateDatabase(productId, quantity);
return true;
}
@Async
public void asyncUpdateDatabase(Long productId, int quantity) {
// 异步更新数据库，避免阻塞
// 使用消息队列或定时任务批量更新
}
}
4.2.2 限流保护
// 使用Guava RateLimiter实现限流
@Component
public class RateLimitService {
// 每秒允许100个请求
private final RateLimiter rateLimiter = RateLimiter.create(100.0);
public boolean tryAcquire() {
return rateLimiter.tryAcquire();
}
}
// 使用Sentinel实现分布式限流
@Configuration
public class SentinelConfig {
@Bean
public SentinelResourceAspect sentinelResourceAspect() {
return new SentinelResourceAspect();
}
}
@Service
public class OrderService {
@SentinelResource(value = "createOrder",
blockHandler = "handleBlock",
fallback = "handleFallback")
public Order createOrder(Order order) {
// 创建订单逻辑
return orderRepository.save(order);
}
// 限流处理
public Order handleBlock(Order order, BlockException ex) {
throw new RuntimeException("请求过于频繁，请稍后重试");
}
// 降级处理
public Order handleFallback(Order order, Throwable t) {
// 返回缓存数据或默认值
return new Order();
}
}
4.2.3 分布式事务
// 使用Seata实现分布式事务
@Configuration
public class SeataConfig {
@Bean
public GlobalTransactionScanner globalTransactionScanner() {
return new GlobalTransactionScanner("order-service", "order-group");
}
}
@Service
public class OrderService {
@Autowired
private InventoryService inventoryService;
@Autowired
private PaymentService paymentService;
@GlobalTransactional(name = "createOrder", rollbackFor = Exception.class)
public Order createOrderWithTransaction(Order order) {
// 1. 扣减库存
boolean inventoryResult = inventoryService.decreaseInventory(
order.getProductId(),
order.getQuantity()
);
if (!inventoryResult) {
throw new RuntimeException("库存不足");
}
// 2. 创建订单
Order savedOrder = orderRepository.save(order);
// 3. 支付处理
boolean paymentResult = paymentService.processPayment(savedOrder);
if (!paymentResult) {
throw new RuntimeException("支付失败");
}
return savedOrder;
}
}
4.3 性能压测与调优
4.3.1 压测工具配置
# 使用JMeter进行压测
# 1. 创建测试计划
# 2. 配置线程组：1000个线程，循环100次
# 3. 配置HTTP请求：POST /api/orders
# 4. 配置监听器：查看结果树、聚合报告、响应时间图
# 使用wrk进行压测
wrk -t12 -c400 -d30s --latency http://localhost:8080/api/orders
4.3.2 性能指标监控
// 使用Micrometer监控性能指标
@Configuration
public class MetricsConfig {
@Bean
public MeterRegistryCustomizer<MeterRegistry> metricsCommonTags() {
return registry -> registry.config().commonTags("application", "order-service");
}
@Bean
public TimedAspect timedAspect(MeterRegistry registry) {
return new TimedAspect(registry);
}
}
@Service
public class OrderService {
@Timed(value = "order.create.time", description = "创建订单耗时")
@Counted(value = "order.create.count", description = "创建订单次数")
public Order createOrder(Order order) {
// 业务逻辑
return orderRepository.save(order);
}
}
五、总结与最佳实践
5.1 高并发处理原则
- 分层优化：从数据库、应用、架构三个层面逐层优化
- 缓存优先：能用缓存解决的，不要直接访问数据库
- 异步处理：耗时操作异步化，避免阻塞主线程
- 水平扩展：通过分库分表、集群架构实现水平扩展
- 监控告警：建立完善的监控体系，及时发现问题
5.2 不同场景下的策略选择
| 场景 | 推荐策略 | 关键技术 |
|---|---|---|
| 读多写少 | 读写分离 + 缓存 | Redis + MySQL主从 |
| 写多读少 | 分库分表 + 消息队列 | ShardingSphere + RabbitMQ |
| 实时性要求高 | 本地缓存 + 异步写库 | Caffeine + 异步任务 |
| 数据一致性要求高 | 分布式事务 + 强一致性 | Seata + Raft协议 |
5.3 持续优化建议
- 定期慢查询分析：每周分析慢查询日志，优化SQL
- 索引定期维护：定期检查索引使用情况，删除无用索引
- 容量规划：根据业务增长预测，提前规划扩容
- 故障演练：定期进行故障演练，验证高可用方案
- 技术选型更新：关注MySQL新版本特性，及时升级
5.4 常见误区与避坑指南
- 过度优化：不要在业务早期就进行复杂的分库分表
- 缓存穿透：缓存空值或使用布隆过滤器防止缓存穿透
- 缓存雪崩：设置随机过期时间，避免大量缓存同时失效
- 数据库连接泄漏：确保连接池正确关闭，使用连接泄漏检测
- 事务过大：避免在事务中执行耗时操作，保持事务短小
六、进阶学习资源
6.1 推荐书籍
- 《高性能MySQL》
- 《MySQL技术内幕：InnoDB存储引擎》
- 《数据密集型应用系统设计》
6.2 开源项目
- ShardingSphere：Apache顶级项目，提供数据分片、读写分离等功能
- Vitess：YouTube开源的数据库中间件
- ProxySQL：MySQL代理，支持查询路由、缓存、重写等
- Percona Toolkit：MySQL性能分析和优化工具集
6.3 在线课程
- MySQL官方文档：https://dev.mysql.com/doc/
- Percona University：https://university.percona.com/
- MySQL Performance Blog：https://www.percona.com/blog/
七、结语
MySQL高并发处理是一个系统工程，需要从数据库优化、应用层优化、架构设计三个层面综合考虑。没有银弹，只有最适合当前业务场景的方案。在实际项目中，建议从简单的优化开始，随着业务增长逐步引入更复杂的架构方案。
记住，优化是一个持续的过程。定期监控、分析、调整，才能让系统在高并发场景下保持稳定、高效。希望本文能为你提供有价值的参考，助你在高并发处理的道路上走得更远。
附录：常用命令速查表
-- 查看当前连接
SHOW PROCESSLIST;
-- 查看索引
SHOW INDEX FROM table_name;
-- 查看表结构
DESC table_name;
-- 查看慢查询日志
SHOW VARIABLES LIKE 'slow_query_log%';
-- 查看InnoDB状态
SHOW ENGINE INNODB STATUS;
-- 查看复制状态
SHOW SLAVE STATUS;
-- 查看缓冲池状态
SHOW ENGINE INNODB STATUS\G
监控指标参考：
- QPS（每秒查询数）：> 1000 需要关注
- TPS（每秒事务数）：> 500 需要关注
- 连接数：> 80% max_connections 需要关注
- 慢查询数：> 10/分钟 需要关注
- 缓冲池命中率：> 95% 为佳
通过本文的系统学习和实践，相信你已经掌握了MySQL高并发处理的核心技能。在实际工作中，结合具体业务场景，灵活运用这些策略，一定能构建出稳定、高效的数据库系统。