架构之水平扩展

架构之水平扩展

引言

在垂直扩展达到单节点物理极限后，水平扩展（Scale Out）成为了支撑业务持续增长的唯一选择。水平扩展架构的核心思想是：通过增加服务器数量，线性扩充系统性能，将原本集中在单节点的负载分散到多个节点上，从而实现系统处理能力的无限扩展。

然而，水平扩展并非简单的"加机器"，它对系统架构设计有着严格的要求。如何在架构的各个层次进行可水平扩展的设计，如何处理好数据一致性、服务发现、负载均衡等分布式系统的核心问题，这些都是水平扩展架构必须面对的挑战。

水平扩展架构的核心理念

水平扩展 vs 垂直扩展

水平扩展和垂直扩展各有其适用场景：

• 垂直扩展：适合业务初期，简单快速，但受限于单节点物理极限
• 水平扩展：适合大规模系统，可扩展性强，但技术复杂度高

水平扩展的价值定位

水平扩展在架构演进中扮演着关键角色：

1. 突破单节点限制：通过分布式架构突破单节点的物理极限
2. 线性扩展能力：理论上可以实现无限的线性扩展
3. 高可用保障：多节点部署提供天然的高可用能力
4. 成本可控：通过标准化硬件实现成本的可预测增长

数据库层的水平扩展

数据分片策略

实践案例：用户数据水平分片

// 分片策略管理器@ComponentpublicclassShardingStrategyManager{privatestaticfinalLoggerlog=LoggerFactory.getLogger(ShardingStrategyManager.class);// 分片算法配置privatefinalMap<String,ShardingAlgorithm>shardingAlgorithms=newConcurrentHashMap<>();@PostConstructpublicvoidinit(){// 初始化分片算法shardingAlgorithms.put("user_id_mod",newModuloShardingAlgorithm(4));// 4个分片shardingAlgorithms.put("order_date_range",newRangeShardingAlgorithm());shardingAlgorithms.put("hash_consistent",newConsistentHashShardingAlgorithm());}/** * 根据分片键计算目标数据库 */publicStringdetermineTargetDatabase(StringshardingKey,StringalgorithmType){ShardingAlgorithmalgorithm=shardingAlgorithms.get(algorithmType);if(algorithm==null){thrownewIllegalArgumentException("未知的分片算法: "+algorithmType);}StringtargetDatabase=algorithm.doSharding(shardingKey);log.debug("分片键: {}, 算法: {}, 目标数据库: {}",shardingKey,algorithmType,targetDatabase);returntargetDatabase;}/** * 用户ID取模分片算法 */publicstaticclassModuloShardingAlgorithmimplementsShardingAlgorithm{privatefinalintshardCount;publicModuloShardingAlgorithm(intshardCount){this.shardCount=shardCount;}@OverridepublicStringdoSharding(StringshardingKey){try{longuserId=Long.parseLong(shardingKey);intshardIndex=(int)(userId%shardCount);return"db_user_"+shardIndex;}catch(NumberFormatExceptione){thrownewIllegalArgumentException("无效的用户ID: "+shardingKey);}}}/** * 范围分片算法（按时间） */publicstaticclassRangeShardingAlgorithmimplementsShardingAlgorithm{privatefinalMap<String,String>rangeMap=newLinkedHashMap<>();publicRangeShardingAlgorithm(){// 按年份分片rangeMap.put("2023","db_order_2023");rangeMap.put("2024","db_order_2024");rangeMap.put("2025","db_order_2025");}@OverridepublicStringdoSharding(StringshardingKey){// 假设shardingKey格式为: 2024-01-15Stringyear=shardingKey.substring(0,4);returnrangeMap.getOrDefault(year,"db_order_current");}}/** * 一致性Hash分片算法 */publicstaticclassConsistentHashShardingAlgorithmimplementsShardingAlgorithm{privatefinalTreeMap<Long,String>virtualNodes=newTreeMap<>();privatestaticfinalintVIRTUAL_NODE_COUNT=150;publicConsistentHashShardingAlgorithm(){// 初始化虚拟节点String[]databases={"db_node_0","db_node_1","db_node_2","db_node_3"};for(Stringdatabase:databases){for(inti=0;i<VIRTUAL_NODE_COUNT;i++){StringvirtualNode=database+"#"+i;longhash=hash(virtualNode);virtualNodes.put(hash,database);}}}@OverridepublicStringdoSharding(StringshardingKey){longhash=hash(shardingKey);SortedMap<Long,String>tailMap=virtualNodes.tailMap(hash);LongtargetHash=tailMap.isEmpty()?virtualNodes.firstKey():tailMap.firstKey();returnvirtualNodes.get(targetHash);}privatelonghash(Stringkey){returnMath.abs(key.hashCode());}}}// 分片数据源路由@ComponentpublicclassShardingDataSourceRouter{@AutowiredprivateShardingStrategyManagershardingStrategyManager;// 数据源配置privatefinalMap<String,DataSource>dataSourceMap=newConcurrentHashMap<>();@PostConstructpublicvoidinitDataSources(){// 初始化多个数据源for(inti=0;i<4;i++){StringdbName="db_user_"+i;DataSourcedataSource=createDataSource(dbName);dataSourceMap.put(dbName,dataSource);}}/** * 获取用户数据源 */publicDataSourcegetUserDataSource(LonguserId){StringtargetDatabase=shardingStrategyManager.determineTargetDatabase(String.valueOf(userId),"user_id_mod");returndataSourceMap.get(targetDatabase);}/** * 获取订单数据源 */publicDataSourcegetOrderDataSource(StringorderDate){StringtargetDatabase=shardingStrategyManager.determineTargetDatabase(orderDate,"order_date_range");returndataSourceMap.get(targetDatabase);}privateDataSourcecreateDataSource(StringdatabaseName){HikariConfigconfig=newHikariConfig();config.setJdbcUrl("jdbc:mysql://localhost:3306/"+databaseName);config.setUsername("root");config.setPassword("password");config.setMaximumPoolSize(20);config.setMinimumIdle(5);returnnewHikariDataSource(config);}}// 分片用户服务@ServicepublicclassShardedUserService{privatestaticfinalLoggerlog=LoggerFactory.getLogger(ShardedUserService.class);@AutowiredprivateShardingDataSourceRouterdataSourceRouter;/** * 根据用户ID查询用户信息 */publicUsergetUserById(LonguserId){DataSourcedataSource=dataSourceRouter.getUserDataSource(userId);try(Connectionconnection=dataSource.getConnection()){Stringsql="SELECT * FROM users WHERE user_id = ?";try(PreparedStatementstmt=connection.prepareStatement(sql)){stmt.setLong(1,userId);try(ResultSetrs=stmt.executeQuery()){if(rs.next()){returnmapResultSetToUser(rs);}}}}catch(SQLExceptione){log.error("查询用户信息失败, userId: {}",userId,e);thrownewRuntimeException("查询用户信息失败",e);}returnnull;}/** * 创建用户（自动路由到正确分片） */publicUsercreateUser(Useruser){DataSourcedataSource=dataSourceRouter.getUserDataSource(user.getUserId());try(Connectionconnection=dataSource.getConnection()){Stringsql="INSERT INTO users (user_id, username, email, created_time) VALUES (?, ?, ?, ?)";try(PreparedStatementstmt=connection.prepareStatement(sql)){stmt.setLong(1,user.getUserId());stmt.setString(2,user.getUsername());stmt.setString(3,user.getEmail());stmt.setTimestamp(4,newTimestamp(System.currentTimeMillis()));intaffectedRows=stmt.executeUpdate();if(affectedRows>0){log.info("用户创建成功, userId: {}, 目标数据库: {}",user.getUserId(),dataSource);returnuser;}}}catch(SQLExceptione){log.error("创建用户失败, userId: {}",user.getUserId(),e);thrownewRuntimeException("创建用户失败",e);}returnnull;}/** * 跨分片查询（聚合查询） */publicList<User>searchUsersAcrossShards(Stringusername){List<User>allUsers=newArrayList<>();// 并行查询所有分片List<CompletableFuture<List<User>>>futures=newArrayList<>();for(inti=0;i<4;i++){finalintshardIndex=i;CompletableFuture<List<User>>future=CompletableFuture.supplyAsync(()->{returnsearchUsersInShard(shardIndex,username);});futures.add(future);}// 聚合结果for(CompletableFuture<List<User>>future:futures){try{List<User>users=future.get(5,TimeUnit.SECONDS);allUsers.addAll(users);}catch(Exceptione){log.error("分片查询失败",e);}}returnallUsers;}privateList<User>searchUsersInShard(intshardIndex,Stringusername){List<User>users=newArrayList<>();StringdbName="db_user_"+shardIndex;DataSourcedataSource=dataSourceRouter.getDataSourceMap().get(dbName);try(Connectionconnection=dataSource.getConnection()){Stringsql="SELECT * FROM users WHERE username LIKE ?";try(PreparedStatementstmt=connection.prepareStatement(sql)){stmt.setString(1,"%"+username+"%");try(ResultSetrs=stmt.executeQuery()){while(rs.next()){users.add(mapResultSetToUser(rs));}}}}catch(SQLExceptione){log.error("分片查询失败, shard: {}, username: {}",shardIndex,username,e);}returnusers;}privateUsermapResultSetToUser(ResultSetrs)throwsSQLException{Useruser=newUser();user.setUserId(rs.getLong("user_id"));user.setUsername(rs.getString("username"));user.setEmail(rs.getString("email"));user.setCreatedTime(rs.getTimestamp("created_time"));returnuser;}}

读写分离与负载均衡

// 读写分离数据源配置@ConfigurationpublicclassReadWriteSplittingConfig{@BeanpublicDataSourcemasterDataSource(){HikariConfigconfig=newHikariConfig();config.setJdbcUrl("jdbc:mysql://master-db:3306/main_db");config.setUsername("root");config.setPassword("password");config.setMaximumPoolSize(30);returnnewHikariDataSource(config);}@BeanpublicList<DataSource>slaveDataSources(){List<DataSource>slaves=newArrayList<>();// 配置多个从库String[]slaveUrls={"jdbc:mysql://slave1-db:3306/main_db","jdbc:mysql://slave2-db:3306/main_db","jdbc:mysql://slave3-db:3306/main_db"};for(Stringurl:slaveUrls){HikariConfigconfig=newHikariConfig();config.setJdbcUrl(url);config.setUsername("root");config.setPassword("password");config.setMaximumPoolSize(20);slaves.add(newHikariDataSource(config));}returnslaves;}@BeanpublicReadWriteSplittingDataSourcereadWriteDataSource(){returnnewReadWriteSplittingDataSource(masterDataSource(),slaveDataSources());}}// 读写分离数据源实现publicclassReadWriteSplittingDataSourceimplementsDataSource{privatefinalDataSourcemasterDataSource;privatefinalList<DataSource>slaveDataSources;privatefinalAtomicIntegercounter=newAtomicInteger(0);publicReadWriteSplittingDataSource(DataSourcemasterDataSource,List<DataSource>slaveDataSources){this.masterDataSource=masterDataSource;this.slaveDataSources=slaveDataSources;}@OverridepublicConnectiongetConnection()throwsSQLException{if(isCurrentThreadReadOnly()){returngetSlaveConnection();}else{returnmasterDataSource.getConnection();}}/** * 轮询算法选择从库 */privateConnectiongetSlaveConnection()throwsSQLException{intindex=counter.getAndIncrement()%slaveDataSources.size();DataSourceselectedSlave=slaveDataSources.get(index);// 健康检查if(!isHealthy(selectedSlave)){// 选择下一个可用的从库for(inti=0;i<slaveDataSources.size();i++){intnextIndex=(index+i+1)%slaveDataSources.size();DataSourcecandidate=slaveDataSources.get(nextIndex);if(isHealthy(candidate)){returncandidate.getConnection();}}// 如果没有可用的从库，使用主库returnmasterDataSource.getConnection();}returnselectedSlave.getConnection();}privatebooleanisCurrentThreadReadOnly(){// 通过ThreadLocal或事务上下文判断是否为只读操作returnReadWriteContextHolder.isReadOnly();}privatebooleanisHealthy(DataSourcedataSource){try(Connectionconn=dataSource.getConnection()){returnconn.isValid(1);}catch(SQLExceptione){returnfalse;}}@OverridepublicConnectiongetConnection(Stringusername,Stringpassword)throwsSQLException{returngetConnection();}// 其他DataSource方法实现...}// 读写分离注解@Target(ElementType.METHOD)@Retention(RetentionPolicy.RUNTIME)public@interfaceReadWriteSplitting{Stringvalue()default"SLAVE";// MASTER, SLAVE, AUTO}// 读写分离AOP切面@Aspect@ComponentpublicclassReadWriteSplittingAspect{@Before("@annotation(readWriteSplitting)")publicvoidsetReadWriteContext(ReadWriteSplittingreadWriteSplitting){Stringroute=readWriteSplitting.value();switch(route){case"MASTER":ReadWriteContextHolder.setMaster();break;case"SLAVE":ReadWriteContextHolder.setSlave();break;case"AUTO":// 根据方法名自动判断StringmethodName=getCurrentMethodName();if(methodName.startsWith("get")||methodName.startsWith("find")||methodName.startsWith("query")){ReadWriteContextHolder.setSlave();}else{ReadWriteContextHolder.setMaster();}break;}}@After("@annotation(readWriteSplitting)")publicvoidclearReadWriteContext(){ReadWriteContextHolder.clear();}privateStringgetCurrentMethodName(){returnThread.currentThread().getStackTrace()[3].getMethodName();}}// 读写上下文持有者publicclassReadWriteContextHolder{privatestaticfinalThreadLocal<String>contextHolder=newThreadLocal<>();publicstaticvoidsetMaster(){contextHolder.set("MASTER");}publicstaticvoidsetSlave(){contextHolder.set("SLAVE");}publicstaticbooleanisReadOnly(){return"SLAVE".equals(contextHolder.get());}publicstaticvoidclear(){contextHolder.remove();}}

Redis的水平扩展

Redis Cluster架构

实践案例：Redis Cluster实现

// Redis Cluster配置@ConfigurationpublicclassRedisClusterConfig{@BeanpublicRedisClusterConfigurationredisClusterConfiguration(){RedisClusterConfigurationclusterConfig=newRedisClusterConfiguration();// 配置集群节点clusterConfig.clusterNode("redis-node1",6379);clusterConfig.clusterNode("redis-node2",6379);clusterConfig.clusterNode("redis-node3",6379);clusterConfig.clusterNode("redis-node4",6379);clusterConfig.clusterNode("redis-node5",6379);clusterConfig.clusterNode("redis-node6",6379);// 配置密码clusterConfig.setPassword(RedisPassword.of("cluster-password"));returnclusterConfig;}@BeanpublicJedisConnectionFactoryjedisConnectionFactory(){JedisPoolConfigpoolConfig=newJedisPoolConfig();poolConfig.setMaxTotal(100);poolConfig.setMaxIdle(50);poolConfig.setMinIdle(10);poolConfig.setTestOnBorrow(true);poolConfig.setTestOnReturn(true);poolConfig.setTestWhileIdle(true);returnnewJedisConnectionFactory(redisClusterConfiguration(),poolConfig);}@BeanpublicRedisTemplate<String,Object>redisTemplate(){RedisTemplate<String,Object>template=newRedisTemplate<>();template.setConnectionFactory(jedisConnectionFactory());// 配置序列化Jackson2JsonRedisSerializer<Object>serializer=newJackson2JsonRedisSerializer<>(Object.class);ObjectMappermapper=newObjectMapper();mapper.setVisibility(PropertyAccessor.ALL,JsonAutoDetect.Visibility.ANY);mapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);serializer.setObjectMapper(mapper);template.setValueSerializer(serializer);template.setKeySerializer(newStringRedisSerializer());template.afterPropertiesSet();returntemplate;}}// Redis集群操作服务@ServicepublicclassRedisClusterService{privatestaticfinalLoggerlog=LoggerFactory.getLogger(RedisClusterService.class);@AutowiredprivateRedisTemplate<String,Object>redisTemplate;@AutowiredprivateRedisClusterConfigurationclusterConfig;/** * 集群信息监控 */publicClusterInfogetClusterInfo(){returnredisTemplate.execute(connection->{StringclusterNodes=connection.clusterGetNodes();ClusterInfoinfo=parseClusterInfo(clusterNodes);log.info("Redis集群信息: {}",info);returninfo;},true);}/** * 数据分布分析 */publicMap<String,Long>analyzeDataDistribution(){Map<String,Long>distribution=newHashMap<>();// 获取每个节点的key数量List<RedisClusterNode>nodes=getClusterNodes();for(RedisClusterNodenode:nodes){if(node.isMaster()){LongkeyCount=getKeyCountForNode(node);distribution.put(node.getHost()+":"+node.getPort(),keyCount);}}returndistribution;}/** * 集群性能监控 */publicClusterPerformanceMetricsgetPerformanceMetrics(){ClusterPerformanceMetricsmetrics=newClusterPerformanceMetrics();// 获取集群节点信息List<RedisClusterNode>nodes=getClusterNodes();for(RedisClusterNodenode:nodes){NodeMetricsnodeMetrics=collectNodeMetrics(node);metrics.addNodeMetrics(nodeMetrics);}// 计算集群整体指标metrics.calculateClusterMetrics();returnmetrics;}/** * 集群扩容 */publicbooleanscaleOutCluster(List<String>newNodes){try{log.info("开始扩容Redis集群，新增节点: {}",newNodes);// 1. 添加新节点到集群for(StringnewNode:newNodes){addNodeToCluster(newNode);}// 2. 重新分配槽位reshardClusterSlots(newNodes);// 3. 验证集群状态booleanisHealthy=verifyClusterHealth();if(isHealthy){log.info("Redis集群扩容成功");returntrue;}else{log.error("Redis集群扩容后状态异常");returnfalse;}}catch(Exceptione){log.error("Redis集群扩容失败",e);returnfalse;}}/** * 集群缩容 */publicbooleanscaleInCluster(List<String>removeNodes){try{log.info("开始缩容Redis集群，移除节点: {}",removeNodes);// 1. 迁移数据到其他节点migrateDataFromNodes(removeNodes);// 2. 从集群中移除节点for(StringremoveNode:removeNodes){removeNodeFromCluster(removeNode);}// 3. 验证集群状态booleanisHealthy=verifyClusterHealth();if(isHealthy){log.info("Redis集群缩容成功");returntrue;}else{log.error("Redis集群缩容后状态异常");returnfalse;}}catch(Exceptione){log.error("Redis集群缩容失败",e);returnfalse;}}/** * 故障节点处理 */publicbooleanhandleNodeFailure(StringfailedNode){try{log.warn("处理Redis集群故障节点: {}",failedNode);// 1. 检查故障节点状态RedisClusterNodefailedClusterNode=findNodeByAddress(failedNode);if(failedClusterNode==null){log.error("未找到故障节点: {}",failedNode);returnfalse;}// 2. 如果是主节点，触发故障转移if(failedClusterNode.isMaster()){booleanfailoverSuccess=triggerFailover(failedClusterNode);if(!failoverSuccess){log.error("故障转移失败: {}",failedNode);returnfalse;}}// 3. 从集群中移除故障节点removeNodeFromCluster(failedNode);// 4. 添加新节点替换（可选）StringreplacementNode="redis-node-replacement:6379";addNodeToCluster(replacementNode);log.info("故障节点处理完成: {}",failedNode);returntrue;}catch(Exceptione){log.error("故障节点处理失败: {}",failedNode,e);returnfalse;}}// 辅助方法privateList<RedisClusterNode>getClusterNodes(){returnredisTemplate.execute(connection->{returnnewArrayList<>(connection.clusterGetNodes());},true);}privateLonggetKeyCountForNode(RedisClusterNodenode){returnredisTemplate.execute(connection->{returnconnection.dbSize();},node);}privateNodeMetricscollectNodeMetrics(RedisClusterNodenode){returnredisTemplate.execute(connection->{Propertiesinfo=connection.info();returnparseNodeMetrics(info);},node);}privatevoidaddNodeToCluster(StringnodeAddress){redisTemplate.execute(connection->{String[]parts=nodeAddress.split(":");connection.clusterMeet(parts[0],Integer.parseInt(parts[1]));returnnull;},true);}privatevoidremoveNodeFromCluster(StringnodeAddress){redisTemplate.execute(connection->{StringnodeId=getNodeIdByAddress(nodeAddress);connection.clusterForget(nodeId);returnnull;},true);}privatevoidreshardClusterSlots(List<String>newNodes){// 重新分配槽位的复杂逻辑log.info("重新分配集群槽位");// 实际实现需要考虑数据迁移、槽位平衡等}privatebooleanverifyClusterHealth(){ClusterInfoinfo=getClusterInfo();returninfo.isHealthy()&&info.getFailedNodes()==0;}privatevoidmigrateDataFromNodes(List<String>removeNodes){// 数据迁移逻辑log.info("从待移除节点迁移数据");}privateRedisClusterNodefindNodeByAddress(Stringaddress){List<RedisClusterNode>nodes=getClusterNodes();for(RedisClusterNodenode:nodes){if(address.equals(node.getHost()+":"+node.getPort())){returnnode;}}returnnull;}privatebooleantriggerFailover(RedisClusterNodefailedNode){// 触发故障转移log.info("触发故障转移，节点: {}",failedNode);returntrue;}privateStringgetNodeIdByAddress(Stringaddress){// 根据地址获取节点IDreturn"node-id-"+address;}privateClusterInfoparseClusterInfo(StringclusterNodes){// 解析集群节点信息returnnewClusterInfo();}privateNodeMetricsparseNodeMetrics(Propertiesinfo){// 解析节点性能指标returnnewNodeMetrics();}}// Redis集群性能优化@ComponentpublicclassRedisClusterOptimizer{privatestaticfinalLoggerlog=LoggerFactory.getLogger(RedisClusterOptimizer.class);@AutowiredprivateRedisClusterServiceclusterService;/** * 集群性能优化 */publicvoidoptimizeCluster(){log.info("开始Redis集群性能优化");// 1. 分析当前性能瓶颈ClusterPerformanceMetricscurrentMetrics=clusterService.getPerformanceMetrics();log.info("当前集群性能指标: {}",currentMetrics);// 2. 识别性能问题List<PerformanceIssue>issues=identifyPerformanceIssues(currentMetrics);// 3. 应用优化策略for(PerformanceIssueissue:issues){applyOptimization(issue);}// 4. 验证优化效果ClusterPerformanceMetricsoptimizedMetrics=clusterService.getPerformanceMetrics();log.info("优化后集群性能指标: {}",optimizedMetrics);// 5. 生成优化报告generateOptimizationReport(currentMetrics,optimizedMetrics);}/** * 内存优化 */publicvoidoptimizeMemoryUsage(){// 内存使用分析Map<String,MemoryUsage>memoryUsage=analyzeMemoryUsage();// 内存优化策略for(Map.Entry<String,MemoryUsage>entry:memoryUsage.entrySet()){Stringnode=entry.getKey();MemoryUsageusage=entry.getValue();if(usage.getUsagePercentage()>80){// 内存使用率过高，执行优化optimizeNodeMemory(node);}}}/** * 网络优化 */publicvoidoptimizeNetwork(){// 网络延迟分析Map<String,NetworkLatency>networkLatency=analyzeNetworkLatency();// 网络拓扑优化optimizeNetworkTopology(networkLatency);}privateList<PerformanceIssue>identifyPerformanceIssues(ClusterPerformanceMetricsmetrics){List<PerformanceIssue>issues=newArrayList<>();// 检查内存使用率if(metrics.getAverageMemoryUsage()>80){issues.add(newPerformanceIssue("HIGH_MEMORY_USAGE","内存使用率过高"));}// 检查网络延迟if(metrics.getAverageNetworkLatency()>10){issues.add(newPerformanceIssue("HIGH_NETWORK_LATENCY","网络延迟过高"));}// 检查负载不均衡if(metrics.getLoadBalanceScore()<0.8){issues.add(newPerformanceIssue("UNBALANCED_LOAD","负载分布不均衡"));}returnissues;}privatevoidapplyOptimization(PerformanceIssueissue){switch(issue.getType()){case"HIGH_MEMORY_USAGE":optimizeMemoryUsage();break;case"HIGH_NETWORK_LATENCY":optimizeNetwork();break;case"UNBALANCED_LOAD":rebalanceCluster();break;}}privatevoidrebalanceCluster(){// 重新平衡集群负载log.info("重新平衡集群负载");}privateMap<String,MemoryUsage>analyzeMemoryUsage(){// 分析内存使用情况returnnewHashMap<>();}privatevoidoptimizeNodeMemory(Stringnode){// 优化节点内存使用log.info("优化节点内存: {}",node);}privateMap<String,NetworkLatency>analyzeNetworkLatency(){// 分析网络延迟returnnewHashMap<>();}privatevoidoptimizeNetworkTopology(Map<String,NetworkLatency>latency){// 优化网络拓扑log.info("优化网络拓扑");}privatevoidgenerateOptimizationReport(ClusterPerformanceMetricsbefore,ClusterPerformanceMetricsafter){// 生成优化报告log.info("生成集群优化报告");}}// 相关数据结构@DatapublicclassClusterInfo{privateinttotalNodes;privateintmasterNodes;privateintslaveNodes;privateintfailedNodes;privatebooleanhealthy;}@DatapublicclassClusterPerformanceMetrics{privatedoubleaverageMemoryUsage;privatedoubleaverageNetworkLatency;privatedoubleloadBalanceScore;privateList<NodeMetrics>nodeMetrics;publicvoidaddNodeMetrics(NodeMetricsmetrics){if(nodeMetrics==null){nodeMetrics=newArrayList<>();}nodeMetrics.add(metrics);}publicvoidcalculateClusterMetrics(){// 计算集群整体指标if(nodeMetrics!=null&&!nodeMetrics.isEmpty()){doubletotalMemory=0;doubletotalLatency=0;for(NodeMetricsmetrics:nodeMetrics){totalMemory+=metrics.getMemoryUsage();totalLatency+=metrics.getNetworkLatency();}this.averageMemoryUsage=totalMemory/nodeMetrics.size();this.averageNetworkLatency=totalLatency/nodeMetrics.size();}}}@DatapublicclassNodeMetrics{privateStringnodeId;privatedoublememoryUsage;privatedoublenetworkLatency;privatelongconnectedClients;privatelongtotalCommandsProcessed;}@DatapublicclassPerformanceIssue{privateStringtype;privateStringdescription;publicPerformanceIssue(Stringtype,Stringdescription){this.type=type;this.description=description;}}@DatapublicclassMemoryUsage{privatelongusedMemory;privatelongtotalMemory;privatedoubleusagePercentage;}@DatapublicclassNetworkLatency{privatedoubleaverageLatency;privatedoublemaxLatency;privatedoubleminLatency;}

服务层的水平扩展

微服务架构设计

实践案例：微服务架构实现

// 服务注册与发现配置@Configuration@EnableDiscoveryClientpublicclassServiceDiscoveryConfig{@BeanpublicServiceInstanceRegistrationserviceInstanceRegistration(){returnServiceInstanceRegistration.builder().defaultUriSpec().address("localhost").port(8080).ttl(30).build();}@BeanpublicDiscoveryClientdiscoveryClient(){returnnewDefaultServiceDiscoveryClient();}}// 微服务基础架构@SpringBootApplication@EnableDiscoveryClient@EnableCircuitBreaker@EnableFeignClientspublicclassMicroserviceApplication{publicstaticvoidmain(String[]args){SpringApplication.run(MicroserviceApplication.class,args);}@BeanpublicRestTemplaterestTemplate(){returnnewRestTemplate();}}// 服务注册中心 - Eureka Server@SpringBootApplication@EnableEurekaServerpublicclassEurekaServerApplication{publicstaticvoidmain(String[]args){SpringApplication.run(EurekaServerApplication.class,args);}}// Eureka配置@ConfigurationpublicclassEurekaConfig{@BeanpublicEurekaInstanceConfigBeaneurekaInstanceConfig(InetUtilsinetUtils){EurekaInstanceConfigBeanconfig=newEurekaInstanceConfigBean(inetUtils);config.setNonSecurePort(8761);config.setIpAddress("localhost");config.setPreferIpAddress(true);config.setLeaseRenewalIntervalInSeconds(30);config.setLeaseExpirationDurationInSeconds(90);returnconfig;}}// 服务提供者@RestController@RequestMapping("/api/users")publicclassUserServiceController{privatestaticfinalLoggerlog=LoggerFactory.getLogger(UserServiceController.class);@AutowiredprivateUserServiceuserService;@Value("${server.port}")privateintserverPort;/** * 获取用户信息 */@GetMapping("/{userId}")@HystrixCommand(fallbackMethod="getUserFallback")publicResponseEntity<User>getUser(@PathVariableLonguserId){log.info("处理用户查询请求, userId: {}, 服务端口: {}",userId,serverPort);Useruser=userService.getUserById(userId);if(user!=null){user.setServerPort(serverPort);// 标记响应的服务端口returnResponseEntity.ok(user);}else{returnResponseEntity.notFound().build();}}/** * 创建用户 */@PostMapping@HystrixCommand(fallbackMethod="createUserFallback")publicResponseEntity<User>createUser(@RequestBody@ValidUserRequestrequest){log.info("处理用户创建请求, 服务端口: {}",serverPort);Useruser=userService.createUser(request);returnResponseEntity.status(HttpStatus.CREATED).body(user);}/** * 批量获取用户 */@PostMapping("/batch")@HystrixCommand(fallbackMethod="getUsersBatchFallback")publicResponseEntity<List<User>>getUsersBatch(@RequestBodyList<Long>userIds){log.info("处理批量用户查询请求, 数量: {}, 服务端口: {}",userIds.size(),serverPort);List<User>users=userService.getUsersByIds(userIds);returnResponseEntity.ok(users);}/** * 服务健康检查 */@GetMapping("/health")publicResponseEntity<Map<String,Object>>health(){Map<String,Object>health=newHashMap<>();health.put("status","UP");health.put("service","user-service");health.put("port",serverPort);health.put("timestamp",System.currentTimeMillis());returnResponseEntity.ok(health);}// 降级方法publicResponseEntity<User>getUserFallback(LonguserId){log.warn("用户查询服务降级, userId: {}",userId);UserfallbackUser=newUser();fallbackUser.setUserId(userId);fallbackUser.setUsername("Fallback User");fallbackUser.setEmail("fallback@example.com");returnResponseEntity.ok(fallbackUser);}publicResponseEntity<User>createUserFallback(UserRequestrequest){log.warn("用户创建服务降级");returnResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE).build();}publicResponseEntity<List<User>>getUsersBatchFallback(List<Long>userIds){log.warn("批量用户查询服务降级");returnResponseEntity.ok(Collections.emptyList());}}// 服务消费者 - Feign客户端@FeignClient(name="user-service",fallback=UserServiceFallback.class)publicinterfaceUserServiceClient{@GetMapping("/api/users/{userId}")ResponseEntity<User>getUser(@PathVariable("userId")LonguserId);@PostMapping("/api/users")ResponseEntity<User>createUser(@RequestBodyUserRequestrequest);@PostMapping("/api/users/batch")ResponseEntity<List<User>>getUsersBatch(@RequestBodyList<Long>userIds);@GetMapping("/api/users/health")ResponseEntity<Map<String,Object>>health();}// Feign客户端降级实现@ComponentpublicclassUserServiceFallbackimplementsUserServiceClient{privatestaticfinalLoggerlog=LoggerFactory.getLogger(UserServiceFallback.class);@OverridepublicResponseEntity<User>getUser(LonguserId){log.warn("用户服务调用失败，执行降级逻辑, userId: {}",userId);UserfallbackUser=newUser();fallbackUser.setUserId(userId);fallbackUser.setUsername("Fallback User");returnResponseEntity.ok(fallbackUser);}@OverridepublicResponseEntity<User>createUser(UserRequestrequest){log.warn("用户创建服务调用失败，执行降级逻辑");returnResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE).build();}@OverridepublicResponseEntity<List<User>>getUsersBatch(List<Long>userIds){log.warn("批量用户查询服务调用失败，执行降级逻辑");returnResponseEntity.ok(Collections.emptyList());}@OverridepublicResponseEntity<Map<String,Object>>health(){log.warn("用户服务健康检查失败");Map<String,Object>health=newHashMap<>();health.put("status","DOWN");health.put("service","user-service-fallback");returnResponseEntity.ok(health);}}// 负载均衡配置@ConfigurationpublicclassLoadBalancerConfig{@BeanpublicIRuleribbonRule(){// 配置负载均衡策略returnnewWeightedResponseTimeRule();// 基于响应时间的权重策略}@BeanpublicIPingribbonPing(){returnnewPingUrl();// 使用URL健康检查}@BeanpublicServerListSubsetFilterserverListFilter(){ServerListSubsetFilterfilter=newServerListSubsetFilter();filter.setSize(5);// 保持5个服务实例的列表returnfilter;}}// 客户端负载均衡实现@ComponentpublicclassLoadBalancedUserService{privatestaticfinalLoggerlog=LoggerFactory.getLogger(LoadBalancedUserService.class);@AutowiredprivateLoadBalancerClientloadBalancer;@AutowiredprivateRestTemplaterestTemplate;/** * 使用Ribbon进行负载均衡调用 */publicUsergetUserWithLoadBalance(LonguserId){// 选择服务实例ServiceInstanceinstance=loadBalancer.choose("user-service");if(instance==null){log.error("没有可用的用户服务实例");thrownewRuntimeException("No available user service instances");}log.info("选择服务实例: {}:{}, 用户ID: {}",instance.getHost(),instance.getPort(),userId);// 构建请求URLStringurl=String.format("http://%s:%d/api/users/%d",instance.getHost(),instance.getPort(),userId);try{// 发起请求ResponseEntity<User>response=restTemplate.getForEntity(url,User.class);Useruser=response.getBody();if(user!=null){user.setServerPort(instance.getPort());// 记录实际调用的服务端口}returnuser;}catch(Exceptione){log.error("调用用户服务失败: {}",url,e);thrownewRuntimeException("Failed to call user service",e);}}/** * 批量调用测试负载均衡效果 */publicMap<Integer,Integer>testLoadBalancing(intrequestCount){Map<Integer,Integer>portDistribution=newHashMap<>();for(inti=0;i<requestCount;i++){try{Useruser=getUserWithLoadBalance(1L);intport=user.getServerPort();portDistribution.put(port,portDistribution.getOrDefault(port,0)+1);// 模拟处理时间Thread.sleep(10);}catch(Exceptione){log.error("测试请求失败",e);}}log.info("负载均衡测试结果: {}",portDistribution);returnportDistribution;}}// 服务配置管理@Configuration@ConfigurationProperties(prefix="microservice")@DatapublicclassMicroserviceConfig{privateStringserviceName;privateStringserviceVersion;privateintinstanceId;privateMap<String,String>metadata=newHashMap<>();@PostConstructpublicvoidinit(){// 生成实例IDthis.instanceId=newRandom().nextInt(10000);metadata.put("startTime",String.valueOf(System.currentTimeMillis()));metadata.put("instanceId",String.valueOf(instanceId));}}// 服务监控与健康检查@ComponentpublicclassServiceHealthMonitor{privatestaticfinalLoggerlog=LoggerFactory.getLogger(ServiceHealthMonitor.class);@AutowiredprivateDiscoveryClientdiscoveryClient;@AutowiredprivateUserServiceClientuserServiceClient;/** * 监控所有服务实例的健康状态 */@Scheduled(fixedRate=30000)// 每30秒检查一次publicvoidmonitorServiceHealth(){List<String>services=discoveryClient.getServices();for(Stringservice:services){List<ServiceInstance>instances=discoveryClient.getInstances(service);for(ServiceInstanceinstance:instances){try{booleanisHealthy=checkInstanceHealth(instance);log.info("服务健康检查: {} - {}:{}, 状态: {}",service,instance.getHost(),instance.getPort(),isHealthy?"健康":"异常");if(!isHealthy){handleUnhealthyInstance(service,instance);}}catch(Exceptione){log.error("健康检查失败: {} - {}:{}",service,instance.getHost(),instance.getPort(),e);}}}}/** * 检查服务实例健康状态 */privatebooleancheckInstanceHealth(ServiceInstanceinstance){try{// 构建健康检查URLStringhealthUrl=String.format("http://%s:%d/api/users/health",instance.getHost(),instance.getPort());RestTemplaterestTemplate=newRestTemplate();ResponseEntity<Map>response=restTemplate.getForEntity(healthUrl,Map.class);if(response.getStatusCode()==HttpStatus.OK){Map<String,Object>health=response.getBody();return"UP".equals(health.get("status"));}returnfalse;}catch(Exceptione){log.error("健康检查请求失败: {}:{}",instance.getHost(),instance.getPort(),e);returnfalse;}}/** * 处理不健康的服务实例 */privatevoidhandleUnhealthyInstance(Stringservice,ServiceInstanceinstance){log.warn("发现不健康的服务实例: {} - {}:{}, 准备进行故障处理",service,instance.getHost(),instance.getPort());// 1. 记录故障信息recordFailure(service,instance);// 2. 通知运维人员notifyOperations(service,instance);// 3. 触发自动恢复机制（如果配置）if(shouldTriggerAutoRecovery(service,instance)){triggerAutoRecovery(service,instance);}}/** * 性能指标收集 */@Scheduled(fixedRate=60000)// 每分钟收集一次publicvoidcollectPerformanceMetrics(){List<String>services=discoveryClient.getServices();for(Stringservice:services){List<ServiceInstance>instances=discoveryClient.getInstances(service);ServiceMetricsmetrics=newServiceMetrics();metrics.setServiceName(service);metrics.setInstanceCount(instances.size());metrics.setTimestamp(System.currentTimeMillis());// 收集各实例的性能数据for(ServiceInstanceinstance:instances){InstanceMetricsinstanceMetrics=collectInstanceMetrics(instance);metrics.addInstanceMetrics(instanceMetrics);}// 存储或上报指标storeMetrics(metrics);}}privateInstanceMetricscollectInstanceMetrics(ServiceInstanceinstance){InstanceMetricsmetrics=newInstanceMetrics();metrics.setInstanceId(instance.getInstanceId());metrics.setHost(instance.getHost());metrics.setPort(instance.getPort());// 这里可以集成实际的性能监控数据// 如：CPU使用率、内存使用率、请求响应时间等returnmetrics;}privatevoidrecordFailure(Stringservice,ServiceInstanceinstance){log.error("记录服务故障: {} - {}:{}",service,instance.getHost(),instance.getPort());// 实际实现中可以写入数据库或发送到监控系统}privatevoidnotifyOperations(Stringservice,ServiceInstanceinstance){log.warn("通知运维人员: {} - {}:{} 服务异常",service,instance.getHost(),instance.getPort());// 实际实现中可以发送邮件、短信或调用告警接口}privatebooleanshouldTriggerAutoRecovery(Stringservice,ServiceInstanceinstance){// 根据配置和故障频率决定是否触发自动恢复returntrue;}privatevoidtriggerAutoRecovery(Stringservice,ServiceInstanceinstance){log.info("触发自动恢复机制: {} - {}:{}",service,instance.getHost(),instance.getPort());// 实际实现中可以调用容器编排平台的API重启服务}privatevoidstoreMetrics(ServiceMetricsmetrics){log.debug("存储服务指标: {}",metrics.getServiceName());// 实际实现中可以写入时序数据库或监控系统}}// 服务限流与熔断@ComponentpublicclassServiceProtection{privatestaticfinalLoggerlog=LoggerFactory.getLogger(ServiceProtection.class);// 限流器privatefinalRateLimiterrateLimiter=RateLimiter.create(1000);// 每秒1000个请求// 熔断器privatefinalCircuitBreakercircuitBreaker=CircuitBreaker.ofDefaults("user-service");/** * 带保护的远程调用 */public<T>TexecuteWithProtection(Supplier<T>supplier,Stringoperation){// 1. 限流检查if(!rateLimiter.tryAcquire()){log.warn("请求被限流: {}",operation);thrownewRateLimitExceededException("请求过于频繁，请稍后再试");}// 2. 熔断检查if(circuitBreaker.getState()==CircuitBreaker.State.OPEN){log.warn("服务熔断中: {}",operation);thrownewServiceUnavailableException("服务暂时不可用");}// 3. 执行调用try{Tresult=supplier.get();circuitBreaker.onSuccess();returnresult;}catch(Exceptione){circuitBreaker.onError(1,e);log.error("服务调用失败: {}",operation,e);thrownewRuntimeException("服务调用失败",e);}}/** * 自适应限流 */publicvoidadaptiveRateLimiting(){// 根据系统负载动态调整限流阈值doublecpuUsage=getSystemCpuUsage();doublememoryUsage=getSystemMemoryUsage();if(cpuUsage>80||memoryUsage>85){// 系统负载高，降低限流阈值doublenewRate=500;// 降低到每秒500个请求rateLimiter.setRate(newRate);log.warn("系统负载过高，调整限流阈值: {}",newRate);}elseif(cpuUsage<50&&memoryUsage<60){// 系统负载低，提高限流阈值doublenewRate=1500;// 提高到每秒1500个请求rateLimiter.setRate(newRate);log.info("系统负载正常，调整限流阈值: {}",newRate);}}privatedoublegetSystemCpuUsage(){// 获取系统CPU使用率return50.0;// 模拟数据}privatedoublegetSystemMemoryUsage(){// 获取系统内存使用率return60.0;// 模拟数据}}// 相关数据结构@DatapublicclassServiceMetrics{privateStringserviceName;privateintinstanceCount;privatelongtimestamp;privateList<InstanceMetrics>instanceMetrics=newArrayList<>();publicvoidaddInstanceMetrics(InstanceMetricsmetrics){instanceMetrics.add(metrics);}}@DatapublicclassInstanceMetrics{privateStringinstanceId;privateStringhost;privateintport;privatedoublecpuUsage;privatedoublememoryUsage;privatedoubleresponseTime;privateintactiveRequests;privatelongtimestamp;}@DatapublicclassUser{privateLonguserId;privateStringusername;privateStringemail;privateintserverPort;// 记录响应的服务端口privateTimestampcreatedTime;}@DatapublicclassUserRequest{@NotNullprivateStringusername;@EmailprivateStringemail;@Size(min=6,max=20)privateStringpassword;}

容器化与编排

# Docker Compose配置 - 微服务部署version:'3.8'services:# Eureka服务注册中心eureka-server:image:eureka-server:latestports:-"8761:8761"environment:-SPRING_PROFILES_ACTIVE=production-EUREKA_CLIENT_REGISTER_WITH_EUREKA=false-EUREKA_CLIENT_FETCH_REGISTRY=falsenetworks:-microservice-networkhealthcheck:test:["CMD","curl","-f","http://localhost:8761/actuator/health"]interval:30stimeout:10sretries:3# 用户服务实例1user-service-1:image:user-service:latestports:-"8081:8080"environment:-SPRING_PROFILES_ACTIVE=production-SERVER_PORT=8080-EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE=http://eureka-server:8761/eureka/-INSTANCE_ID=user-service-1depends_on:-eureka-server-mysql-master-redis-clusternetworks:-microservice-networkdeploy:replicas:1resources:limits:cpus:'0.5'memory:512Mreservations:cpus:'0.25'memory:256M# 用户服务实例2user-service-2:image:user-service:latestports:-"8082:8080"environment:-SPRING_PROFILES_ACTIVE=production-SERVER_PORT=8080-EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE=http://eureka-server:8761/eureka/-INSTANCE_ID=user-service-2depends_on:-eureka-server-mysql-master-redis-clusternetworks:-microservice-networkdeploy:replicas:1resources:limits:cpus:'0.5'memory:512Mreservations:cpus:'0.25'memory:256M# 用户服务实例3user-service-3:image:user-service:latestports:-"8083:8080"environment:-SPRING_PROFILES_ACTIVE=production-SERVER_PORT=8080-EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE=http://eureka-server:8761/eureka/-INSTANCE_ID=user-service-3depends_on:-eureka-server-mysql-master-redis-clusternetworks:-microservice-networkdeploy:replicas:1resources:limits:cpus:'0.5'memory:512Mreservations:cpus:'0.25'memory:256M# API网关api-gateway:image:api-gateway:latestports:-"80:8080"environment:-SPRING_PROFILES_ACTIVE=production-EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE=http://eureka-server:8761/eureka/-ZUUL_ROUTES_USER_SERVICE_URL=http://user-servicedepends_on:-eureka-servernetworks:-microservice-network# 负载均衡器 - Nginxnginx-lb:image:nginx:alpineports:-"8080:80"volumes:-./nginx.conf:/etc/nginx/nginx.confdepends_on:-user-service-1-user-service-2-user-service-3networks:-microservice-network# MySQL主库mysql-master:image:mysql:8.0environment:-MYSQL_ROOT_PASSWORD=root123-MYSQL_DATABASE=user_dbports:-"3306:3306"volumes:-mysql-master-data:/var/lib/mysqlnetworks:-microservice-network# MySQL从库1mysql-slave-1:image:mysql:8.0environment:-MYSQL_ROOT_PASSWORD=root123-MYSQL_DATABASE=user_dbports:-"3307:3306"volumes:-mysql-slave-1-data:/var/lib/mysqldepends_on:-mysql-masternetworks:-microservice-network# Redis集群节点1redis-node-1:image:redis:6-alpineports:-"7001:6379"command:redis-server--cluster-enabled yes--cluster-config-file nodes.conf--cluster-node-timeout 5000--appendonly yesvolumes:-redis-node-1-data:/datanetworks:-microservice-network# Redis集群节点2redis-node-2:image:redis:6-alpineports:-"7002:6379"command:redis-server--cluster-enabled yes--cluster-config-file nodes.conf--cluster-node-timeout 5000--appendonly yesvolumes:-redis-node-2-data:/datanetworks:-microservice-network# Redis集群节点3redis-node-3:image:redis:6-alpineports:-"7003:6379"command:redis-server--cluster-enabled yes--cluster-config-file nodes.conf--cluster-node-timeout 5000--appendonly yesvolumes:-redis-node-3-data:/datanetworks:-microservice-networknetworks:microservice-network:driver:bridgevolumes:mysql-master-data:mysql-slave-1-data:redis-node-1-data:redis-node-2-data:redis-node-3-data:

# Kubernetes部署配置 - 水平扩展apiVersion:apps/v1kind:Deploymentmetadata:name:user-servicenamespace:microservicesspec:replicas:3# 初始副本数selector:matchLabels:app:user-servicetemplate:metadata:labels:app:user-servicespec:containers:-name:user-serviceimage:user-service:latestports:-containerPort:8080env:-name:SPRING_PROFILES_ACTIVEvalue:"production"-name:EUREKA_CLIENT_SERVICE_URL_DEFAULTZONEvalue:"http://eureka-server:8761/eureka/"resources:requests:memory:"256Mi"cpu:"250m"limits:memory:"512Mi"cpu:"500m"livenessProbe:httpGet:path:/actuator/healthport:8080initialDelaySeconds:30periodSeconds:10readinessProbe:httpGet:path:/actuator/healthport:8080initialDelaySeconds:20periodSeconds:5---apiVersion:v1kind:Servicemetadata:name:user-servicenamespace:microservicesspec:selector:app:user-serviceports:-port:8080targetPort:8080type:ClusterIP---# 水平自动扩缩容配置apiVersion:autoscaling/v2kind:HorizontalPodAutoscalermetadata:name:user-service-hpanamespace:microservicesspec:scaleTargetRef:apiVersion:apps/v1kind:Deploymentname:user-serviceminReplicas:3maxReplicas:20metrics:-type:Resourceresource:name:cputarget:type:UtilizationaverageUtilization:70-type:Resourceresource:name:memorytarget:type:UtilizationaverageUtilization:80behavior:scaleDown:stabilizationWindowSeconds:300policies:-type:Percentvalue:10periodSeconds:60scaleUp:stabilizationWindowSeconds:60policies:-type:Percentvalue:50periodSeconds:60

水平扩展的挑战与解决方案

数据一致性挑战

实践案例：分布式事务处理

// 分布式事务管理器@ComponentpublicclassDistributedTransactionManager{privatestaticfinalLoggerlog=LoggerFactory.getLogger(DistributedTransactionManager.class);@AutowiredprivateTransactionLogRepositorytransactionLogRepository;@AutowiredprivateApplicationEventPublishereventPublisher;/** * 执行分布式事务 */@Transactionalpublic<T>TexecuteDistributedTransaction(DistributedTransaction<T>transaction){StringtransactionId=generateTransactionId();log.info("开始分布式事务, ID: {}",transactionId);try{// 1. 记录事务开始recordTransactionStart(transactionId,transaction);// 2. 执行事务Tresult=transaction.execute();// 3. 记录事务成功recordTransactionSuccess(transactionId);log.info("分布式事务执行成功, ID: {}",transactionId);returnresult;}catch(Exceptione){log.error("分布式事务执行失败, ID: {}",transactionId,e);// 4. 记录事务失败并触发补偿recordTransactionFailure(transactionId,e);triggerCompensation(transactionId);thrownewDistributedTransactionException("分布式事务执行失败",e);}}/** * TCC模式实现 */publicclassTCCDistributedTransaction{privatefinalStringtransactionId;privatefinalList<TCCParticipant>participants;publicTCCDistributedTransaction(StringtransactionId){this.transactionId=transactionId;this.participants=newArrayList<>();}publicvoidaddParticipant(TCCParticipantparticipant){participants.add(participant);}/** * Try阶段 */publicbooleantryPhase(){log.info("TCC Try阶段开始, 事务ID: {}",transactionId);List<Boolean>tryResults=newArrayList<>();for(TCCParticipantparticipant:participants){try{booleanresult=participant.tryExecute();tryResults.add(result);if(!result){log.warn("TCC Try阶段失败, 参与者: {}",participant.getName());returnfalse;}}catch(Exceptione){log.error("TCC Try阶段异常, 参与者: {}",participant.getName(),e);returnfalse;}}log.info("TCC Try阶段成功, 事务ID: {}",transactionId);returntrue;}/** * Confirm阶段 */publicbooleanconfirmPhase(){log.info("TCC Confirm阶段开始, 事务ID: {}",transactionId);for(TCCParticipantparticipant:participants){try{participant.confirm();}catch(Exceptione){log.error("TCC Confirm阶段异常, 参与者: {}",participant.getName(),e);// 记录异常但继续执行其他参与者的confirm}}log.info("TCC Confirm阶段完成, 事务ID: {}",transactionId);returntrue;}/** * Cancel阶段 */publicbooleancancelPhase(){log.info("TCC Cancel阶段开始, 事务ID: {}",transactionId);for(TCCParticipantparticipant:participants){try{participant.cancel();}catch(Exceptione){log.error("TCC Cancel阶段异常, 参与者: {}",participant.getName(),e);// 记录异常但继续执行其他参与者的cancel}}log.info("TCC Cancel阶段完成, 事务ID: {}",transactionId);returntrue;}}/** * Saga模式实现 */publicclassSagaDistributedTransaction{privatefinalStringtransactionId;privatefinalList<SagaStep>steps;privatefinalList<SagaStep>completedSteps;publicSagaDistributedTransaction(StringtransactionId){this.transactionId=transactionId;this.steps=newArrayList<>();this.completedSteps=newArrayList<>();}publicvoidaddStep(SagaStepstep){steps.add(step);}/** * 执行Saga事务 */publicbooleanexecute(){log.info("Saga事务执行开始, ID: {}",transactionId);for(SagaStepstep:steps){try{log.info("执行Saga步骤: {}",step.getName());step.execute();completedSteps.add(step);log.info("Saga步骤执行成功: {}",step.getName());}catch(Exceptione){log.error("Saga步骤执行失败: {}",step.getName(),e);// 执行补偿compensate();returnfalse;}}log.info("Saga事务执行成功, ID: {}",transactionId);returntrue;}/** * 执行补偿 */privatevoidcompensate(){log.info("开始Saga补偿, ID: {}",transactionId);// 逆序执行补偿Collections.reverse(completedSteps);for(SagaStepstep:completedSteps){try{log.info("执行Saga补偿步骤: {}",step.getName());step.compensate();}catch(Exceptione){log.error("Saga补偿步骤执行异常: {}",step.getName(),e);// 记录补偿失败，可能需要人工干预}}log.info("Saga补偿完成, ID: {}",transactionId);}}// 辅助方法privateStringgenerateTransactionId(){return"TX-"+System.currentTimeMillis()+"-"+Thread.currentThread().getId();}privatevoidrecordTransactionStart(StringtransactionId,DistributedTransaction<?>transaction){TransactionLoglog=newTransactionLog();log.setTransactionId(transactionId);log.setTransactionType(transaction.getType());log.setStatus("STARTED");log.setStartTime(LocalDateTime.now());transactionLogRepository.save(log);}privatevoidrecordTransactionSuccess(StringtransactionId){updateTransactionStatus(transactionId,"SUCCESS");}privatevoidrecordTransactionFailure(StringtransactionId,Exceptione){updateTransactionStatus(transactionId,"FAILED");// 记录异常信息}privatevoidupdateTransactionStatus(StringtransactionId,Stringstatus){TransactionLoglog=transactionLogRepository.findByTransactionId(transactionId);if(log!=null){log.setStatus(status);log.setEndTime(LocalDateTime.now());transactionLogRepository.save(log);}}privatevoidtriggerCompensation(StringtransactionId){// 发布补偿事件CompensationEventevent=newCompensationEvent(transactionId);eventPublisher.publishEvent(event);}}// TCC参与者接口publicinterfaceTCCParticipant{StringgetName();booleantryExecute();booleanconfirm();booleancancel();}// Saga步骤接口publicinterfaceSagaStep{StringgetName();voidexecute()throwsException;voidcompensate()throwsException;}// 本地消息表实现最终一致性@ComponentpublicclassLocalMessageTable{privatestaticfinalLoggerlog=LoggerFactory.getLogger(LocalMessageTable.class);@AutowiredprivateMessageRepositorymessageRepository;@AutowiredprivateRabbitTemplaterabbitTemplate;/** * 发送可靠消息 */@TransactionalpublicvoidsendReliableMessage(Stringexchange,StringroutingKey,Objectmessage){// 1. 保存消息到本地表MessageRecordrecord=newMessageRecord();record.setMessageId(generateMessageId());record.setExchange(exchange);record.setRoutingKey(routingKey);record.setMessageContent(serializeMessage(message));record.setStatus("PENDING");record.setCreatedTime(LocalDateTime.now());messageRepository.save(record);// 2. 发送消息到MQtry{rabbitTemplate.convertAndSend(exchange,routingKey,message,messagePostProcessor->{messagePostProcessor.getMessageProperties().setMessageId(record.getMessageId());returnmessagePostProcessor;});// 3. 更新消息状态record.setStatus("SENT");record.setSentTime(LocalDateTime.now());messageRepository.save(record);log.info("可靠消息发送成功, ID: {}",record.getMessageId());}catch(Exceptione){log.error("消息发送失败, ID: {}",record.getMessageId(),e);// 消息保持在PENDING状态，等待定时任务重试}}/** * 定时扫描未发送的消息 */@Scheduled(fixedRate=30000)// 每30秒扫描一次publicvoidscanPendingMessages(){log.info("开始扫描待发送消息");List<MessageRecord>pendingMessages=messageRepository.findByStatusAndCreatedTimeBefore("PENDING",LocalDateTime.now().minusMinutes(1));for(MessageRecordmessage:pendingMessages){try{resendMessage(message);}catch(Exceptione){log.error("消息重发失败, ID: {}",message.getMessageId(),e);// 更新重试次数message.incrementRetryCount();if(message.getRetryCount()>=3){message.setStatus("FAILED");}messageRepository.save(message);}}log.info("待发送消息扫描完成，处理数量: {}",pendingMessages.size());}privatevoidresendMessage(MessageRecordmessage){// 重新发送消息ObjectmessageContent=deserializeMessage(message.getMessageContent());rabbitTemplate.convertAndSend(message.getExchange(),message.getRoutingKey(),messageContent,messagePostProcessor->{messagePostProcessor.getMessageProperties().setMessageId(message.getMessageId());returnmessagePostProcessor;});// 更新状态message.setStatus("SENT");message.setSentTime(LocalDateTime.now());messageRepository.save(message);log.info("消息重发成功, ID: {}",message.getMessageId());}privateStringgenerateMessageId(){return"MSG-"+System.currentTimeMillis()+"-"+Thread.currentThread().getId();}privateStringserializeMessage(Objectmessage){try{returnnewObjectMapper().writeValueAsString(message);}catch(JsonProcessingExceptione){thrownewRuntimeException("消息序列化失败",e);}}privateObjectdeserializeMessage(StringmessageContent){try{returnnewObjectMapper().readValue(messageContent,Object.class);}catch(IOExceptione){thrownewRuntimeException("消息反序列化失败",e);}}}// 消息监听器@ComponentpublicclassReliableMessageListener{privatestaticfinalLoggerlog=LoggerFactory.getLogger(ReliableMessageListener.class);@AutowiredprivateMessageRepositorymessageRepository;@RabbitListener(queues="reliable.message.queue")publicvoidhandleMessage(Messagemessage,Channelchannel)throwsIOException{StringmessageId=message.getMessageProperties().getMessageId();try{log.info("接收到可靠消息, ID: {}",messageId);// 1. 检查消息是否已处理（幂等性）if(isMessageProcessed(messageId)){log.info("消息已处理，跳过重复处理, ID: {}",messageId);channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);return;}// 2. 处理消息processMessage(message);// 3. 标记消息已处理markMessageAsProcessed(messageId);// 4. 确认消息channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);log.info("消息处理成功, ID: {}",messageId);}catch(Exceptione){log.error("消息处理失败, ID: {}",messageId,e);// 拒绝消息并重新入队channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);}}privatebooleanisMessageProcessed(StringmessageId){returnmessageRepository.existsByMessageIdAndStatus(messageId,"PROCESSED");}privatevoidprocessMessage(Messagemessage){// 实际的消息处理逻辑StringmessageBody=newString(message.getBody());log.info("处理消息内容: {}",messageBody);// 这里可以调用具体的业务逻辑}privatevoidmarkMessageAsProcessed(StringmessageId){MessageRecordrecord=messageRepository.findByMessageId(messageId);if(record!=null){record.setStatus("PROCESSED");record.setProcessedTime(LocalDateTime.now());messageRepository.save(record);}}}// 相关数据结构@Data@Entity@Table(name="transaction_log")publicclassTransactionLog{@IdprivateStringtransactionId;privateStringtransactionType;privateStringstatus;privateLocalDateTimestartTime;privateLocalDateTimeendTime;privateStringerrorMessage;}@Data@Entity@Table(name="message_record")publicclassMessageRecord{@IdprivateStringmessageId;privateStringexchange;privateStringroutingKey;@Column(columnDefinition="TEXT")privateStringmessageContent;privateStringstatus;privateintretryCount=0;privateLocalDateTimecreatedTime;privateLocalDateTimesentTime;privateLocalDateTimeprocessedTime;publicvoidincrementRetryCount(){this.retryCount++;}}// 分布式事务异常publicclassDistributedTransactionExceptionextendsRuntimeException{publicDistributedTransactionException(Stringmessage){super(message);}publicDistributedTransactionException(Stringmessage,Throwablecause){super(message,cause);}}