数据库中间件从MyCAT到ShardingSphere的迁移复盘：分库分表策略的重构与性能压测对比

数据库中间件从MyCAT到ShardingSphere的迁移复盘：分库分表策略的重构与性能压测对比

一、项目背景与业务挑战

2024年底，我们运维团队接到一个令人头疼的迁移任务：将核心业务数据库的中间件从MyCAT迁移到ShardingSphere。这不是一次简单的"换个组件"操作，而是牵涉到分库分表策略重构、SQL兼容性适配、连接池管理重构、以及性能基准对比验证的系统工程。

MyCAT的维护困境。MyCAT作为我们2018年选型的分库分表中间件，已经运行了6年。但随着业务规模增长，MyCAT的三个致命问题逐渐暴露：一是社区停止维护——最后一次正式Release是2019年，大量Bug和安全漏洞无人修复，我们自行维护了47个补丁但精力有限；二是SQL兼容性差——MyCAT只支持有限的SQL语法，复杂查询（子查询、跨分片JOIN、聚合函数）经常报错或返回错误结果，研发团队不得不将大量业务逻辑降级到应用层处理；三是性能瓶颈——MyCAT的单线程模型导致高并发下吞吐量受限，QPS超过8000时延迟急剧上升，P99延迟超过2秒。

ShardingSphere的优势与迁移的必要性。Apache ShardingSphere作为活跃社区维护的分库分表中间件，在SQL兼容性、并发性能、生态集成三个方面远超MyCAT：支持99%的MySQL语法（包括子查询、跨分片JOIN、复杂聚合）、基于Netty的多线程NIO架构提供高并发吞吐、与Spring Boot/K8s/云原生生态深度集成。迁移到ShardingSphere是业务发展的刚性需求。

迁移的核心挑战。这次迁移不是"停机替换"——业务不可中断。我们需要解决四个关键问题：一是分库分表策略差异——MyCAT使用rule.xml配置分片规则，ShardingSphere使用YAML配置，且分片算法的表达能力不同；二是SQL兼容性差异——大量在MyCAT中需要应用层 workaround的SQL，在ShardingSphere中可以直接执行，但也有一些MyCAT特有的SQL扩展需要适配；三是连接池管理差异——MyCAT自建连接池，ShardingSphere依赖HikariCP，连接数和超时配置需要重新调优；四是性能基准验证——必须通过压测证明ShardingSphere在吞吐量、延迟、资源消耗上不低于MyCAT，否则迁移不达标。

整个迁移周期从立项到全量切换历时4个月，涉及3个业务集群、12个逻辑库、48个分片、超过200TB数据。本文将完整复盘这次迁移的方案设计、实践落地、关键挑战和最终结果。

二、核心方案

2.1 四阶段渐进式迁移策略

我们设计了"评估-并行-验证-切换"四阶段迁移策略，每个阶段有明确的准入和退出条件：

2.2 分片策略映射与重构

MyCAT和ShardingSphere的分片配置存在结构性差异。MyCAT使用XML定义分片规则（tableRule+function），ShardingSphere使用YAML定义（shardingRule+shardingAlgorithms）。我们开发了自动化映射工具，将MyCAT的分片规则解析并转化为ShardingSphere的YAML配置。

关键映射逻辑：

• MyCAT的PartitionByMod算法 → ShardingSphere的MOD分片算法
• MyCAT的PartitionByLong（范围分片） → ShardingSphere的RANGE_INTERVAL分片算法
• MyCAT的PartitionByHashMod→ ShardingSphere的HASH_MOD分片算法
• MyCAT的PartitionByDate（按日期分片） → ShardingSphere的INTERVAL分片算法

但不是所有映射都是一一对应的——MyCAT支持自定义Java分片算法，ShardingSphere也支持但接口不同。对于我们自行开发的3个自定义分片算法（按业务ID前缀分片、按地域分片、按用户等级分片），需要按照ShardingSphere的ShardingAlgorithm接口重新实现。

2.3 性能压测对比设计

性能压测是迁移决策的关键依据。我们搭建了两组对照集群：MyCAT集群（2 Proxy实例）和ShardingSphere集群（2 Proxy实例），后端连接相同的物理MySQL分片，确保压测对比的公平性。压测维度包括：吞吐量（QPS）、延迟（P50/P95/P99）、连接池效率、SQL兼容性通过率、资源消耗（CPU/Memory/Network）、跨分片查询性能。

三、实践落地

3.1 分片策略映射工具

from dataclasses import dataclass, field from typing import List, Dict, Optional, Tuple from enum import Enum import xml.etree.ElementTree as ET import yaml import re import logging logger = logging.getLogger(__name__) class MyCATAlgorithmType(Enum): """MyCAT分片算法类型""" MOD = "PartitionByMod" # 取模分片 LONG_RANGE = "PartitionByLong" # 范围分片 HASH_MOD = "PartitionByHashMod" # 哈希取模分片 DATE = "PartitionByDate" # 按日期分片 CUSTOM = "CustomAlgorithm" # 自定义算法 class ShardingSphereAlgorithmType(Enum): """ShardingSphere分片算法类型""" MOD = "MOD" # 取模分片 RANGE_INTERVAL = "RANGE_INTERVAL" # 范围分片 HASH_MOD = "HASH_MOD" # 哈希取模分片 INTERVAL = "INTERVAL" # 按日期分片 CLASS_BASED = "CLASS_BASED" # 基于类的自定义算法 # MyCAT → ShardingSphere 算法类型映射 ALGORITHM_TYPE_MAPPING = { MyCATAlgorithmType.MOD: ShardingSphereAlgorithmType.MOD, MyCATAlgorithmType.LONG_RANGE: ShardingSphereAlgorithmType.RANGE_INTERVAL, MyCATAlgorithmType.HASH_MOD: ShardingSphereAlgorithmType.HASH_MOD, MyCATAlgorithmType.DATE: ShardingSphereAlgorithmType.INTERVAL, MyCATAlgorithmType.CUSTOM: ShardingSphereAlgorithmType.CLASS_BASED, } @dataclass class MyCATTableRule: """MyCAT分片规则""" table_name: str # 逻辑表名 data_node: str # 数据节点分布 algorithm_type: MyCATAlgorithmType # 分片算法类型 algorithm_params: Dict[str, str] # 算法参数 partition_count: int # 分片数量 primary_key: str # 分片键 @dataclass class ShardingSphereRule: """ShardingSphere分片规则""" logic_table: str # 逻辑表名 actual_data_nodes: str # 实际数据节点分布表达式 sharding_column: str # 分片列 sharding_algorithm_name: str # 分片算法名称 sharding_algorithm_type: ShardingSphereAlgorithmType # 分片算法类型 sharding_algorithm_props: Dict[str, str] # 算法属性 @dataclass class MigrationMappingResult: """迁移映射结果""" logic_schema: str # 逻辑库名 mycat_rules: List[MyCATTableRule] # MyCAT原始规则 sharding_sphere_rules: List[ShardingSphereRule] # 映射后的ShardingSphere规则 unmapped_rules: List[MyCATTableRule] # 无法自动映射的规则 compatibility_issues: List[Dict] # 兼容性问题列表 class MyCATToShardingSphereMapper: """MyCAT分片规则 → ShardingSphere YAML配置的自动映射工具""" def __init__(self, mycat_config_path: str, cluster_config: Dict): self.mycat_config_path = mycat_config_path self.cluster_config = cluster_config # 集群配置（分片命名规则等） def map_rules(self) -> MigrationMappingResult: """执行分片规则映射""" try: # Step 1: 解析MyCAT rule.xml mycat_rules = self._parse_mycat_rule_xml() # Step 2: 逐规则映射到ShardingSphere格式 ss_rules = [] unmapped = [] issues = [] for rule in mycat_rules: try: ss_rule = self._map_single_rule(rule) ss_rules.append(ss_rule) except Exception as e: logger.warning(f"规则映射失败: {rule.table_name}, 原因: {e}") unmapped.append(rule) issues.append({ "table": rule.table_name, "reason": str(e), "type": "algorithm_mapping_failure", }) # Step 3: 生成ShardingSphere YAML配置 result = MigrationMappingResult( logic_schema=self.cluster_config.get("schema_name", "ds"), mycat_rules=mycat_rules, sharding_sphere_rules=ss_rules, unmapped_rules=unmapped, compatibility_issues=issues, ) logger.info( f"规则映射完成: 总规则={len(mycat_rules)}, " f"成功映射={len(ss_rules)}, 未映射={len(unmapped)}, " f"兼容性问题={len(issues)}" ) return result except Exception as e: logger.error(f"规则映射异常: {e}", exc_info=True) raise def _parse_mycat_rule_xml(self) -> List[MyCATTableRule]: """解析MyCAT rule.xml文件""" rules = [] try: tree = ET.parse(self.mycat_config_path) root = tree.getroot() # 解析tableRule节点 for table_rule in root.findall(".//tableRule"): rule_name = table_rule.get("name", "") rule_node = table_rule.find("rule") if rule_node is None: continue columns = rule_node.find("columns").text.strip() if rule_node.find("columns") is not None else "" algorithm_name = rule_node.find("algorithm").text.strip() if rule_node.find("algorithm") is not None else "" # 解析对应的function节点获取算法参数 func_node = root.find(f".//function[@name='{algorithm_name}']") if func_node is None: logger.warning(f"未找到算法定义: {algorithm_name}") continue algorithm_class = func_node.get("class", "") algorithm_type = self._identify_algorithm_type(algorithm_class) algorithm_params = self._extract_algorithm_params(func_node) partition_count = int(func_node.get("partitionCount", "0")) rules.append(MyCATTableRule( table_name=rule_name, data_node=self._resolve_data_nodes(rule_name), algorithm_type=algorithm_type, algorithm_params=algorithm_params, partition_count=partition_count, primary_key=columns, )) except ET.ParseError as e: logger.error(f"MyCAT rule.xml解析失败: {e}") raise return rules def _identify_algorithm_type(self, class_name: str) -> MyCATAlgorithmType: """识别MyCAT算法类型""" # 常见MyCAT分片算法的类名映射 CLASS_TO_TYPE = { "io.mycat.route.function.PartitionByMod": MyCATAlgorithmType.MOD, "io.mycat.route.function.PartitionByLong": MyCATAlgorithmType.LONG_RANGE, "io.mycat.route.function.PartitionByHashMod": MyCATAlgorithmType.HASH_MOD, "io.mycat.route.function.PartitionByDate": MyCATAlgorithmType.DATE, } for cls, alg_type in CLASS_TO_TYPE.items(): if cls in class_name: return alg_type # 自定义算法 return MyCATAlgorithmType.CUSTOM def _extract_algorithm_params(self, func_node) -> Dict[str, str]: """提取MyCAT算法参数""" params = {} for attr_name in ["partitionCount", "partitionLength", "mapFile", "beginDate", "endDate", "partitions", "defaultNode"]: attr_value = func_node.get(attr_name) if attr_value: params[attr_name] = attr_value # 提取property子节点参数 for prop in func_node.findall("property"): prop_name = prop.get("name", "") prop_value = prop.text.strip() if prop.text else "" if prop_name and prop_value: params[prop_name] = prop_value return params def _map_single_rule(self, mycat_rule: MyCATTableRule) -> ShardingSphereRule: """映射单条MyCAT规则到ShardingSphere格式""" # 确定目标算法类型 ss_algorithm_type = ALGORITHM_TYPE_MAPPING.get(mycat_rule.algorithm_type) if ss_algorithm_type is None: raise ValueError(f"无法映射算法类型: {mycat_rule.algorithm_type}") # 确定算法名称 algorithm_name = f"{mycat_rule.table_name}_sharding_algorithm" # 映射算法参数 ss_props = self._map_algorithm_props( mycat_rule.algorithm_type, ss_algorithm_type, mycat_rule.algorithm_params, mycat_rule.partition_count ) # 映射数据节点分布表达式 actual_data_nodes = self._map_data_nodes( mycat_rule.table_name, mycat_rule.data_node, mycat_rule.partition_count ) return ShardingSphereRule( logic_table=mycat_rule.table_name, actual_data_nodes=actual_data_nodes, sharding_column=mycat_rule.primary_key, sharding_algorithm_name=algorithm_name, sharding_algorithm_type=ss_algorithm_type, sharding_algorithm_props=ss_props, ) def _map_algorithm_props(self, src_type: MyCATAlgorithmType, target_type: ShardingSphereAlgorithmType, params: Dict[str, str], partition_count: int) -> Dict[str, str]: """映射算法参数""" if target_type == ShardingSphereAlgorithmType.MOD: return { "sharding-count": str(partition_count), } elif target_type == ShardingSphereAlgorithmType.RANGE_INTERVAL: # MyCAT范围分片的mapFile → ShardingSphere的range-lower/cupper map_file_content = params.get("mapFile_content", "") return self._parse_range_map_file(map_file_content, partition_count) elif target_type == ShardingSphereAlgorithmType.HASH_MOD: return { "sharding-count": str(partition_count), } elif target_type == ShardingSphereAlgorithmType.INTERVAL: return { "datetime-pattern": params.get("dateFormat", "yyyy-MM-dd"), "datetime-interval-amount": params.get("step", "1"), "datetime-interval-unit": params.get("stepUnit", "MONTHS"), "sharding-scope-pattern-offset": "0", } elif target_type == ShardingSphereAlgorithmType.CLASS_BASED: # 自定义算法需要指定实现类 custom_class = params.get("customClass", "") return { "strategy": "STANDARD", "algorithmClassName": self._adapt_custom_class(custom_class), } else: raise ValueError(f"未知目标算法类型: {target_type}") def _map_data_nodes(self, table_name: str, data_node: str, partition_count: int) -> str: """映射数据节点分布表达式""" # MyCAT: dn0,dn1,dn2,... → ShardingSphere: ds_${0..partition_count-1}.table_name naming_rule = self.cluster_config.get("node_naming_rule", "ds_${0..%d}") ds_pattern = naming_rule % (partition_count - 1) return f"{ds_pattern}.{table_name}" def generate_yaml_config(self, mapping_result: MigrationMappingResult) -> str: """生成ShardingSphere YAML配置文件""" config = { "dataSources": self._generate_datasource_config(), "rules": [{ "shardingRule": { "tables": {}, "shardingAlgorithms": {}, "bindingTables": [], } }] } # 填充分片表配置 for rule in mapping_result.sharding_sphere_rules: config["rules"][0]["shardingRule"]["tables"][rule.logic_table] = { "actualDataNodes": rule.actual_data_nodes, "shardingColumn": rule.sharding_column, "shardingAlgorithmName": rule.sharding_algorithm_name, } # 填充分片算法配置 config["rules"][0]["shardingRule"]["shardingAlgorithms"][rule.sharding_algorithm_name] = { "type": rule.sharding_algorithm_type.value, "props": rule.sharding_algorithm_props, } return yaml.dump(config, allow_unicode=True, default_flow_style=False, sort_keys=False) def _generate_datasource_config(self) -> Dict: """生成数据源配置""" datasources = {} for i in range(self.cluster_config.get("shard_count", 0)): ds_name = f"ds_{i}" datasources[ds_name] = { "url": f"jdbc:mysql://{self.cluster_config.get('mysql_host')}:3306/shard_{i}", "username": self.cluster_config.get("mysql_user", "root"), "password": self.cluster_config.get("mysql_password", ""), "connectionTimeoutMilliseconds": 30000, "idleTimeoutMilliseconds": 60000, "maxLifetimeMilliseconds": 1800000, "maxPoolSize": 50, "minPoolSize": 1, } return datasources def _adapt_custom_class(self, mycat_class: str) -> str: """适配自定义分片算法的类名（MyCAT接口→ShardingSphere接口）""" # 记录需要重写的自定义算法 logger.info(f"自定义算法需要重写: {mycat_class}") return f"com.sharding.custom.{mycat_class.split('.')[-1]}"

3.2 SQL兼容性扫描器

from dataclasses import dataclass from typing import List, Dict, Tuple import logging logger = logging.getLogger(__name__) @dataclass class SQLCompatibilityResult: """SQL兼容性测试结果""" sql_id: str original_sql: str mycat_result: str # MyCAT执行结果描述 ss_result: str # ShardingSphere执行结果描述 is_compatible: bool # 是否兼容 difference_type: str # 差异类型：syntax_error / result_diff / performance_diff / ok class SQLCompatibilityScanner: """SQL兼容性扫描器：业务SQL在ShardingSphere上的执行测试""" def __init__(self, mycat_executor, ss_executor, sql_repository): self.mycat_executor = mycat_executor # MyCAT执行器 self.ss_executor = ss_executor # ShardingSphere执行器 self.sql_repository = sql_repository # SQL仓库（业务历史SQL） def scan(self, schema_name: str) -> Dict: """扫描指定逻辑库的所有业务SQL兼容性""" try: # Step 1: 从SQL仓库获取该逻辑库的历史SQL sql_list = self.sql_repository.get_sqls_by_schema(schema_name) logger.info(f"获取到 {schema_name} 的历史SQL: {len(sql_list)}条") results = [] compatible_count = 0 incompatible_count = 0 diff_types = {} # Step 2: 逐条SQL在两个中间件上执行对比 for sql_info in sql_list: try: result = self._test_single_sql(sql_info) results.append(result) if result.is_compatible: compatible_count += 1 else: incompatible_count += 1 diff_types[result.difference_type] = diff_types.get(result.difference_type, 0) + 1 except Exception as e: logger.warning(f"SQL测试异常: {sql_info['id']}, {e}") incompatible_count += 1 diff_types["test_error"] = diff_types.get("test_error", 0) + 1 # Step 3: 计算兼容率并生成报告 compatibility_rate = compatible_count / len(sql_list) if sql_list else 0 report = { "schema": schema_name, "total_sqls": len(sql_list), "compatible_count": compatible_count, "incompatible_count": incompatible_count, "compatibility_rate": compatibility_rate, "diff_type_distribution": diff_types, "incompatible_sqls": [r for r in results if not r.is_compatible], } logger.info( f"SQL兼容性扫描完成: {schema_name}, " f"兼容率={compatibility_rate:.2%}, " f"不兼容={incompatible_count}条" ) return report except Exception as e: logger.error(f"SQL兼容性扫描异常: {e}", exc_info=True) return {"schema": schema_name, "error": str(e)} def _test_single_sql(self, sql_info: Dict) -> SQLCompatibilityResult: """测试单条SQL在两个中间件上的执行结果""" sql_id = sql_info["id"] sql_text = sql_info["sql"] # 在MyCAT上执行（对照组） try: mycat_result = self.mycat_executor.execute( sql=sql_text, schema=sql_info.get("schema") ) except Exception as e: mycat_result = f"MyCAT执行错误: {e}" # 在ShardingSphere上执行（实验组） try: ss_result = self.ss_executor.execute( sql=sql_text, schema=sql_info.get("schema") ) except Exception as e: ss_result = f"ShardingSphere执行错误: {e}" # 对比结果 is_compatible, diff_type = self._compare_results(mycat_result, ss_result, sql_text) return SQLCompatibilityResult( sql_id=sql_id, original_sql=sql_text, mycat_result=str(mycat_result)[:200], ss_result=str(ss_result)[:200], is_compatible=is_compatible, difference_type=diff_type, ) def _compare_results(self, mycat_result, ss_result, sql_text: str) -> Tuple[bool, str]: """对比两个中间件的执行结果""" # 两者都执行错误 if isinstance(mycat_result, str) and "错误" in mycat_result: if isinstance(ss_result, str) and "错误" in ss_result: # 同样不支持 → 兼容（行为一致） return True, "ok" # ShardingSphere执行成功但MyCAT不支持 → 兼容性提升 if isinstance(mycat_result, str) and "错误" in mycat_result: if not isinstance(ss_result, str) or "错误" not in ss_result: return True, "ok" # ShardingSphere支持了MyCAT不支持的SQL # ShardingSphere执行错误但MyCAT支持 → 不兼容 if isinstance(ss_result, str) and "错误" in ss_result: if not isinstance(mycat_result, str) or "错误" not in mycat_result: return False, "syntax_error" # 两者都成功，对比结果集 if isinstance(mycat_result, list) and isinstance(ss_result, list): if len(mycat_result) == len(ss_result): # 行数一致，检查内容差异 if mycat_result == ss_result: return True, "ok" else: return False, "result_diff" else: return False, "result_diff" # 执行时间差异（性能维度） # 这里简化处理，实际需要更细粒度的性能对比 return True, "ok"

3.3 性能压测对比框架

from dataclasses import dataclass, field from typing import List, Dict import time import statistics import logging logger = logging.getLogger(__name__) @dataclass class BenchmarkConfig: """压测配置""" scenario_name: str qps_target: int # 目标QPS duration_seconds: int # 持续时间 concurrency: int # 并发数 sql_mix_ratio: Dict[str, float] # SQL类型占比：point_select/range_scan/join/update/insert @dataclass class BenchmarkMetric: """压测指标""" scenario: str middleware: str # mycat / sharding_sphere actual_qps: float # 实际QPS latency_p50_ms: float # P50延迟 latency_p95_ms: float # P95延迟 latency_p99_ms: float # P99延迟 error_rate: float # 错误率 cpu_usage_percent: float # CPU使用率 memory_usage_mb: float # 内存使用 connection_pool_active: int # 连接池活跃连接数 connection_pool_idle: int # 连接池空闲连接数 cross_shard_query_ratio: float # 跨分片查询占比 @dataclass class BenchmarkComparison: """压测对比结果""" scenario: str mycat_metrics: BenchmarkMetric ss_metrics: BenchmarkMetric qps_improvement: float # QPS提升百分比 latency_p99_improvement: float # P99延迟改善百分比 error_rate_improvement: float # 错误率改善百分比 is_ss_better: bool # ShardingSphere是否整体优于MyCAT # 压测场景配置 BENCHMARK_SCENARIOS = { "point_select": BenchmarkConfig( scenario_name="单分片点查", qps_target=10000, duration_seconds=300, concurrency=100, sql_mix_ratio={"point_select": 1.0}, ), "mixed workload": BenchmarkConfig( scenario_name="混合读写负载", qps_target=8000, duration_seconds=300, concurrency=80, sql_mix_ratio={ "point_select": 0.45, "range_scan": 0.15, "join": 0.10, "update": 0.20, "insert": 0.10, }, ), "cross_shard_join": BenchmarkConfig( scenario_name="跨分片JOIN查询", qps_target=2000, duration_seconds=300, concurrency=20, sql_mix_ratio={"join": 1.0}, ), "high_concurrency_write": BenchmarkConfig( scenario_name="高并发写入", qps_target=5000, duration_seconds=300, concurrency=50, sql_mix_ratio={"insert": 0.6, "update": 0.4}, ), } class BenchmarkRunner: """性能压测对比框架""" def __init__(self, mycat_proxy, ss_proxy, metrics_collector): self.mycat_proxy = mycat_proxy # MyCAT Proxy实例 self.ss_proxy = ss_proxy # ShardingSphere Proxy实例 self.metrics_collector = metrics_collector # 指标采集器 def run_comparison(self) -> List[BenchmarkComparison]: """运行全部压测场景对比""" comparisons = [] for scenario_name, config in BENCHMARK_SCENARIOS.items(): try: logger.info(f"开始压测场景: {scenario_name}") # 在MyCAT上执行压测 mycat_metrics = self._run_benchmark("mycat", config) # 在ShardingSphere上执行压测 ss_metrics = self._run_benchmark("sharding_sphere", config) # 对比结果 comparison = self._compare_metrics(scenario_name, mycat_metrics, ss_metrics) comparisons.append(comparison) logger.info( f"压测对比完成: {scenario_name}, " f"ShardingSphere整体优于MyCAT={comparison.is_ss_better}, " f"QPS提升={comparison.qps_improvement:.1f}%, " f"P99延迟改善={comparison.latency_p99_improvement:.1f}%" ) except Exception as e: logger.error(f"压测场景 {scenario_name} 异常: {e}", exc_info=True) return comparisons def _run_benchmark(self, middleware: str, config: BenchmarkConfig) -> BenchmarkMetric: """在指定中间件上运行压测""" proxy = self.mycat_proxy if middleware == "mycat" else self.ss_proxy # 采集压测前的资源基线 baseline = self.metrics_collector.collect_resource_baseline(middleware) # 执行压测 latencies = [] error_count = 0 total_count = 0 start_time = time.time() while time.time() - start_time < config.duration_seconds: # 根据SQL类型占比选择本次执行的SQL sql_type = self._pick_sql_type(config.sql_mix_ratio) sql = self._generate_test_sql(sql_type, config) try: exec_start = time.time() result = proxy.execute(sql, timeout=10) exec_latency = (time.time() - exec_start) * 1000 # 转为毫秒 latencies.append(exec_latency) if result.get("error"): error_count += 1 except Exception as e: error_count += 1 logger.debug(f"压测SQL执行异常: {e}") total_count += 1 # 计算压测指标 actual_qps = total_count / config.duration_seconds error_rate = error_count / total_count if total_count > 0 else 1.0 sorted_latencies = sorted(latencies) p50 = sorted_latencies[int(len(sorted_latencies) * 0.50)] if sorted_latencies else 0 p95 = sorted_latencies[int(len(sorted_latencies) * 0.95)] if sorted_latencies else 0 p99 = sorted_latencies[int(len(sorted_latencies) * 0.99)] if sorted_latencies else 0 # 采集压测后的资源指标 post_resources = self.metrics_collector.collect_resource_snapshot(middleware) return BenchmarkMetric( scenario=config.scenario_name, middleware=middleware, actual_qps=actual_qps, latency_p50_ms=p50, latency_p95_ms=p95, latency_p99_ms=p99, error_rate=error_rate, cpu_usage_percent=post_resources.get("cpu_usage", 0), memory_usage_mb=post_resources.get("memory_usage_mb", 0), connection_pool_active=post_resources.get("active_connections", 0), connection_pool_idle=post_resources.get("idle_connections", 0), cross_shard_query_ratio=post_resources.get("cross_shard_ratio", 0), ) def _pick_sql_type(self, mix_ratio: Dict[str, float]) -> str: """根据SQL类型占比随机选择本次执行的SQL类型""" import random types = list(mix_ratio.keys()) weights = list(mix_ratio.values()) return random.choices(types, weights=weights, k=1)[0] def _generate_test_sql(self, sql_type: str, config: BenchmarkConfig) -> str: """生成测试SQL""" SQL_TEMPLATES = { "point_select": "SELECT * FROM t_order WHERE order_id = {id}", "range_scan": "SELECT * FROM t_order WHERE order_id BETWEEN {id} AND {id}+100", "join": "SELECT o.*, u.name FROM t_order o JOIN t_user u ON o.user_id = u.id WHERE o.order_id = {id}", "update": "UPDATE t_order SET status = 1 WHERE order_id = {id}", "insert": "INSERT INTO t_order (order_id, user_id, status) VALUES ({id}, {id}%1000, 0)", } template = SQL_TEMPLATES.get(sql_type, SQL_TEMPLATES["point_select"]) return template.replace("{id}", str(int(time.time() * 1000) % 100000)) def _compare_metrics(self, scenario: str, mycat: BenchmarkMetric, ss: BenchmarkMetric) -> BenchmarkComparison: """对比两个中间件的压测指标""" # QPS提升百分比 qps_improvement = ( (ss.actual_qps - mycat.actual_qps) / mycat.actual_qps * 100 if mycat.actual_qps > 0 else 0 ) # P99延迟改善百分比（减少为改善） latency_p99_improvement = ( (mycat.latency_p99_ms - ss.latency_p99_ms) / mycat.latency_p99_ms * 100 if mycat.latency_p99_ms > 0 else 0 ) # 错误率改善百分比 error_rate_improvement = ( (mycat.error_rate - ss.error_rate) / mycat.error_rate * 100 if mycat.error_rate > 0 else 0 ) # 综合判定：QPS提升+P99改善+错误率改善 三项均为正 → ShardingSphere优于MyCAT is_better = ( qps_improvement > 0 and latency_p99_improvement > 0 and error_rate_improvement >= 0 ) return BenchmarkComparison( scenario=scenario, mycat_metrics=mycat, ss_metrics=ss, qps_improvement=qps_improvement, latency_p99_improvement=latency_p99_improvement, error_rate_improvement=error_rate_improvement, is_ss_better=is_better, )

四、关键挑战与应对策略

4.1 自定义分片算法的重写与验证

MyCAT有3个自定义分片算法（按业务ID前缀分片、按地域分片、按用户等级分片），ShardingSphere的ShardingAlgorithm接口与MyCAT的分片接口差异显著——MyCAT返回Integer分片序号，ShardingSphere返回Collection<String>数据节点名。重写过程需要：一是将分片结果从整数索引映射到数据节点名格式（ds_0而非0）；二是ShardingSphere的分片范围查询需要同时返回路由到的多个数据节点，而MyCAT只返回一个。我们在重写后对3个算法分别执行了10万次路由验证（覆盖全值域），确保路由结果与原算法100%一致。

4.2 跨分片JOIN的性能差异

MyCAT对跨分片JOIN采用"笛卡尔积+内存合并"策略，性能极差（2表JOIN超过3秒）。ShardingSphere采用"流式归并+内存限制"策略，同样场景下JOIN耗时约1.2秒，但仍远慢于单分片查询。对于核心业务的4个高频跨分片JOIN查询，我们设计了"宽表预聚合"方案——将JOIN结果预先聚合到宽表中，查询时直接读取宽表而非实时JOIN，4个查询的P99延迟从1.2秒降到50毫秒。

4.3 连接池参数的重新调优

MyCAT自建连接池默认配置（maxActive=100, idleCheck=60s），ShardingSphere使用HikariCP（默认maxPoolSize=50, idleTimeout=600s）。初期我们直接使用HikariCP默认配置，结果在高并发场景下出现连接等待超时（等待超过30秒）。通过4轮参数调优（基于压测反馈逐步调整），最终将maxPoolSize从50调整到80、connectionTimeout从30s调整到5s、idleTimeout从600s调整到300s、maxLifetime从1800s调整到1200s。调优后连接等待超时消失，连接池利用率从45%提升到72%。

4.4 双写期间的数据一致性保障

阶段二的"双中间件并行运行"需要确保写入数据在两个中间件中一致。我们采用"主写MyCAT+同步复制到ShardingSphere"的策略——应用层写入流量仍走MyCAT（保持业务稳定），同时由同步服务将写入事件复制到ShardingSphere。数据一致性校验采用"逐分片CRC校验"——每10分钟对每个分片的表数据计算CRC32摘要并对比，不一致时分片标记为"需修复"并自动触发修复流程。双写期间发现3次数据不一致（均因并发写入时的时序差异），均在10分钟内自动修复。

4.5 MyCAT特有SQL的适配问题

MyCAT有一些特有的SQL扩展语法（如/*!mycat:sql=select * from ...*/注释路由提示），这些语法在ShardingSphere中不支持。我们排查出业务代码中有23处使用了MyCAT特有语法，需要逐一手动适配。适配策略：一是将路由提示替换为ShardingSphere的HintManager API（Java代码中设置分片路由Hint）；二是将MyCAT注释语法中的SQL提取出来直接执行（去掉注释包装）。23处适配在2周内完成，每处适配后都经过回归测试验证。

五、总结

MyCAT到ShardingSphere的迁移是一次深度基础设施替换项目，4个月的渐进式迁移最终取得了以下成果：

性能对比数据（4个压测场景汇总）：

• SQL兼容性：MyCAT通过率72%（28%的SQL不支持），ShardingSphere通过率96%（4%的SQL需适配），兼容性提升24个百分点
• 迁移成功率：48个分片全部成功迁移，0数据丢失，0服务中断
• 回滚预案：双写阶段保留MyCAT作为回退通道，全量切换后保留2周降级备用，最终平稳退役

核心经验总结：

1. 渐进式迁移是大型中间件替换的唯一安全路径。四阶段策略（评估→并行→验证→切换）将风险逐级降低：评估阶段发现兼容性问题，并行阶段验证数据一致性，压测阶段证明性能达标，切换阶段确保平稳过渡。任何试图一步到位的方案都会将所有风险集中在单次切换上，失败代价极高。

2. 分片策略映射工具化是效率关键。48个分片规则如果纯手工迁移，需要至少3周且容易出错。自动化映射工具将规则解析→类型映射→参数转换→YAML生成全流程自动化，耗时从3周缩短到2天。但自定义算法必须人工重写和验证——这部分占总工时的40%，无法自动化。

3. 双写+一致性校验是并行阶段的基石。并行阶段的核心目标不是"两个中间件都可用"，而是"数据绝对一致"。CRC校验每10分钟一次的全量对比，发现不一致时立即修复，这保证了并行阶段可以随时回退到MyCAT而无需担心数据缺失。

4. 连接池调优不能凭经验，必须基于压测数据。HikariCP的默认参数对ShardingSphere Proxy场景不适用。4轮压测驱动的参数调优，最终找到适合我们业务负载的连接池配置。经验值可以作为起点，但终点必须由数据决定。

5. 宽表预聚合是跨分片JOIN的性能杀手。分库分表中间件的跨分片JOIN天然有性能瓶颈，ShardingSphere比MyCAT快2倍多但仍不满足核心业务的延迟要求。宽表预聚合通过空间换时间，将实时JOIN转为预计算宽表查询，从根本上解决了跨分片JOIN的性能问题。

展望下一步，我们计划将ShardingSphere从Proxy模式切换到Sidecar模式（通过K8s DaemonSet部署），进一步降低中间件的网络开销和单点风险，同时为多集群联邦查询做好准备。