当前位置：首页 > news >正文

Kafka消费者在金融领域的深度实践：从交易处理到风险控制的完整架构

news 2025/10/26 9:06:44

在金融行业，Kafka消费者承担着处理高价值、低延迟交易数据的重任。本文将深入探讨Kafka消费者在金融领域的特殊应用，涵盖交易处理、风险控制、合规审计等关键场景。

文章目录

一、金融交易处理架构
- 1.1 高可用交易处理
- 1.2 低延迟交易优化
二、实时风险控制系统
- 2.1 复杂风险规则引擎
- 2.2 交易监控与异常检测
三、合规与审计系统
- 3.1 实时合规检查
- 3.2 完整审计追踪
四、容灾与业务连续性
- 4.1 多活数据中心架构
总结
- 核心要求
- 关键技术
- 运维保障

一、金融交易处理架构

1.1 高可用交易处理

@Component
@Primary
public class FinancialTransactionConsumer {private final TransactionValidator transactionValidator;private final RiskEngine riskEngine;private final SettlementService settlementService;private final AuditLogger auditLogger;@KafkaListener(topics = "${kafka.topics.transactions}",containerFactory = "highAvailabilityContainerFactory",concurrency = "${transaction.processor.concurrency:10}")public void processFinancialTransaction(ConsumerRecord<String, String> record) {TransactionProcessingContext context = createProcessingContext(record);try {// 1. 交易反序列化与验证FinancialTransaction transaction = deserializeAndValidate(record, context);// 2. 交易预处理TransactionPreProcessingResult preProcessResult = preProcessTransaction(transaction);if (!preProcessResult.shouldProceed()) {handlePreProcessingRejection(transaction, preProcessResult);return;}// 3. 实时风险检查RiskAssessment riskAssessment = riskEngine.assessTransaction(transaction);if (riskAssessment.getRiskLevel() == RiskLevel.HIGH) {handleHighRiskTransaction(transaction, riskAssessment);return;}// 4. 资金检查与预留FundReservationResult reservation = reserveFunds(transaction);if (!reservation.isSuccessful()) {handleInsufficientFunds(transaction, reservation);return;}// 5. 执行交易TransactionExecutionResult executionResult = executeTransaction(transaction);// 6. 资金结算SettlementResult settlementResult = settlementService.settle(transaction);// 7. 交易后处理postProcessTransaction(transaction, executionResult, settlementResult);// 8. 审计日志auditLogger.logSuccessfulTransaction(transaction, context);// 9. 发送交易确认sendTransactionConfirmation(transaction);} catch (TransactionProcessingException e) {handleTransactionProcessingFailure(record, context, e);} catch (Exception e) {handleUnexpectedError(record, context, e);}}private FinancialTransaction deserializeAndValidate(ConsumerRecord<String, String> record, TransactionProcessingContext context) {// 金融级数据验证FinancialTransaction transaction = financialMessageConverter.deserialize(record.value());// 基础验证ValidationResult basicValidation = transactionValidator.validateBasic(transaction);if (!basicValidation.isValid()) {throw new TransactionValidationException("基础验证失败", basicValidation.getErrors());}// 业务规则验证ValidationResult businessValidation = transactionValidator.validateBusinessRules(transaction);if (!businessValidation.isValid()) {throw new TransactionValidationException("业务规则验证失败", businessValidation.getErrors());}// 合规性验证ComplianceCheckResult compliance = complianceChecker.check(transaction);if (!compliance.isCompliant()) {throw new ComplianceViolationException("合规性检查失败", compliance.getViolations());}return transaction;}private TransactionExecutionResult executeTransaction(FinancialTransaction transaction) {// 使用事务性处理确保数据一致性return transactionTemplate.execute(status -> {try {// 更新账户余额accountService.updateBalance(transaction);// 记录交易流水transactionLedgerService.recordTransaction(transaction);// 更新风险暴露riskExposureService.updateExposure(transaction);return TransactionExecutionResult.success(transaction.getTransactionId());} catch (Exception e) {status.setRollbackOnly();throw new TransactionExecutionException("交易执行失败", e);}});}
}

1.2 低延迟交易优化

@Configuration
@EnableKafka
public class LowLatencyConsumerConfig {@Beanpublic ConcurrentKafkaListenerContainerFactory<String, String> lowLatencyContainerFactory() {ConcurrentKafkaListenerContainerFactory<String, String> factory =new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(lowLatencyConsumerFactory());// 低延迟优化配置factory.setBatchListener(false); // 禁用批量处理factory.setConcurrency(calculateOptimalConcurrency());ContainerProperties containerProps = factory.getContainerProperties();containerProps.setAckMode(ContainerProperties.AckMode.RECORD); // 逐条确认containerProps.setIdleEventInterval(100L); // 更短的空闲检测间隔containerProps.setPollTimeout(100L); // 更短的轮询超时// 自定义错误处理器factory.setErrorHandler(new LowLatencyErrorHandler());return factory;}@Beanpublic ConsumerFactory<String, String> lowLatencyConsumerFactory() {Map<String, Object> props = new HashMap<>();// 低延迟优化配置props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1); // 有数据立即返回props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 10); // 最大等待10msprops.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10); // 每次拉取少量记录props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 10000); // 较短的请求超时props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 15000); // 较短的会话超时props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000); // 更频繁的心跳// 网络优化props.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG, 65536);props.put(ConsumerConfig.SEND_BUFFER_CONFIG, 131072);return new DefaultKafkaConsumerFactory<>(props);}
}@Component
public class UltraLowLatencyProcessor {private final Disruptor<TransactionEvent> disruptor;private final SequenceBarrier sequenceBarrier;public UltraLowLatencyProcessor() {// 使用Disruptor模式实现超低延迟处理this.disruptor = new Disruptor<>(TransactionEvent::new,1024, // ring buffer大小DaemonThreadFactory.INSTANCE,ProducerType.MULTI,new BusySpinWaitStrategy() // 忙等待策略，最低延迟);// 设置事件处理器disruptor.handleEventsWith(new TransactionValidationHandler()).then(new RiskAssessmentHandler()).then(new ExecutionHandler());this.sequenceBarrier = disruptor.getRingBuffer().newBarrier();disruptor.start();}@KafkaListener(topics = "ultra-low-latency-transactions")public void processUltraLowLatency(ConsumerRecord<String, String> record) {long sequence = disruptor.getRingBuffer().next();try {TransactionEvent event = disruptor.getRingBuffer().get(sequence);event.setRecord(record);event.setStartTime(System.nanoTime());} finally {disruptor.getRingBuffer().publish(sequence);}}// Disruptor事件处理器public class TransactionValidationHandler implements EventHandler<TransactionEvent> {@Overridepublic void onEvent(TransactionEvent event, long sequence, boolean endOfBatch) {// 在环内执行验证，避免上下文切换event.setValidationResult(quickValidator.validate(event.getRecord()));}}
}

二、实时风险控制系统

2.1 复杂风险规则引擎

@Component
public class RealTimeRiskConsumer {private final RiskRuleEngine riskRuleEngine;private final RiskDataProvider riskDataProvider;private final AlertService alertService;@KafkaListener(topics = {"transactions", "market-data", "positions"})public void assessRealTimeRisk(ConsumerRecord<String, String> record) {RiskAssessmentContext context = createRiskContext(record);try {// 1. 风险数据收集RiskData riskData = collectRiskData(context);// 2. 风险规则评估List<RiskRuleEvaluation> evaluations = riskRuleEngine.evaluateRules(riskData);// 3. 风险聚合RiskProfile riskProfile = aggregateRisks(evaluations);// 4. 风险限额检查LimitCheckResult limitCheck = checkRiskLimits(riskProfile);if (limitCheck.isBreached()) {handleLimitBreach(riskProfile, limitCheck);}// 5. 风险报告生成RiskReport riskReport = generateRiskReport(riskProfile, evaluations);// 6. 实时风险监控monitorRealTimeRisk(riskReport);} catch (RiskAssessmentException e) {handleRiskAssessmentError(context, e);}}private RiskData collectRiskData(RiskAssessmentContext context) {return RiskData.builder()// 市场风险数据.marketData(riskDataProvider.getCurrentMarketData()).volatilityData(riskDataProvider.getVolatilitySurface()).correlationMatrix(riskDataProvider.getCorrelationMatrix())// 信用风险数据.counterpartyExposure(riskDataProvider.getCounterpartyExposure()).creditRatings(riskDataProvider.getCreditRatings())// 流动性风险数据.liquidityMetrics(riskDataProvider.getLiquidityMetrics()).fundingCosts(riskDataProvider.getFundingCosts())// 操作风险数据.operationalMetrics(riskDataProvider.getOperationalMetrics()).build();}private RiskProfile aggregateRisks(List<RiskRuleEvaluation> evaluations) {RiskAggregator aggregator = new RiskAggregator();for (RiskRuleEvaluation evaluation : evaluations) {switch (evaluation.getRiskType()) {case MARKET_RISK:aggregator.addMarketRisk(evaluation);break;case CREDIT_RISK:aggregator.addCreditRisk(evaluation);break;case LIQUIDITY_RISK:aggregator.addLiquidityRisk(evaluation);break;case OPERATIONAL_RISK:aggregator.addOperationalRisk(evaluation);break;}}return aggregator.calculateTotalRisk();}
}@Component
public class AdvancedRiskRuleEngine {private final List<RiskRule> riskRules;private final RiskRuleCompiler ruleCompiler;public List<RiskRuleEvaluation> evaluateRules(RiskData riskData) {List<RiskRuleEvaluation> evaluations = new ArrayList<>();// 并行执行风险规则评估List<CompletableFuture<RiskRuleEvaluation>> futures = riskRules.stream().map(rule -> evaluateRuleAsync(rule, riskData)).collect(Collectors.toList());// 等待所有规则评估完成CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).orTimeout(100, TimeUnit.MILLISECONDS) // 超时控制.join();// 收集结果for (CompletableFuture<RiskRuleEvaluation> future : futures) {if (future.isDone() && !future.isCompletedExceptionally()) {evaluations.add(future.join());}}return evaluations;}private CompletableFuture<RiskRuleEvaluation> evaluateRuleAsync(RiskRule rule, RiskData riskData) {return CompletableFuture.supplyAsync(() -> {try {RuleEvaluationContext context = RuleEvaluationContext.builder().riskData(riskData).rule(rule).evaluationTime(Instant.now()).build();return rule.evaluate(context);} catch (Exception e) {return RiskRuleEvaluation.failure(rule, e);}}, riskRuleExecutor);}// 动态规则加载@Scheduled(fixedRate = 60000) // 每分钟检查规则更新public void reloadRiskRules() {List<RiskRule> updatedRules = riskRuleLoader.loadLatestRules();if (!updatedRules.equals(this.riskRules)) {logger.info("检测到风险规则更新，重新加载规则");this.riskRules.clear();this.riskRules.addAll(updatedRules);ruleCompiler.compileRules(updatedRules);}}
}// 复杂风险规则实现
@Component
public class VaRCalculationRule implements RiskRule {private final VaRCalculator varCalculator;private final MonteCarloSimulator monteCarloSimulator;@Overridepublic RiskRuleEvaluation evaluate(RuleEvaluationContext context) {try {// 1. 计算投资组合VaRVaRResult varResult = varCalculator.calculateHistoricalVaR(context.getRiskData().getPortfolio(),context.getRiskData().getMarketData(),1, // 1天0.99 // 99%置信度);// 2. 压力测试StressTestResult stressTest = performStressTest(context.getRiskData().getPortfolio(),getStressScenarios());// 3. 蒙特卡洛模拟MonteCarloResult monteCarlo = monteCarloSimulator.simulate(context.getRiskData().getPortfolio(),10000 // 模拟次数);return RiskRuleEvaluation.builder().rule(this).riskType(RiskType.MARKET_RISK).severity(calculateSeverity(varResult, stressTest, monteCarlo)).metrics(buildRiskMetrics(varResult, stressTest, monteCarlo)).passed(!isVaRBreach(varResult)).build();} catch (Exception e) {return RiskRuleEvaluation.failure(this, e);}}private boolean isVaRBreach(VaRResult varResult) {return varResult.getVarAmount().compareTo(getVarLimit()) > 0;}
}

2.2 交易监控与异常检测

@Component
public class TransactionMonitoringConsumer {private final AnomalyDetector anomalyDetector;private final PatternRecognitionEngine patternEngine;private final SuspiciousActivityReporter activityReporter;@KafkaListener(topics = "transaction-events")public void monitorTransactions(ConsumerRecord<String, String> record) {TransactionEvent event = deserializeTransaction(record);try {// 1. 实时异常检测AnomalyDetectionResult anomalyResult = anomalyDetector.detectAnomalies(event);if (anomalyResult.hasAnomalies()) {handleDetectedAnomalies(event, anomalyResult);}// 2. 行为模式分析BehaviorPattern pattern = patternEngine.analyzeBehavior(event);if (pattern.isSuspicious()) {handleSuspiciousPattern(event, pattern);}// 3. 网络关系分析NetworkAnalysisResult networkAnalysis = analyzeTransactionNetwork(event);if (networkAnalysis.hasSuspiciousConnections()) {handleSuspiciousNetwork(event, networkAnalysis);}// 4. 合规规则检查ComplianceViolationResult complianceCheck = checkComplianceRules(event);if (complianceCheck.hasViolations()) {handleComplianceViolation(event, complianceCheck);}} catch (MonitoringException e) {handleMonitoringError(event, e);}}private AnomalyDetectionResult detectAnomalies(TransactionEvent event) {return AnomalyDetectionResult.builder()// 统计异常.statisticalAnomalies(statisticalDetector.detect(event))// 时间序列异常.timeSeriesAnomalies(timeSeriesDetector.detect(event))// 聚类异常.clusteringAnomalies(clusteringDetector.detect(event))// 机器学习异常.mlAnomalies(mlAnomalyDetector.detect(event)).build();}private void handleDetectedAnomalies(TransactionEvent event, AnomalyDetectionResult anomalyResult) {// 计算异常严重程度AnomalySeverity severity = calculateAnomalySeverity(anomalyResult);if (severity == AnomalySeverity.HIGH) {// 高风险异常：立即阻断交易blockTransaction(event);triggerImmediateInvestigation(event, anomalyResult);} else if (severity == AnomalySeverity.MEDIUM) {// 中等风险异常：标记并报告flagTransaction(event);reportSuspiciousActivity(event, anomalyResult);} else if (severity == AnomalySeverity.LOW) {// 低风险异常：记录供后续分析recordAnomaly(event, anomalyResult);}// 更新异常检测模型updateAnomalyDetectionModels(event, anomalyResult);}
}@Component
public class MachineLearningAnomalyDetector {private final MLModelLoader modelLoader;private final FeatureEngineeringService featureService;public MLAnomalyResult detect(TransactionEvent event) {try {// 1. 特征工程FeatureVector features = featureService.extractFeatures(event);// 2. 模型推理Map<String, MLModel> models = modelLoader.getActiveModels();List<ModelPrediction> predictions = new ArrayList<>();for (MLModel model : models.values()) {ModelPrediction prediction = model.predict(features);predictions.add(prediction);}// 3. 结果集成MLAnomalyResult result = ensemblePredictions(predictions);// 4. 模型性能监控monitorModelPerformance(event, predictions, result);return result;} catch (Exception e) {logger.error("机器学习异常检测失败", e);return MLAnomalyResult.error(e);}}private MLAnomalyResult ensemblePredictions(List<ModelPrediction> predictions) {EnsembleStrategy strategy = getEnsembleStrategy();switch (strategy) {case MAJORITY_VOTE:return majorityVote(predictions);case WEIGHTED_AVERAGE:return weightedAverage(predictions);case STACKING:return stackingEnsemble(predictions);default:return defaultEnsemble(predictions);}}@Scheduled(fixedRate = 300000) // 5分钟public void monitorModelDrift() {for (String modelId : modelLoader.getActiveModelIds()) {ModelDriftDetectionResult drift = modelDriftDetector.detectDrift(modelId);if (drift.isDriftDetected()) {logger.warn("检测到模型漂移: {}", modelId);handleModelDrift(modelId, drift);}}}
}

三、合规与审计系统

3.1 实时合规检查

@Component
public class RealTimeComplianceConsumer {private final RegulationEngine regulationEngine;private final ComplianceRuleLoader ruleLoader;private final SanctionsScreeningService sanctionsScreening;@KafkaListener(topics = "financial-transactions")public void performRealTimeComplianceCheck(ConsumerRecord<String, String> record) {FinancialTransaction transaction = deserializeTransaction(record);ComplianceCheckContext context = createComplianceContext(transaction);try {// 1. 制裁名单筛查SanctionsScreeningResult sanctionsResult = sanctionsScreening.screen(transaction);if (sanctionsResult.isMatch()) {handleSanctionsMatch(transaction, sanctionsResult);return;}// 2. 反洗钱检查AMLCheckResult amlResult = performAMLCheck(transaction);if (amlResult.isSuspicious()) {handleAMLFlag(transaction, amlResult);}// 3. 了解你的客户(KYC)验证KYCCheckResult kycResult = performKYCCheck(transaction);if (!kycResult.isVerified()) {handleKYCFailure(transaction, kycResult);}// 4. 交易限额检查LimitCheckResult limitCheck = checkTransactionLimits(transaction);if (limitCheck.isExceeded()) {handleLimitExceeded(transaction, limitCheck);}// 5. 监管规则检查RegulationCheckResult regulationCheck = regulationEngine.checkCompliance(transaction);if (!regulationCheck.isCompliant()) {handleRegulationViolation(transaction, regulationCheck);}// 6. 记录合规检查结果recordComplianceResult(transaction, context, sanctionsResult, amlResult, kycResult, limitCheck, regulationCheck);} catch (ComplianceCheckException e) {handleComplianceCheckError(transaction, context, e);}}private AMLCheckResult performAMLCheck(FinancialTransaction transaction) {return AMLCheckResult.builder()// 交易模式分析.patternAnalysis(analyzeTransactionPatterns(transaction))// 网络分析.networkAnalysis(analyzeTransactionNetwork(transaction))// 行为分析.behaviorAnalysis(analyzeCustomerBehavior(transaction))// 地理位置分析.geoAnalysis(analyzeGeographicPatterns(transaction)).build();}
}@Component
public class RegulationEngine {private final Map<String, RegulatoryRule> activeRules;private final RegulationParser regulationParser;public RegulationCheckResult checkCompliance(FinancialTransaction transaction) {List<RuleViolation> violations = new ArrayList<>();// 并行执行监管规则检查activeRules.values().parallelStream().forEach(rule -> {RuleCheckResult result = rule.check(transaction);if (!result.isCompliant()) {violations.addAll(result.getViolations());}});return RegulationCheckResult.builder().transactionId(transaction.getTransactionId()).compliant(violations.isEmpty()).violations(violations).checkedRules(activeRules.size()).checkTime(Instant.now()).build();}@EventListenerpublic void onRegulationUpdate(RegulationUpdateEvent event) {logger.info("检测到监管规则更新: {}", event.getRegulationId());try {// 解析新规则RegulatoryRule newRule = regulationParser.parse(event.getNewRuleDefinition());// 验证规则RuleValidationResult validation = ruleValidator.validate(newRule);if (!validation.isValid()) {throw new RuleValidationException("规则验证失败", validation.getErrors());}// 更新规则activeRules.put(event.getRegulationId(), newRule);logger.info("监管规则更新成功: {}", event.getRegulationId());} catch (Exception e) {logger.error("监管规则更新失败: {}", event.getRegulationId(), e);handleRuleUpdateFailure(event, e);}}
}

3.2 完整审计追踪

@Component
public class ComprehensiveAuditConsumer {private final AuditTrailService auditTrailService;private final CryptographicService cryptoService;private final ArchiveService archiveService;@KafkaListener(topics = "#{'${audit.topics}'.split(',')}")public void recordAuditTrail(ConsumerRecord<String, String> record) {AuditEvent auditEvent = createAuditEvent(record);try {// 1. 数据完整性保护String dataHash = cryptoService.calculateHash(record.value());auditEvent.setDataHash(dataHash);// 2. 数字签名String signature = cryptoService.sign(record.value());auditEvent.setDigitalSignature(signature);// 3. 时间戳服务TimestampResult timestamp = timeStampService.getTimestamp(record.value());auditEvent.setTimestampToken(timestamp.getToken());// 4. 存储审计记录AuditRecord auditRecord = auditTrailService.store(auditEvent);// 5. 实时审计分析performRealTimeAuditAnalysis(auditRecord);// 6. 长期归档if (shouldArchive(auditEvent)) {archiveService.archive(auditRecord);}} catch (AuditException e) {handleAuditFailure(record, e);}}private AuditEvent createAuditEvent(ConsumerRecord<String, String> record) {return AuditEvent.builder().eventId(generateEventId()).topic(record.topic()).partition(record.partition()).offset(record.offset()).timestamp(Instant.ofEpochMilli(record.timestamp())).key(record.key()).value(record.value()).headers(extractHeaders(record.headers())).sourceSystem(getSourceSystem(record)).eventType(classifyEventType(record)).sensitivityLevel(assessSensitivity(record)).retentionPeriod(calculateRetention(record)).build();}private void performRealTimeAuditAnalysis(AuditRecord auditRecord) {// 实时异常检测AuditAnomalyResult anomalies = auditAnomalyDetector.detect(auditRecord);if (anomalies.hasAnomalies()) {handleAuditAnomalies(auditRecord, anomalies);}// 合规性检查AuditComplianceResult compliance = auditComplianceChecker.check(auditRecord);if (!compliance.isCompliant()) {handleAuditComplianceIssue(auditRecord, compliance);}// 数据质量检查DataQualityResult dataQuality = dataQualityChecker.check(auditRecord);if (!dataQuality.isAcceptable()) {handleDataQualityIssue(auditRecord, dataQuality);}}
}@Component
public class RegulatoryReportingConsumer {private final ReportGenerator reportGenerator;private final ReportValidator reportValidator;private final SubmissionService submissionService;@KafkaListener(topics = "reporting-events")public void handleRegulatoryReporting(ConsumerRecord<String, String> record) {ReportingEvent event = deserializeReportingEvent(record);try {// 1. 报告生成RegulatoryReport report = reportGenerator.generateReport(event);// 2. 报告验证ReportValidationResult validation = reportValidator.validate(report);if (!validation.isValid()) {handleReportValidationFailure(report, validation);return;}// 3. 监管格式转换FormattedReport formattedReport = formatForRegulator(report);// 4. 数字签名SignedReport signedReport = signReport(formattedReport);// 5. 提交到监管机构SubmissionResult submission = submissionService.submit(signedReport);// 6. 确认处理handleSubmissionResult(report, submission);} catch (ReportingException e) {handleReportingFailure(event, e);}}@Scheduled(cron = "0 0 6 * * ?") // 每天6点执行public void generateDailyReports() {for (ReportType reportType : getRequiredDailyReports()) {try {logger.info("开始生成日报: {}", reportType);DailyReportRequest request = DailyReportRequest.builder().reportType(reportType).reportDate(LocalDate.now().minusDays(1)).generationTime(Instant.now()).build();// 发送报告生成请求kafkaTemplate.send("report-generation", reportType.name(), request);} catch (Exception e) {logger.error("日报生成请求失败: {}", reportType, e);}}}
}

四、容灾与业务连续性

4.1 多活数据中心架构

@Component
public class MultiDataCenterConsumer {private final DataCenterSelector dcSelector;private final ReplicationService replicationService;private final FailoverManager failoverManager;@KafkaListener(topics = "cross-dc-transactions")public void processInMultiDC(ConsumerRecord<String, String> record) {String preferredDC = dcSelector.selectDataCenter(record);if (isLocalDataCenter(preferredDC)) {// 本地处理processLocally(record);} else {// 路由到其他数据中心routeToRemoteDC(record, preferredDC);}// 异步复制到其他数据中心replicateToOtherDCs(record);}@EventListenerpublic void onDataCenterFailure(DataCenterFailureEvent event) {logger.warn("数据中心故障: {}", event.getDataCenterId());// 触发故障转移FailoverPlan failoverPlan = failoverManager.createFailoverPlan(event);executeFailover(failoverPlan);}private void executeFailover(FailoverPlan plan) {try {// 1. 停止故障数据中心的消费者stopConsumers(plan.getFailedDC());// 2. 重新分配分区reassignPartitions(plan);// 3. 更新配置updateConsumerConfigs(plan);// 4. 启动备用数据中心的消费者startConsumers(plan.getBackupDC());// 5. 验证故障转移verifyFailover(plan);logger.info("故障转移完成: {} -> {}", plan.getFailedDC(), plan.getBackupDC());} catch (Exception e) {logger.error("故障转移失败", e);escalateFailoverFailure(plan, e);}}
}@Component
public class DisasterRecoveryConsumer {private final BackupService backupService;private final RecoveryPointManager recoveryPointManager;@Scheduled(fixedRate = 300000) // 5分钟public void createRecoveryPoint() {try {// 1. 创建恢复点RecoveryPoint recoveryPoint = recoveryPointManager.createRecoveryPoint();// 2. 备份关键状态backupService.backupConsumerState(recoveryPoint);// 3. 验证备份完整性BackupVerificationResult verification = backupService.verifyBackup(recoveryPoint);if (!verification.isValid()) {throw new BackupException("备份验证失败", verification.getErrors());}// 4. 记录恢复点recoveryPointManager.recordRecoveryPoint(recoveryPoint);logger.info("恢复点创建成功: {}", recoveryPoint.getPointId());} catch (Exception e) {logger.error("恢复点创建失败", e);handleBackupFailure(e);}}@EventListenerpublic void onDisasterRecovery(DisasterRecoveryEvent event) {logger.info("开始灾难恢复流程");try {// 1. 获取最新恢复点RecoveryPoint recoveryPoint = recoveryPointManager.getLatestValidRecoveryPoint();// 2. 恢复消费者状态backupService.restoreConsumerState(recoveryPoint);// 3. 重新启动消费者restartConsumersWithRecoveryState();// 4. 验证恢复结果verifyRecovery(recoveryPoint);logger.info("灾难恢复完成");} catch (Exception e) {logger.error("灾难恢复失败", e);escalateDisasterRecoveryFailure(e);}}
}