MySQL架构师之路：海量数据下的性能优化与分库分表实战

第一章：数据库性能优化全面思路

1.1 千万级数据表的优化思路

面对面试官提出的"单表1千万数据，未来1年增长500万"场景，正确的回答思路：

1.2 不分库分表优化策略

软优化：数据库层面的优化

SQL优化与索引优化：

sql

-- 1. 分析慢查询 SHOW VARIABLES LIKE 'slow_query_log%'; SET GLOBAL slow_query_log = ON; SET GLOBAL long_query_time = 1; -- 2. 使用EXPLAIN分析执行计划 EXPLAIN SELECT * FROM orders WHERE user_id = 1000 AND created_at > '2023-01-01' ORDER BY amount DESC; -- 3. 创建复合索引优化查询 CREATE INDEX idx_user_created_amount ON orders(user_id, created_at, amount);

数据库参数调优示例：

ini

# my.cnf 优化配置 [mysqld] # 内存配置 innodb_buffer_pool_size = 16G # 设置为物理内存的70%-80% innodb_buffer_pool_instances = 8 # 提高并发性 # 连接配置 max_connections = 1000 thread_cache_size = 100 wait_timeout = 600 # 日志配置 innodb_log_file_size = 2G innodb_log_buffer_size = 64M sync_binlog = 1 innodb_flush_log_at_trx_commit = 2 # 查询缓存（MySQL 8.0已移除） # query_cache_size = 128M # query_cache_type = 1

表结构优化：

sql

-- 垂直分表：大字段分离 -- 原表 CREATE TABLE products ( id BIGINT PRIMARY KEY, name VARCHAR(200), description TEXT, -- 大字段，访问频率低 specifications JSON, -- 大字段 price DECIMAL(10,2), created_at TIMESTAMP ); -- 优化后：主表 + 详情表 CREATE TABLE products ( id BIGINT PRIMARY KEY, name VARCHAR(200), price DECIMAL(10,2), created_at TIMESTAMP, INDEX idx_name_price(name, price) ); CREATE TABLE product_details ( product_id BIGINT PRIMARY KEY, description TEXT, specifications JSON, FOREIGN KEY (product_id) REFERENCES products(id) );

硬优化：系统硬件升级

yaml

硬件升级方案： 1. SSD存储：将机械硬盘升级为NVMe SSD - 随机IOPS从100提升到50000+ - 顺序读写从100MB/s提升到3000MB/s+ 2. 内存升级：从32GB升级到128GB+ - 增加InnoDB Buffer Pool容量 - 减少磁盘IO次数 3. CPU升级：增加核心数 - 支持更高并发连接 - 并行查询处理能力提升 4. 网络升级：千兆升级到万兆 - 减少数据传输延迟 - 支持更大带宽

1.3 架构优化策略

读写分离方案：

java

// Spring Boot + MyBatis 读写分离配置 @Configuration public class DataSourceConfig { @Bean @ConfigurationProperties("spring.datasource.master") public DataSource masterDataSource() { return DataSourceBuilder.create().build(); } @Bean @ConfigurationProperties("spring.datasource.slave") public DataSource slaveDataSource() { return DataSourceBuilder.create().build(); } @Bean public DataSource dynamicDataSource() { Map<Object, Object> targetDataSources = new HashMap<>(); targetDataSources.put("master", masterDataSource()); targetDataSources.put("slave", slaveDataSource()); DynamicDataSource dataSource = new DynamicDataSource(); dataSource.setTargetDataSources(targetDataSources); dataSource.setDefaultTargetDataSource(masterDataSource()); return dataSource; } } // 使用AOP实现读写分离 @Aspect @Component public class ReadWriteSeparateAspect { @Around("@annotation(org.springframework.transaction.annotation.Transactional)") public Object switchDataSource(ProceedingJoinPoint point) throws Throwable { MethodSignature signature = (MethodSignature) point.getSignature(); Transactional transactional = signature.getMethod() .getAnnotation(Transactional.class); boolean isReadOnly = transactional.readOnly(); if (isReadOnly) { DynamicDataSourceContextHolder.setDataSourceType("slave"); } else { DynamicDataSourceContextHolder.setDataSourceType("master"); } try { return point.proceed(); } finally { DynamicDataSourceContextHolder.clearDataSourceType(); } } }

缓存策略优化：

java

// 多级缓存架构 @Component public class MultiLevelCacheService { @Autowired private RedisTemplate<String, Object> redisTemplate; @Autowired private CaffeineCacheManager caffeineCacheManager; // 一级缓存：本地缓存（Caffeine） private Cache<String, Object> localCache = Caffeine.newBuilder() .maximumSize(10000) .expireAfterWrite(5, TimeUnit.MINUTES) .build(); // 二级缓存：分布式缓存（Redis） // 三级缓存：数据库 public Object getWithMultiLevel(String key, Supplier<Object> loader) { // 1. 查询本地缓存 Object value = localCache.getIfPresent(key); if (value != null) { return value; } // 2. 查询Redis缓存 value = redisTemplate.opsForValue().get(key); if (value != null) { localCache.put(key, value); return value; } // 3. 查询数据库 synchronized (key.intern()) { // 双重检查锁 value = redisTemplate.opsForValue().get(key); if (value == null) { value = loader.get(); if (value != null) { // 写入Redis（设置过期时间） redisTemplate.opsForValue() .set(key, value, 30, TimeUnit.MINUTES); // 写入本地缓存 localCache.put(key, value); } } } return value; } }

第二章：分库分表深度解析

2.1 分库分表的六大核心问题

问题一：跨节点Join关联查询

解决方案对比：

sql

-- 方案1：字段冗余（空间换时间） CREATE TABLE orders ( id BIGINT, user_id BIGINT, user_name VARCHAR(100), -- 冗余用户姓名 user_avatar VARCHAR(255), -- 冗余用户头像 amount DECIMAL(10,2), PRIMARY KEY (id) ); -- 方案2：广播表（小表复制） -- 字典表在所有分片中都存在完整副本 CREATE TABLE dictionary ( id INT PRIMARY KEY, type VARCHAR(50), code VARCHAR(50), name VARCHAR(100) ) ENGINE=InnoDB; -- 方案3：数据异构（ES查询） -- 将订单数据同步到Elasticsearch PUT /orders/_mapping { "properties": { "order_id": {"type": "long"}, "user_id": {"type": "long"}, "shop_id": {"type": "long"}, "amount": {"type": "double"}, "created_at": {"type": "date"} } }

问题二：分布式事务

解决方案实现：

java

// TCC（Try-Confirm-Cancel）模式实现 @Service public class OrderTccService { @Transactional public boolean createOrder(OrderDTO order) { // 第一阶段：Try（尝试执行业务） boolean inventorySuccess = inventoryTccService.tryLock(order.getProductId(), order.getQuantity()); boolean couponSuccess = couponTccService.tryLock(order.getUserId(), order.getCouponId()); if (!inventorySuccess || !couponSuccess) { // 有任何失败，触发Cancel inventoryTccService.cancel(order.getProductId(), order.getQuantity()); couponTccService.cancel(order.getUserId(), order.getCouponId()); return false; } // 第二阶段：Confirm（确认执行） try { orderService.confirmCreate(order); inventoryTccService.confirm(order.getProductId(), order.getQuantity()); couponTccService.confirm(order.getUserId(), order.getCouponId()); return true; } catch (Exception e) { // 失败触发Cancel inventoryTccService.cancel(order.getProductId(), order.getQuantity()); couponTccService.cancel(order.getUserId(), order.getCouponId()); throw e; } } }

问题三：分页、排序、函数计算

Sharding-JDBC解决方案：

java

// Sharding-JDBC自动处理跨分片查询 @Repository public class OrderRepository { // 分页查询：Sharding-JDBC会自动合并多个分片的结果 @Select("SELECT * FROM orders WHERE user_id = #{userId} " + "ORDER BY created_at DESC LIMIT #{offset}, #{limit}") List<Order> findUserOrders(@Param("userId") Long userId, @Param("offset") int offset, @Param("limit") int limit); // 聚合函数：Sharding-JDBC会处理SUM、COUNT等函数的合并 @Select("SELECT COUNT(*) as total, SUM(amount) as total_amount " + "FROM orders WHERE created_at >= #{startDate}") Map<String, Object> getOrderStats(@Param("startDate") Date startDate); }

问题四：全局主键生成

分布式ID生成方案对比：

java

// 方案1：Snowflake算法（推荐） @Component public class SnowflakeIdGenerator { private final Snowflake snowflake; public SnowflakeIdGenerator() { // 数据中心ID + 机器ID（可以通过配置中心获取） long datacenterId = getDatacenterId(); long machineId = getMachineId(); this.snowflake = new Snowflake(datacenterId, machineId); } public Long nextId() { return snowflake.nextId(); } // 解析Snowflake ID public Map<String, Long> parseId(Long id) { long timestamp = (id >> 22) + 1288834974657L; long datacenterId = (id >> 17) & 0x1F; long machineId = (id >> 12) & 0x1F; long sequence = id & 0xFFF; return Map.of( "timestamp", timestamp, "datacenterId", datacenterId, "machineId", machineId, "sequence", sequence ); } } // 方案2：号段模式 @Service public class SegmentIdGenerator { @Autowired private RedisTemplate<String, String> redisTemplate; public Long nextId(String bizType) { String key = "id:segment:" + bizType; // 使用Lua脚本保证原子性 String luaScript = "local current = redis.call('get', KEYS[1]) " + "if current == false then " + " redis.call('set', KEYS[1], 1000) " + " return 1 " + "else " + " local nextId = redis.call('incr', KEYS[1]) " + " return nextId " + "end"; Long result = redisTemplate.execute( new DefaultRedisScript<>(luaScript, Long.class), Collections.singletonList(key) ); return result; } }

2.2 垂直分库分表实战

垂直分表案例：电商商品表优化

sql

-- 优化前：单表包含所有字段 CREATE TABLE products ( id BIGINT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(200), category_id INT, price DECIMAL(10,2), stock INT, description TEXT, -- 大字段，访问频率低 specifications JSON, -- 大字段，结构复杂 images JSON, -- 大字段，存储图片信息 created_at TIMESTAMP, updated_at TIMESTAMP, INDEX idx_category_price(category_id, price) ); -- 优化后：垂直分表 -- 主表：存储高频访问的核心字段 CREATE TABLE products ( id BIGINT PRIMARY KEY, name VARCHAR(200), category_id INT, price DECIMAL(10,2), stock INT, status TINYINT, created_at TIMESTAMP, updated_at TIMESTAMP, INDEX idx_category_price(category_id, price), INDEX idx_status_created(status, created_at) ); -- 详情表：存储大字段和低频访问字段 CREATE TABLE product_details ( product_id BIGINT PRIMARY KEY, description TEXT, specifications JSON, images JSON, seo_title VARCHAR(255), seo_keywords VARCHAR(500), seo_description TEXT, FOREIGN KEY (product_id) REFERENCES products(id) ); -- 扩展表：存储业务扩展字段 CREATE TABLE product_extensions ( product_id BIGINT PRIMARY KEY, sales_count INT DEFAULT 0, view_count INT DEFAULT 0, favorite_count INT DEFAULT 0, avg_rating DECIMAL(3,2), tags JSON, FOREIGN KEY (product_id) REFERENCES products(id) );

垂直分库案例：微服务架构

yaml

# 垂直分库配置示例 datasources: # 用户服务数据库 user-service: url: jdbc:mysql://localhost:3306/user_db username: root password: password driver-class-name: com.mysql.cj.jdbc.Driver # 商品服务数据库 product-service: url: jdbc:mysql://localhost:3307/product_db username: root password: password driver-class-name: com.mysql.cj.jdbc.Driver # 订单服务数据库 order-service: url: jdbc:mysql://localhost:3308/order_db username: root password: password driver-class-name: com.mysql.cj.jdbc.Driver # 支付服务数据库 payment-service: url: jdbc:mysql://localhost:3309/payment_db username: root password: password driver-class-name: com.mysql.cj.jdbc.Driver

2.3 水平分库分表实战

水平分表策略实现

java

// 基于用户ID的哈希分表策略 @Component public class UserShardingStrategy implements PreciseShardingAlgorithm<Long> { private static final String TABLE_PREFIX = "user_"; private static final int TABLE_COUNT = 8; // 分成8张表 @Override public String doSharding(Collection<String> tableNames, PreciseShardingValue<Long> shardingValue) { Long userId = shardingValue.getValue(); // 哈希取模算法 int tableSuffix = Math.abs(userId.hashCode()) % TABLE_COUNT; return TABLE_PREFIX + tableSuffix; } } // 基于时间范围的分表策略 @Component public class TimeRangeShardingStrategy implements RangeShardingAlgorithm<Date> { private static final String TABLE_PREFIX = "order_"; @Override public Collection<String> doSharding(Collection<String> tableNames, RangeShardingValue<Date> shardingValue) { Range<Date> range = shardingValue.getValueRange(); List<String> result = new ArrayList<>(); // 获取时间范围 Date lower = range.lowerEndpoint(); Date upper = range.upperEndpoint(); // 计算涉及的所有月份 Calendar calendar = Calendar.getInstance(); calendar.setTime(lower); while (!calendar.getTime().after(upper)) { int year = calendar.get(Calendar.YEAR); int month = calendar.get(Calendar.MONTH) + 1; String tableName = String.format("%s%d_%02d", TABLE_PREFIX, year, month); if (tableNames.contains(tableName)) { result.add(tableName); } calendar.add(Calendar.MONTH, 1); } return result; } }

Sharding-JDBC配置示例

yaml

# application-sharding.yml spring: shardingsphere: datasource: names: ds0,ds1,ds2,ds3 ds0: type: com.zaxxer.hikari.HikariDataSource driver-class-name: com.mysql.cj.jdbc.Driver jdbc-url: jdbc:mysql://localhost:3306/db0 username: root password: password ds1: type: com.zaxxer.hikari.HikariDataSource driver-class-name: com.mysql.cj.jdbc.Driver jdbc-url: jdbc:mysql://localhost:3306/db1 username: root password: password # ... ds2, ds3 配置类似 sharding: tables: # 订单表分片配置 orders: actual-data-nodes: ds$->{0..3}.orders_$->{0..15} table-strategy: standard: sharding-column: order_id precise-algorithm-class-name: com.example.OrderShardingAlgorithm key-generator: column: order_id type: SNOWFLAKE # 订单项表分片配置（绑定表） order_items: actual-data-nodes: ds$->{0..3}.order_items_$->{0..15} table-strategy: standard: sharding-column: order_id precise-algorithm-class-name: com.example.OrderShardingAlgorithm # 用户表分片配置 users: actual-data-nones: ds$->{0..3}.users_$->{0..7} table-strategy: standard: sharding-column: user_id precise-algorithm-class-name: com.example.UserShardingAlgorithm key-generator: column: user_id type: SNOWFLAKE # 广播表配置 broadcast-tables: dict_config, dict_type # 绑定表配置 binding-tables: - orders,order_items props: sql: show: true

2.4 分库分表后的查询优化

多维度查询解决方案

java

// 方案1：冗余字段 + 多份存储 @Service public class OrderQueryService { @Autowired private OrderRepository orderRepository; @Autowired private ElasticsearchRestTemplate elasticsearchTemplate; @Autowired private RedisTemplate<String, Object> redisTemplate; /** * 用户查询自己的订单（基于user_id分片） */ public Page<Order> getUserOrders(Long userId, Pageable pageable) { // 直接通过分片键查询，效率最高 return orderRepository.findByUserId(userId, pageable); } /** * 商家查询店铺订单（需要多维度查询） */ public Page<Order> getShopOrders(Long shopId, Pageable pageable) { // 方案1：通过ES查询（数据异构） NativeSearchQuery searchQuery = new NativeSearchQueryBuilder() .withQuery(QueryBuilders.termQuery("shop_id", shopId)) .withPageable(pageable) .build(); SearchHits<OrderEsDTO> searchHits = elasticsearchTemplate .search(searchQuery, OrderEsDTO.class); // 获取订单ID列表，然后从数据库查询完整数据 List<Long> orderIds = searchHits.getSearchHits().stream() .map(hit -> hit.getContent().getOrderId()) .collect(Collectors.toList()); return orderRepository.findByIdIn(orderIds, pageable); } /** * 复杂条件查询 */ public List<Order> complexQuery(OrderQueryDTO query) { // 方案2：使用缓存层聚合 String cacheKey = buildCacheKey(query); List<Order> cachedResult = (List<Order>) redisTemplate .opsForValue().get(cacheKey); if (cachedResult != null) { return cachedResult; } // 方案3：并行查询多个分片 List<CompletableFuture<List<Order>>> futures = new ArrayList<>(); for (int i = 0; i < 4; i++) { // 假设4个分片 int shardIndex = i; futures.add(CompletableFuture.supplyAsync(() -> { return orderRepository.findByShard(query, shardIndex); })); } // 合并结果 List<Order> result = futures.stream() .map(CompletableFuture::join) .flatMap(List::stream) .sorted(Comparator.comparing(Order::getCreatedAt).reversed()) .collect(Collectors.toList()); // 缓存结果 redisTemplate.opsForValue().set(cacheKey, result, 5, TimeUnit.MINUTES); return result; } }

分页查询优化

sql

-- 传统分页的问题：OFFSET越大性能越差 SELECT * FROM orders ORDER BY created_at DESC LIMIT 1000000, 20; -- 优化方案1：游标分页（基于ID） SELECT * FROM orders WHERE id > ? AND created_at >= ? AND created_at <= ? ORDER BY id ASC LIMIT 20; -- 优化方案2：延迟关联 SELECT * FROM orders t JOIN ( SELECT id FROM orders WHERE user_id = ? ORDER BY created_at DESC LIMIT 1000000, 20 ) AS tmp ON t.id = tmp.id; -- 优化方案3：业务层分页（推荐） -- 应用层维护分页状态，每次查询固定数量的记录

第三章：分库分表扩容方案

3.1 二次扩容实施方案

方案一：在线扩容（不停机）

具体实现代码：

java

@Component public class DatabaseMigrationService { @Autowired private DataSource oldDataSource; @Autowired private DataSource newDataSource; @Autowired private KafkaTemplate<String, String> kafkaTemplate; /** * 数据迁移任务 */ @Scheduled(fixedDelay = 60000) // 每分钟执行一次 public void migrateData() { // 1. 查询需要迁移的数据（基于更新时间） List<Order> orders = jdbcTemplate.query( oldDataSource, "SELECT * FROM orders WHERE gmt_modified > ? LIMIT 1000", new Object[]{lastMigrationTime}, new BeanPropertyRowMapper<>(Order.class) ); // 2. 批量写入新数据库 if (!orders.isEmpty()) { batchInsert(newDataSource, orders); // 3. 更新迁移进度 updateMigrationProgress(orders.get(orders.size() - 1).getGmtModified()); } } /** * 双写机制 */ @Transactional public void createOrderWithDoubleWrite(Order order) { // 写入旧数据库 orderRepository.save(order); // 写入新数据库 try { newOrderRepository.save(order); } catch (Exception e) { // 记录失败，后续补偿 log.error("双写失败，记录到消息队列", e); kafkaTemplate.send("db-migration-fail", order.getId().toString()); } // 记录变更到消息队列（用于数据校验） kafkaTemplate.send("db-change-log", buildChangeLog("INSERT", order)); } /** * 数据校验 */ public void verifyData() { // 对比新旧数据库数据 String sql = "SELECT COUNT(*) as count, " + "MD5(GROUP_CONCAT(id ORDER BY id)) as checksum " + "FROM orders WHERE gmt_created >= ?"; Map<String, Object> oldStats = jdbcTemplate.queryForMap( oldDataSource, sql, migrationStartTime); Map<String, Object> newStats = jdbcTemplate.queryForMap( newDataSource, sql, migrationStartTime); if (!oldStats.equals(newStats)) { log.error("数据校验失败，需要人工干预"); sendAlert("数据不一致警报", oldStats, newStats); } } }

方案二：停机迁移（严格一致性）

java

// 停机迁移方案 @Service public class ShutdownMigrationService { /** * 执行停机迁移 */ public void executeMigration() { // 1. 发布停机公告 sendShutdownNotification("系统将于2024-01-01 00:00进行数据库迁移，预计停机4小时"); // 2. 停止应用服务 stopApplicationServices(); // 3. 执行全量数据迁移 migrateFullData(); // 4. 执行增量数据迁移（停机期间产生的数据） migrateIncrementalData(); // 5. 数据一致性校验 verifyDataConsistency(); // 6. 切换配置 updateShardingConfiguration(); // 7. 重启应用服务 startApplicationServices(); // 8. 功能验证 validateBusinessFunctions(); // 9. 清理旧数据 cleanupOldData(); } private void migrateFullData() { // 使用数据迁移工具（如mysqldump + 自定义脚本） String command = String.format( "mysqldump -h%s -u%s -p%s %s | " + "mysql -h%s -u%s -p%s %s", oldHost, oldUser, oldPassword, oldDatabase, newHost, newUser, newPassword, newDatabase ); try { Process process = Runtime.getRuntime().exec(command); int exitCode = process.waitFor(); if (exitCode != 0) { throw new MigrationException("全量数据迁移失败"); } } catch (Exception e) { throw new MigrationException("迁移执行异常", e); } } }

3.2 动态扩容策略

一致性哈希算法实现

java

// 一致性哈希分片算法 @Component public class ConsistentHashShardingAlgorithm implements PreciseShardingAlgorithm<String> { private final TreeMap<Long, String> virtualNodes = new TreeMap<>(); private final int virtualNodeCount = 160; // 每个物理节点对应160个虚拟节点 public ConsistentHashShardingAlgorithm(List<String> physicalNodes) { // 初始化一致性哈希环 for (String node : physicalNodes) { for (int i = 0; i < virtualNodeCount; i++) { String virtualNodeName = node + "#" + i; long hash = hash(virtualNodeName); virtualNodes.put(hash, node); } } } @Override public String doSharding(Collection<String> availableTargetNames, PreciseShardingValue<String> shardingValue) { String key = shardingValue.getValue(); long hash = hash(key); // 找到第一个大于等于该hash值的节点 SortedMap<Long, String> tailMap = virtualNodes.tailMap(hash); if (tailMap.isEmpty()) { // 环回，取第一个节点 return virtualNodes.firstEntry().getValue(); } return tailMap.get(tailMap.firstKey()); } /** * 添加新节点 */ public void addNode(String newNode) { for (int i = 0; i < virtualNodeCount; i++) { String virtualNodeName = newNode + "#" + i; long hash = hash(virtualNodeName); virtualNodes.put(hash, newNode); } } /** * 移除节点 */ public void removeNode(String oldNode) { Iterator<Map.Entry<Long, String>> iterator = virtualNodes.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<Long, String> entry = iterator.next(); if (entry.getValue().equals(oldNode)) { iterator.remove(); } } } private long hash(String key) { // MurmurHash算法，分布更均匀 return MurmurHash.hash64(key.getBytes()); } }

第四章：监控与运维

4.1 分库分表监控体系

java

// 分库分表监控组件 @Component @Slf4j public class ShardingMonitor { @Autowired private ShardingSphereDataSource dataSource; @Autowired private MeterRegistry meterRegistry; @Scheduled(fixedRate = 60000) // 每分钟监控一次 public void monitorShardingStatus() { try { // 1. 监控分片数据分布 monitorDataDistribution(); // 2. 监控查询性能 monitorQueryPerformance(); // 3. 监控连接池状态 monitorConnectionPool(); // 4. 监控慢查询 monitorSlowQueries(); // 5. 监控数据倾斜 monitorDataSkew(); } catch (Exception e) { log.error("分片监控异常", e); } } private void monitorDataDistribution() { // 查询每个分片的数据量 Map<String, Long> shardDataCounts = new HashMap<>(); for (int i = 0; i < 4; i++) { for (int j = 0; j < 16; j++) { String tableName = String.format("orders_%d", j); String sql = String.format("SELECT COUNT(*) FROM %s", tableName); Long count = jdbcTemplate.queryForObject( String.format("ds%d", i), sql, Long.class); String shardKey = String.format("ds%d.%s", i, tableName); shardDataCounts.put(shardKey, count); // 记录到监控指标 meterRegistry.gauge("sharding.data.count", Tags.of("shard", shardKey), count); } } // 计算数据分布均匀度 double avg = shardDataCounts.values().stream() .mapToLong(Long::longValue).average().orElse(0); double variance = shardDataCounts.values().stream() .mapToDouble(count -> Math.pow(count - avg, 2)) .average().orElse(0); if (variance / avg > 0.3) { // 阈值30% sendAlert("数据分布不均匀警告", shardDataCounts); } } private void monitorSlowQueries() { // 监控慢查询日志 String slowQueryLog = "/var/log/mysql/slow.log"; try (BufferedReader reader = new BufferedReader(new FileReader(slowQueryLog))) { String line; List<String> slowQueries = new ArrayList<>(); while ((line = reader.readLine()) != null) { if (line.contains("Query_time")) { slowQueries.add(line); } } if (!slowQueries.isEmpty()) { log.warn("发现慢查询: {}", slowQueries.size()); meterRegistry.counter("sharding.slow.queries") .increment(slowQueries.size()); } } catch (IOException e) { log.error("读取慢查询日志失败", e); } } }

4.2 运维工具集

yaml

# 分库分表运维工具配置 sharding: tools: # 数据校验工具 style="margin-top:12px">