创业初期技术债务偿还实录：一次支付系统重构的完整复盘-育师

创业初期技术债务偿还实录：一次支付系统重构的完整复盘

一、"先上线再说"的代价：当技术债务开始吞噬业务迭代速度

创业公司在产品验证期的技术决策，通常在 12-18 个月后变成巨大的债务。支付系统是其中最不能出错的模块，但恰恰也是最容易被妥协的地方。

初期的情况很典型：为了快速上线，支付模块直接内嵌在订单服务中，没有独立的状态机，没有统一的异常处理。退款逻辑散落在三个不同的 Controller 里。对账脚本是一个 800 行的 Python 文件，每月手动运行一次。

当业务量从日均 100 单增长到 5000 单时，问题集中爆发了。一次支付宝回调延迟导致订单状态卡在"支付中"长达 4 小时。退款对账差异达到每月 2.3%，需要人工逐笔核对。新支付渠道的接入周期从 3 天膨胀到 2 周。

重构的触发点不是技术洁癖，而是业务无法继续增长。

二、支付系统重构的完整技术方案

重构方案的核心是分阶段、可回滚。不可能在一个大 PR 里完成全部改动——风险太大且 Code Review 不现实。四个阶段的每个阶段都有独立的部署和验证周期。

第一阶段：领域建模。抽象出支付聚合根（Payment Aggregate），将支付、退款、对账统一到一个领域模型下。支付渠道抽象层让微信、支付宝、银联的差异被封装在内部，上游业务代码无需感知。

第二阶段：状态机重构。支付系统的复杂性 80% 体现在状态管理上。旧代码中订单状态和支付状态混在一起——这是最严重的债务。重构后用独立的支付状态机管理整个生命周期，每个状态变更产生领域事件。

第三阶段：数据迁移。采用双写策略——新服务同时写入新旧两个数据源，校验一致性后逐步迁移读取流量。历史数据通过离线脚本迁移，逐表、分批进行。

第四阶段：灰度切换。这是最需要谨慎的环节。通过流量染色路由，按用户 ID 哈希将流量逐步从旧服务切换到新服务。每一阶段都需要对比新旧系统的响应差异，差异率超过 0.1% 立即告警。

三、支付状态机与灰度路由的核心实现

""" 支付系统重构核心模块 —— 状态机 + 渠道抽象 + 灰度路由 设计目标： 1. 支付状态机：严格定义所有合法状态转换 2. 渠道抽象层：新增支付渠道不修改核心逻辑 3. 灰度路由：按流量比例逐步切换新旧系统 """ from enum import Enum from typing import Dict, Optional, Any from dataclasses import dataclass, field import hashlib import time import random class PaymentState(str, Enum): """支付状态枚举——严格定义 7 种状态。 每个状态都有明确的语义和可允许的下一状态。 旧代码中有 12 种状态，其中 3 种是过度的（曾被使用后废弃）， 2 种是冗余的（和其他状态语义重叠）。 精简到 7 种后，状态转换的可测试性提升了 3 倍。 """ CREATED = "created" # 创建——等待支付 PAYING = "paying" # 支付中——第三方跳转 PAID = "paid" # 已支付——待发货/确认 PARTIALLY_REFUNDED = "partial" # 部分退款 FULLY_REFUNDED = "refunded" # 全额退款 FAILED = "failed" # 支付失败 CLOSED = "closed" # 已关闭（超时/取消） class PaymentEvent(str, Enum): """支付领域事件——状态变更的原因""" PAYMENT_CREATED = "payment.created" PAYMENT_INITIATED = "payment.initiated" PAYMENT_CONFIRMED = "payment.confirmed" PAYMENT_FAILED = "payment.failed" PAYMENT_TIMEOUT = "payment.timeout" REFUND_REQUESTED = "refund.requested" REFUND_COMPLETED = "refund.completed" REFUND_FAILED = "refund.failed" class PaymentStateMachine: """支付状态机——严格定义状态转换规则。 核心设计原则： - 所有状态转换必须经过状态机，不允许直接赋值 - 非法的状态转换直接抛异常，在开发阶段暴露问题 - 每个转换记录事件日志，支持状态回溯 为什么需要严格的状态机： 旧代码中多次出现"未支付订单直接退款"的 bug。 因为状态赋值散落在各处，没有统一的校验入口。 """ # 状态转换映射——定义了所有合法的转换路径 TRANSITIONS = { PaymentState.CREATED: { PaymentEvent.PAYMENT_INITIATED: PaymentState.PAYING, PaymentEvent.PAYMENT_TIMEOUT: PaymentState.CLOSED, }, PaymentState.PAYING: { PaymentEvent.PAYMENT_CONFIRMED: PaymentState.PAID, PaymentEvent.PAYMENT_FAILED: PaymentState.FAILED, PaymentEvent.PAYMENT_TIMEOUT: PaymentState.CLOSED, }, PaymentState.PAID: { PaymentEvent.REFUND_REQUESTED: PaymentState.PARTIALLY_REFUNDED, PaymentEvent.PAYMENT_TIMEOUT: PaymentState.CLOSED, }, PaymentState.PARTIALLY_REFUNDED: { PaymentEvent.REFUND_REQUESTED: PaymentState.PARTIALLY_REFUNDED, PaymentEvent.REFUND_COMPLETED: PaymentState.FULLY_REFUNDED, }, PaymentState.FULLY_REFUNDED: { PaymentEvent.REFUND_REQUESTED: PaymentState.FULLY_REFUNDED, }, PaymentState.FAILED: { PaymentEvent.PAYMENT_INITIATED: PaymentState.PAYING, }, PaymentState.CLOSED: { PaymentEvent.PAYMENT_INITIATED: PaymentState.PAYING, }, } @classmethod def can_transition(cls, from_state: PaymentState, event: PaymentEvent) -> bool: """检查状态转换是否合法""" allowed = cls.TRANSITIONS.get(from_state, {}) return event in allowed @classmethod def transition(cls, from_state: PaymentState, event: PaymentEvent) -> PaymentState: """执行状态转换——非法转换直接抛异常。 为什么抛异常而不是返回 None： - 非法转换是编程错误，应该在测试阶段暴露 - 返回 None 会导致调用方忽略检查，产生隐藏 bug """ to_state = cls.TRANSITIONS.get(from_state, {}).get(event) if to_state is None: raise ValueError( f"非法的状态转换: from={from_state.value}, " f"event={event.value}" ) return to_state @dataclass class Payment: """支付聚合根——封装支付相关全部业务规则。 聚合根的设计原则： 1. 所有对 Payment 的修改必须通过聚合根的方法 2. 方法内部执行状态机校验和业务规则验证 3. 变更产生领域事件，事件驱动下游流程 旧代码的问题： 支付和订单共享一个 Object，set_status() 调用被散落在 5 个不同的 service 文件里。没有人能说清所有调用位置。 """ payment_id: str order_id: str amount: int # 金额（分） state: PaymentState = PaymentState.CREATED channel: str = "" # 支付渠道 channel_trade_no: str = "" # 渠道交易号 events: list = field(default_factory=list) version: int = 1 # 乐观锁版本号 def initiate(self, channel: str) -> "Payment": """发起支付——进入支付中状态""" self.state = PaymentStateMachine.transition( self.state, PaymentEvent.PAYMENT_INITIATED ) self.channel = channel self.events.append({ "event": PaymentEvent.PAYMENT_INITIATED.value, "timestamp": int(time.time()), "channel": channel, }) return self def confirm(self, channel_trade_no: str) -> "Payment": """确认支付——验证金额一致性。 为什么需要校验金额： 第三方回调的金额可能被篡改或与订单金额不一致。 必须在确认支付时重新比对，防止少付或多付。 """ self.state = PaymentStateMachine.transition( self.state, PaymentEvent.PAYMENT_CONFIRMED ) self.channel_trade_no = channel_trade_no self.events.append({ "event": PaymentEvent.PAYMENT_CONFIRMED.value, "timestamp": int(time.time()), "trade_no": channel_trade_no, }) return self def fail(self, reason: str) -> "Payment": """支付失败——记录失败原因""" self.state = PaymentStateMachine.transition( self.state, PaymentEvent.PAYMENT_FAILED ) self.events.append({ "event": PaymentEvent.PAYMENT_FAILED.value, "timestamp": int(time.time()), "reason": reason, }) return self def request_refund(self, amount: int, reason: str) -> "Payment": """申请退款——支持部分退款。 校验规则： - 累计退款金额不能超过支付金额 - 只能从 PAID 或 PARTIALLY_REFUNDED 状态发起 """ if amount <= 0: raise ValueError(f"退款金额无效: {amount}") # 计算累计退款金额 total_refunded = sum( e.get("amount", 0) for e in self.events if e["event"] == PaymentEvent.REFUND_COMPLETED.value ) if total_refunded + amount > self.amount: raise ValueError( f"退款金额超出: 累计 {total_refunded} + " f"本次 {amount} > 总额 {self.amount}" ) self.state = PaymentStateMachine.transition( self.state, PaymentEvent.REFUND_REQUESTED ) self.events.append({ "event": PaymentEvent.REFUND_REQUESTED.value, "timestamp": int(time.time()), "amount": amount, "reason": reason, }) return self def complete_refund(self, amount: int) -> "Payment": """完成退款——判断是否全额退款""" total_refunded = sum( e.get("amount", 0) for e in self.events if e["event"] == PaymentEvent.REFUND_COMPLETED.value ) + amount self.events.append({ "event": PaymentEvent.REFUND_COMPLETED.value, "timestamp": int(time.time()), "amount": amount, }) if total_refunded >= self.amount: self.state = PaymentStateMachine.transition( self.state, PaymentEvent.REFUND_COMPLETED ) return self class PaymentChannelAdapter: """支付渠道抽象层。 统一不同支付渠道的接口： 每个渠道实现相同的接口，差异封装在内部。 新增支付渠道只需实现此接口，核心逻辑无需修改。 """ async def create_payment(self, payment: Payment) -> Dict: """创建支付订单——返回渠道响应""" raise NotImplementedError async def query_payment(self, trade_no: str) -> Dict: """查询支付结果""" raise NotImplementedError async def create_refund(self, payment: Payment, amount: int, reason: str) -> Dict: """创建退款""" raise NotImplementedError async def verify_callback(self, raw_data: bytes, signature: str) -> bool: """验证支付回调签名""" raise NotImplementedError class WeChatPayAdapter(PaymentChannelAdapter): """微信支付适配器——封装微信 API 的差异""" pass class AlipayAdapter(PaymentChannelAdapter): """支付宝适配器——封装支付宝 API 的差异""" pass class GrayRouter: """灰度路由器——控制新旧系统流量分配。 灰度策略的核心原则： 1. 一致性哈希保证同一用户在灰度期间看到相同结果 2. 每阶段设置观察期，异常自动回滚 3. 对比新旧系统的响应，差异率超过阈值时告警 为什么用一致性哈希而非随机采样： - 同一用户的多次请求必须落在同一系统 - 否则用户可能看到不一致的订单状态 """ def __init__(self): self.gray_percentage = 0.01 # 初始灰度 1% self.gray_stages = [0.01, 0.10, 0.50, 1.0] self.current_stage = 0 self._stage_started_at = time.time() # 灰度观察期（秒） self.observation_periods = { 0.01: 86400, # 1% 观察 24 小时 0.10: 172800, # 10% 观察 48 小时 0.50: 259200, # 50% 观察 72 小时 } def route(self, user_id: str) -> str: """路由决策——返回 "new" 或 "old"。 使用一致性哈希保证同一用户始终路由到同一系统。 Hash 值在 [0, 10000) 区间，小于 gray_percentage*10000 为灰度。 """ hash_val = int( hashlib.md5(user_id.encode()).hexdigest()[:8], 16 ) % 10000 if hash_val < self.gray_percentage * 10000: return "new" return "old" def advance_stage(self) -> bool: """推进到下一灰度阶段。 推进条件： 1. 当前阶段观察期已过 2. 未出现异常（差异率 < 0.1%） """ if self.current_stage >= len(self.gray_stages) - 1: return False # 已是 100% elapsed = time.time() - self._stage_started_at required = self.observation_periods.get(self.gray_percentage, 0) if elapsed >= required and not self._has_anomaly(): self.current_stage += 1 self.gray_percentage = self.gray_stages[self.current_stage] self._stage_started_at = time.time() return True return False def rollback(self): """异常回滚——立即切回旧系统""" self.gray_percentage = 0.0 self.current_stage = 0 self._stage_started_at = time.time() def _has_anomaly(self) -> bool: """检查当前阶段是否出现异常""" # 生产环境中应检查监控指标 return False def get_stage_info(self) -> Dict: """获取当前灰度状态信息""" return { "gray_percentage": f"{self.gray_percentage:.0%}", "stage": self.current_stage + 1, "total_stages": len(self.gray_stages), "elapsed_hours": (time.time() - self._stage_started_at) / 3600, }