如何设计一个支付系统
·
如何设计一个支付系统
📌 一、题目解读与需求澄清
本质问题：
设计一个能够安全、可靠地处理资金转移的分布式系统，确保每一笔交易都被准确记录，且在任何故障场景下都不会丢失或重复扣款。
功能性需求（必须向面试官确认）
- 支付场景范围：是电商内部钱包系统、第三方支付网关（类似 Stripe），还是完整的支付平台（类似支付宝）？
- 支付方式：是否需要支持信用卡、借记卡、银行转账、数字钱包等多种方式？还是只处理内部账户余额扣减?
- 核心功能边界：
- 是否需要支持退款、部分退款？
- 是否需要预授权（先冻结资金，后确认扣款）？
- 是否需要分账（一笔订单分给多个商家）？
- 用户类型：C端用户支付、B端商家收款、平台抽成的三方模型？
- 交易流程：同步返回结果还是异步通知？超时时间限制？
非功能性需求
- 强一致性：钱绝不能多扣、少扣，必须保证账户余额的绝对准确
- 高可用性：支付系统 Downtime 直接等于收入损失，要求 99.99% 可用性
- 幂等性：网络重试、用户重复点击都不能导致重复扣款
- 低延迟：用户感知的支付确认时间需控制在 3 秒内
- 可审计性：所有资金流动必须有完整日志，支持对账和监管
容易忽略的隐藏需求
- 对账系统：每日与银行/第三方支付机构对账，发现并修复差异
- 异常处理：支付中断后的恢复机制（比如用户支付后网络断开）
- 风控系统：异常交易检测（短时间大量小额支付、跨国大额支付）
- 分布式事务：库存扣减 + 订单创建 + 账户扣款需要协调一致
- 汇率转换：跨境支付的实时汇率问题
📐 二、规模估算（Back-of-the-Envelope Estimation）
假设参数
| 指标 | 假设值 | 说明 |
|---|---|---|
| DAU | 5000万 | 日活用户 |
| 每用户日均支付次数 | 0.2次 | 并非所有用户每天都支付 |
| 日支付订单量 | 1000万 | 50M × 0.2 |
| 峰值倍率 | 5倍 | 促销活动（如双11） |
QPS 计算
写 QPS（支付请求）：
平均写 QPS = 10,000,000 / 86400 ≈ 116 QPS
峰值写 QPS = 116 × 5 = 580 QPS
读 QPS（查询订单、余额）：
假设读写比 = 10:1
平均读 QPS ≈ 1,160 QPS
峰值读 QPS ≈ 5,800 QPS
存储需求
单条支付记录大小估算：
- 订单 ID: 8 字节
- 用户 ID: 8 字节
- 商户 ID: 8 字节
- 金额: 8 字节
- 时间戳: 8 字节
- 状态/支付方式等: 100 字节
- 总计: ~140 字节
每日存储：
10,000,000 × 140 字节 ≈ 1.4 GB/天
年存储（含索引和日志，3倍冗余）：
1.4 GB × 365 × 3 ≈ 1.5 TB/年
带宽需求
峰值写入: 580 QPS × 140 字节 ≈ 81 KB/s（可忽略）
峰值读取: 5,800 QPS × 500 字节 ≈ 2.9 MB/s（含查询结果）
关键结论
- 写多读少场景：但读的并发量仍然不低（订单详情、支付状态查询）
- 存储压力可控：传统关系型数据库配合分库分表可满足
- 核心挑战在于事务一致性而非性能：500+ TPS 的强一致写入需要精心设计
🏗️ 三、高层架构设计（High-Level Design）
核心组件图
┌─────────────┐
│ 用户端 │
│ (App/Web) │
└──────┬──────┘
│
│ HTTPS
▼
┌─────────────────────────────────────────┐
│ API Gateway + 限流 │
└──────────────┬──────────────────────────┘
│
▼
┌──────────────────────────────────────────┐
│ 支付服务层(Payment Service) │
│ ┌──────────┐ ┌──────────┐ ┌─────────┐│
│ │ 预下单 │ │ 支付执行 │ │ 查询 ││
│ │ Service │ │ Service │ │ Service ││
│ └──────────┘ └──────────┘ └─────────┘│
└──────┬───────────────┬───────────────────┘
│ │
│ │
┌──────▼───────┐ ┌───▼────────────┐
│ 订单数据库 │ │ 账户数据库 │
│ (分库分表) │ │ (强一致性) │
└──────────────┘ └────────┬────────┘
│
┌────────▼────────┐
│ 流水日志表 │
│ (Append-Only) │
└─────────────────┘
│
▼
┌────────────────────────────┐
│ 第三方支付渠道 │
│ (微信/支付宝/银行) │
└────────────────────────────┘
│
▼
┌────────────────────────────┐
│ 异步处理层 │
│ ┌──────┐ ┌──────┐ │
│ │ 对账 │ │ 风控 │ │
│ └──────┘ └──────┘ │
└────────────────────────────┘
核心组件职责
-
API Gateway
- 统一入口，SSL 终止
- 限流熔断（防止单用户恶意请求）
- 请求日志记录
-
支付服务层
- 预下单 Service：创建支付订单，生成唯一流水号
- 支付执行 Service：调用第三方支付，执行账户扣款
- 查询 Service：提供订单状态查询（给用户端和商户端）
-
订单数据库
- 存储支付订单主表
- 按用户 ID 或订单 ID 分库分表
-
账户数据库
- 用户余额表（需要强一致性保证）
- 使用行级锁或乐观锁防止并发问题
-
流水日志表
- 所有账户变动的 Append-Only 日志
- 用于对账和审计
-
异步处理层
- 对账任务：每日凌晨与第三方对账
- 风控任务：实时/离线分析异常交易
核心数据流
支付流程（成功路径）：
- 用户点击"支付" → API Gateway 接收请求
- 预下单 Service 生成订单记录（状态=PENDING）
- 支付执行 Service 调用第三方支付接口（如支付宝）
- 第三方返回成功 → 更新订单状态为 SUCCESS
- 记录流水日志（用户账户 -100元）
- 返回用户"支付成功"
异步回调流程：
- 第三方支付发送异步通知到回调接口
- 验证签名 + 幂等性检查（避免重复处理）
- 更新订单状态 + 记录流水
- 返回 200 OK 给第三方
核心数据库表设计
支付订单表（payment_order）
CREATE TABLE payment_order (
order_id BIGINT PRIMARY KEY, -- 雪花算法生成
user_id BIGINT NOT NULL,
merchant_id BIGINT NOT NULL,
amount DECIMAL(15,2) NOT NULL, -- 金额(单位:元)
currency VARCHAR(3) DEFAULT 'CNY',
status ENUM('PENDING','SUCCESS','FAILED','REFUNDED'),
payment_method VARCHAR(20), -- 'alipay','wechat','card'
third_party_txn_id VARCHAR(64), -- 第三方交易号
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP ON UPDATE NOW(),
INDEX idx_user_id (user_id),
INDEX idx_created_at (created_at)
) ENGINE=InnoDB;
账户余额表（account_balance）
CREATE TABLE account_balance (
account_id BIGINT PRIMARY KEY,
user_id BIGINT UNIQUE NOT NULL,
balance DECIMAL(15,2) NOT NULL,
version INT NOT NULL DEFAULT 0, -- 乐观锁版本号
updated_at TIMESTAMP ON UPDATE NOW(),
INDEX idx_user_id (user_id)
) ENGINE=InnoDB;
流水日志表（transaction_log）
CREATE TABLE transaction_log (
log_id BIGINT PRIMARY KEY AUTO_INCREMENT,
order_id BIGINT NOT NULL,
account_id BIGINT NOT NULL,
amount DECIMAL(15,2) NOT NULL, -- 正数=收入,负数=支出
balance_before DECIMAL(15,2),
balance_after DECIMAL(15,2),
txn_type ENUM('PAYMENT','REFUND','WITHDRAWAL'),
created_at TIMESTAMP DEFAULT NOW(),
INDEX idx_order_id (order_id),
INDEX idx_account_id_created (account_id, created_at)
) ENGINE=InnoDB;
🔬 四、深入细节（Deep Dive）
子问题1：如何保证支付的幂等性？
问题描述：
用户重复点击支付按钮、网络重试、前端多次提交，都可能导致同一笔订单被扣款多次。必须确保相同的支付请求只处理一次。
解决方案分析：
方案A：客户端生成唯一请求 ID（推荐）
- 客户端在发起支付时生成 UUID 作为
idempotency_key - 服务端收到请求后,先查询该 key 是否已处理
- 如果已处理,直接返回之前的结果;如果未处理,执行支付逻辑
方案B：基于订单ID + 状态机
- 订单状态只能单向流转: PENDING → SUCCESS/FAILED
- 在数据库层面用唯一索引防止重复创建
推荐方案及理由：
结合两种方案:
- 使用客户端生成的
idempotency_key作为第一道防线 - 订单状态机 + 数据库唯一约束作为第二道防线
- 分布式锁(Redis)防止并发处理同一订单
关键代码示例：
import redis
import uuid
from decimal import Decimal
class PaymentService:
def __init__(self):
self.redis = redis.Redis()
self.db = Database() # 假设的数据库连接
def process_payment(self, user_id, amount, idempotency_key=None):
"""
处理支付请求(幂等设计)
"""
# 1. 生成或验证幂等性key
if not idempotency_key:
idempotency_key = str(uuid.uuid4())
# 2. 检查该key是否已处理过
cache_key = f"payment:idempotency:{idempotency_key}"
cached_result = self.redis.get(cache_key)
if cached_result:
return json.loads(cached_result) # 返回之前的结果
# 3. 分布式锁防止并发(5秒超时)
lock_key = f"payment:lock:{user_id}"
lock = self.redis.set(lock_key, "1", nx=True, ex=5)
if not lock:
return {"status": "PROCESSING", "msg": "请勿重复提交"}
try:
# 4. 创建订单(数据库层面保证唯一性)
order_id = self._generate_order_id()
# 5. 执行支付逻辑(调用第三方或扣减余额)
result = self._execute_payment(user_id, amount, order_id)
# 6. 缓存结果(24小时过期)
self.redis.setex(cache_key, 86400, json.dumps(result))
return result
finally:
# 7. 释放锁
self.redis.delete(lock_key)
def _execute_payment(self, user_id, amount, order_id):
"""
实际执行支付(带事务)
"""
with self.db.transaction():
# 查询账户并加行锁(SELECT FOR UPDATE)
account = self.db.query(
"SELECT balance, version FROM account_balance "
"WHERE user_id = %s FOR UPDATE",
[user_id]
).fetchone()
if account['balance'] < amount:
raise InsufficientBalance("余额不足")
# 乐观锁更新余额
affected = self.db.execute(
"UPDATE account_balance "
"SET balance = balance - %s, version = version + 1 "
"WHERE user_id = %s AND version = %s",
[amount, user_id, account['version']]
)
if affected == 0:
raise ConcurrentUpdateError("并发冲突,请重试")
# 记录流水日志
self.db.execute(
"INSERT INTO transaction_log "
"(order_id, account_id, amount, balance_before, balance_after) "
"VALUES (%s, %s, %s, %s, %s)",
[order_id, account['account_id'], -amount,
account['balance'], account['balance'] - amount]
)
# 更新订单状态
self.db.execute(
"UPDATE payment_order SET status = 'SUCCESS' "
"WHERE order_id = %s",
[order_id]
)
return {"status": "SUCCESS", "order_id": order_id}
子问题2：如何处理分布式事务(订单+支付+库存)?
问题描述：
电商场景下,一次购买需要同时完成:
- 创建订单
- 扣减库存
- 扣款
这三个操作可能分布在不同的微服务和数据库中,如何保证一致性?
解决方案分析：
方案A：2PC(两阶段提交)
- 优点:强一致性
- 缺点:性能差,协调者单点故障,不适合高并发支付
方案B：TCC(Try-Confirm-Cancel)
- 优点:性能较好,支持补偿
- 缺点:业务侵入性强,需要实现三个接口
方案C：Saga 模式(推荐)
- 将分布式事务拆成一系列本地事务
- 每个本地事务有对应的补偿操作
- 通过消息队列串联
方案D：最终一致性(异步补偿)
- 支付成功后发消息队列
- 下游服务监听消息完成订单创建
- 失败时人工介入或自动重试
推荐方案:
支付场景用 Saga + 消息队列,因为:
- 支付本身必须是强一致的(账户扣款不能有问题)
- 但订单创建、库存扣减可以接受最终一致(通过补偿机制)
关键代码示例(Saga编排):
from enum import Enum
class SagaStep(Enum):
CREATE_ORDER = 1
DEDUCT_INVENTORY = 2
EXECUTE_PAYMENT = 3
class PaymentSaga:
def __init__(self):
self.mq = MessageQueue()
self.completed_steps = []
def execute(self, user_id, product_id, amount):
"""
执行支付Saga流程
"""
try:
# Step 1: 创建订单(预留状态)
order_id = self._create_pending_order(user_id, product_id, amount)
self.completed_steps.append(SagaStep.CREATE_ORDER)
# Step 2: 扣减库存(预占)
self._reserve_inventory(product_id, quantity=1)
self.completed_steps.append(SagaStep.DEDUCT_INVENTORY)
# Step 3: 执行支付(关键步骤)
payment_result = self._execute_payment(user_id, amount, order_id)
self.completed_steps.append(SagaStep.EXECUTE_PAYMENT)
# 全部成功,确认订单和库存
self._confirm_order(order_id)
self._confirm_inventory(product_id)
return {"status": "SUCCESS", "order_id": order_id}
except Exception as e:
# 任何步骤失败,执行补偿(反向回滚)
self._compensate()
raise e
def _compensate(self):
"""
补偿机制:按相反顺序撤销已完成的步骤
"""
for step in reversed(self.completed_steps):
if step == SagaStep.EXECUTE_PAYMENT:
# 退款
self._refund_payment()
elif step == SagaStep.DEDUCT_INVENTORY:
# 释放库存
self._release_inventory()
elif step == SagaStep.CREATE_ORDER:
# 取消订单
self._cancel_order()
def _execute_payment(self, user_id, amount, order_id):
"""
执行支付(调用支付服务)
"""
response = requests.post("http://payment-service/pay", json={
"user_id": user_id,
"amount": amount,
"order_id": order_id
})
if response.status_code != 200:
raise PaymentFailedException("支付失败")
return response.json()
def _reserve_inventory(self, product_id, quantity):
"""
预占库存(通过Redis原子操作)
"""
redis_key = f"inventory:reserved:{product_id}"
current = self.redis.decr(redis_key, quantity)
if current < 0:
# 回滚
self.redis.incr(redis_key, quantity)
raise InsufficientInventory("库存不足")
异步补偿方案(更常用):
# 支付成功后发送消息
def on_payment_success(order_id, user_id, amount):
message = {
"event": "PAYMENT_SUCCESS",
"order_id": order_id,
"user_id": user_id,
"amount": amount,
"timestamp": time.time()
}
# 发送到消息队列(保证至少一次投递)
kafka_producer.send("payment-events", message)
# 订单服务监听消息
def handle_payment_event(message):
order_id = message['order_id']
# 幂等性检查
if is_order_already_confirmed(order_id):
return
try:
# 确认订单
update_order_status(order_id, 'PAID')
# 扣减库存
deduct_inventory(message['product_id'])
except Exception as e:
# 失败重试(消息队列会自动重投)
logger.error(f"订单处理失败: {e}")
raise # 触发重试
子问题3: 如何设计对账系统?
问题描述:
支付系统与第三方支付(支付宝/微信)之间,因为网络、系统故障等原因,可能出现账目不一致。需要每日对账发现并修复差异。
解决方案分析:
对账的核心是三方数据对比:
- 我方订单数据库
- 我方流水日志
- 第三方支付账单
对账流程:
每日凌晨2点(低峰期)
↓
下载第三方对账单(CSV/TXT文件)
↓
解析并入库到 reconciliation_record 表
↓
与本地订单表逐条比对
↓
发现差异 → 记录到 reconciliation_diff 表
↓
人工审核 + 自动补单/退款
差异类型:
| 差异类型 | 描述 | 处理方式 |
|---|---|---|
| 长款 | 第三方有记录,我方无 | 补单(可能是异步回调丢失) |
| 短款 | 我方有记录,第三方无 | 查询第三方,可能需退款 |
| 金额不一致 | 订单号匹配但金额不同 | 人工审核 |
| 状态不一致 | 我方SUCCESS,第三方FAILED | 以第三方为准,触发退款 |
关键代码示例:
import csv
from datetime import datetime, timedelta
class ReconciliationService:
def __init__(self):
self.db = Database()
self.third_party = ThirdPartyAPI()
def daily_reconciliation(self, date):
"""
执行每日对账
"""
# 1. 下载第三方对账文件
bill_file = self.third_party.download_bill(date)
# 2. 解析对账文件
third_party_records = self._parse_bill_file(bill_file)
# 3. 查询本地订单
local_orders = self._fetch_local_orders(date)
# 4. 比对并记录差异
diffs = self._compare_records(local_orders, third_party_records)
# 5. 自动处理部分差异
self._auto_resolve(diffs)
# 6. 生成对账报告
report = self._generate_report(diffs)
return report
def _parse_bill_file(self, file_path):
"""
解析第三方对账文件(CSV格式示例)
"""
records = {}
with open(file_path, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
# 第三方交易号作为key
txn_id = row['transaction_id']
records[txn_id] = {
'amount': Decimal(row['amount']),
'status': row['status'],
'trade_time': row['trade_time'],
'order_id': row.get('out_trade_no') # 我方订单号
}
return records
def _fetch_local_orders(self, date):
"""
查询本地当日所有订单
"""
start_time = f"{date} 00:00:00"
end_time = f"{date} 23:59:59"
sql = """
SELECT order_id, third_party_txn_id, amount, status, created_at
FROM payment_order
WHERE created_at BETWEEN %s AND %s
AND status != 'PENDING'
"""
rows = self.db.query(sql, [start_time, end_time]).fetchall()
# 转为字典(以第三方交易号为key)
orders = {}
for row in rows:
if row['third_party_txn_id']:
orders[row['third_party_txn_id']] = row
return orders
def _compare_records(self, local_orders, third_party_records):
"""
核心对比逻辑
"""
diffs = []
# 遍历第三方记录
for txn_id, third_party in third_party_records.items():
local = local_orders.get(txn_id)
if not local:
# 长款:第三方有,我方无
diffs.append({
'type': 'EXTRA_IN_THIRD_PARTY',
'txn_id': txn_id,
'third_party_amount': third_party['amount'],
'order_id': third_party.get('order_id')
})
elif local['amount'] != third_party['amount']:
# 金额不一致
diffs.append({
'type': 'AMOUNT_MISMATCH',
'txn_id': txn_id,
'local_amount': local['amount'],
'third_party_amount': third_party['amount']
})
elif self._status_mismatch(local['status'], third_party['status']):
# 状态不一致
diffs.append({
'type': 'STATUS_MISMATCH',
'txn_id': txn_id,
'local_status': local['status'],
'third_party_status': third_party['status']
})
# 遍历本地记录查找短款
for txn_id, local in local_orders.items():
if txn_id not in third_party_records:
# 短款:我方有,第三方无
diffs.append({
'type': 'MISSING_IN_THIRD_PARTY',
'txn_id': txn_id,
'order_id': local['order_id'],
'local_amount': local['amount']
})
return diffs
def _auto_resolve(self, diffs):
"""
自动处理部分可修复的差异
"""
for diff in diffs:
if diff['type'] == 'EXTRA_IN_THIRD_PARTY':
# 长款:可能是异步回调丢失,查询订单是否存在
order_id = diff.get('order_id')
if order_id:
# 补单逻辑
self._补单(order_id, diff['txn_id'], diff['third_party_amount'])
elif diff['type'] == 'STATUS_MISMATCH':
# 状态不一致:以第三方为准
if diff['third_party_status'] == 'FAILED':
# 触发退款
self._trigger_refund(diff['txn_id'])
def _generate_report(self, diffs):
"""
生成对账报告
"""
return {
'date': datetime.now().strftime('%Y-%m-%d'),
'total_diffs': len(diffs),
'auto_resolved': sum(1 for d in diffs if d.get('resolved')),
'need_manual_review': [d for d in diffs if not d.get('resolved')]
}
子问题4: 如何设计安全的支付接口?
问题描述:
支付接口是黑客攻击的重点目标,需要防止:
- 重放攻击
- 中间人篡改
- 恶意刷单
- 金额篡改
解决方案:
1. HTTPS + 证书绑定
所有支付请求必须通过HTTPS,后端验证客户端证书
2. 请求签名机制
签名算法: HMAC-SHA256
签名内容: 请求参数按字母序排列 + 时间戳 + 随机数 + 密钥
3. 时间戳防重放
def verify_request(params, signature, timestamp):
# 1. 检查时间戳(5分钟内有效)
if abs(time.time() - timestamp) > 300:
raise RequestExpired("请求已过期")
# 2. 检查签名
expected_sign = generate_signature(params, timestamp, SECRET_KEY)
if signature != expected_sign:
raise InvalidSignature("签名验证失败")
# 3. 防重放(Redis记录已处理的签名)
redis_key = f"sign:used:{signature}"
if redis.exists(redis_key):
raise ReplayAttack("重复请求")
redis.setex(redis_key, 300, "1") # 5分钟过期
4. 金额校验
# 前端提交的金额仅作展示,后端重新计算
def validate_payment_amount(order_id, client_amount):
# 从数据库查询订单真实金额
real_amount = db.query(
"SELECT total_amount FROM orders WHERE id = %s",
[order_id]
).fetchone()['total_amount']
# 容差校验(允许1分钱误差,防止浮点精度问题)
if abs(real_amount - client_amount) > 0.01:
raise AmountMismatch("金额校验失败")
return real_amount # 使用服务端计算的金额
5. 风控规则
class RiskControl:
def check_transaction(self, user_id, amount, ip):
"""
实时风控检查
"""
# 规则1: 单笔限额
if amount > 50000:
return self._manual_review("大额交易")
# 规则2: 频率限制(1分钟内不超过3笔)
recent_count = redis.zcount(
f"txn:freq:{user_id}",
time.time() - 60,
time.time()
)
if recent_count >= 3:
return self._block("频繁交易")
# 规则3: IP异常检测(短时间内切换多个国家)
last_country = redis.get(f"user:country:{user_id}")
current_country = self._ip_to_country(ip)
if last_country and last_country != current_country:
return self._manual_review("异地登录")
# 规则4: 黑名单检查
if self._is_blacklisted(user_id, ip):
return self._block("黑名单用户")
# 记录本次交易
redis.zadd(f"txn:freq:{user_id}", {time.time(): time.time()})
redis.setex(f"user:country:{user_id}", 3600, current_country)
return {"status": "PASS"}
子问题5: 如何处理退款逻辑?
问题描述:
退款需要考虑:
- 部分退款(订单100元,退50元)
- 多次退款
- 退款到原支付方式
- 退款失败的处理
解决方案:
退款表设计:
CREATE TABLE refund_order (
refund_id BIGINT PRIMARY KEY,
original_order_id BIGINT NOT NULL, -- 原支付订单
user_id BIGINT NOT NULL,
refund_amount DECIMAL(15,2) NOT NULL,
refund_reason VARCHAR(255),
status ENUM('PENDING','SUCCESS','FAILED'),
third_party_refund_id VARCHAR(64),
created_at TIMESTAMP DEFAULT NOW(),
INDEX idx_original_order (original_order_id)
);
退款约束检查:
def create_refund(order_id, refund_amount, reason):
"""
创建退款单(需检查约束)
"""
# 1. 查询原订单
order = db.query(
"SELECT amount, status FROM payment_order WHERE order_id = %s",
[order_id]
).fetchone()
if order['status'] != 'SUCCESS':
raise InvalidRefund("订单状态不允许退款")
# 2. 计算已退款总额
refunded_total = db.query(
"SELECT COALESCE(SUM(refund_amount), 0) as total "
"FROM refund_order "
"WHERE original_order_id = %s AND status = 'SUCCESS'",
[order_id]
).fetchone()['total']
# 3. 校验退款金额
if refunded_total + refund_amount > order['amount']:
raise ExceedRefundLimit(
f"退款总额超限: 已退{refunded_total}, 本次{refund_amount}, "
f"订单金额{order['amount']}"
)
# 4. 创建退款单
refund_id = generate_id()
db.execute(
"INSERT INTO refund_order "
"(refund_id, original_order_id, user_id, refund_amount, "
" refund_reason, status) "
"VALUES (%s, %s, %s, %s, %s, 'PENDING')",
[refund_id, order_id, order['user_id'], refund_amount, reason]
)
# 5. 异步调用第三方退款
async_refund.delay(refund_id)
return refund_id
def process_refund(refund_id):
"""
执行退款(调用第三方)
"""
refund = db.query(
"SELECT * FROM refund_order WHERE refund_id = %s",
[refund_id]
).fetchone()
try:
# 调用第三方退款接口
result = third_party_api.refund(
original_txn_id=refund['third_party_txn_id'],
amount=refund['refund_amount']
)
# 更新退款状态
db.execute(
"UPDATE refund_order "
"SET status = 'SUCCESS', third_party_refund_id = %s "
"WHERE refund_id = %s",
[result['refund_id'], refund_id]
)
# 记录流水(退款是正数)
log_transaction(
order_id=refund['original_order_id'],
amount=refund['refund_amount'], # 正数
txn_type='REFUND'
)
except Exception as e:
# 退款失败,记录日志并重试
db.execute(
"UPDATE refund_order SET status = 'FAILED' "
"WHERE refund_id = %s",
[refund_id]
)
logger.error(f"退款失败: {e}")
# 加入重试队列
retry_queue.put(refund_id)
子问题6: 如何设计高可用架构?
问题描述:
支付系统宕机直接等于收入损失,需要99.99%的可用性(年宕机不超过52分钟)
解决方案:
1. 多机房部署(异地多活)
用户请求
│
▼
[智能DNS] ← 根据地理位置路由
/ \
/ \
华东机房 华北机房
(主) (备)
│ │
[数据库] [数据库]
(主) ←→ (从,实时同步)
2. 服务降级策略
class CircuitBreaker:
"""熔断器模式"""
def __init__(self, failure_threshold=5, timeout=60):
self.failure_count = 0
self.failure_threshold = failure_threshold
self.timeout = timeout
self.last_failure_time = None
self.state = 'CLOSED' # CLOSED, OPEN, HALF_OPEN
def call(self, func, *args, **kwargs):
if self.state == 'OPEN':
# 检查是否可以尝试恢复
if time.time() - self.last_failure_time > self.timeout:
self.state = 'HALF_OPEN'
else:
raise ServiceUnavailable("服务熔断中")
try:
result = func(*args, **kwargs)
# 成功,重置计数器
if self.state == 'HALF_OPEN':
self.state = 'CLOSED'
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = 'OPEN'
raise e
# 使用示例
circuit_breaker = CircuitBreaker()
def call_third_party_payment(order_id, amount):
return circuit_breaker.call(
third_party_api.pay,
order_id=order_id,
amount=amount
)
3. 数据库高可用
主库(Master)
│
├── 同步复制到从库1(Slave1, 华东)
└── 异步复制到从库2(Slave2, 华北)
故障切换:
1. 检测主库心跳失败(3次)
2. 提升从库1为新主库
3. DNS切换(30秒内)
4. 旧主库恢复后作为从库
4. 限流保护
from redis import Redis
class RateLimiter:
"""滑动窗口限流"""
def __init__(self, redis_client):
self.redis = redis_client
def is_allowed(self, user_id, max_requests=100, window=60):
"""
限流检查: 用户在window秒内最多max_requests次请求
"""
key = f"ratelimit:{user_id}"
now = time.time()
# 清理过期记录
self.redis.zremrangebyscore(key, 0, now - window)
# 统计当前窗口内请求数
count = self.redis.zcard(key)
if count >= max_requests:
return False
# 记录本次请求
self.redis.zadd(key, {now: now})
self.redis.expire(key, window)
return True
5. 异步化非关键路径
# 支付成功后的非阻塞操作
def on_payment_success(order_id):
# 同步返回给用户(关键路径)
response = {"status": "SUCCESS", "order_id": order_id}
# 异步执行(非关键路径,不阻塞响应)
celery.send_task('send_receipt_email', args=[order_id])
celery.send_task('update_user_credit_score', args=[order_id])
celery.send_task('sync_to_data_warehouse', args=[order_id])
return response
子问题7: 如何设计支付通知机制?
问题描述:
第三方支付(如支付宝)通过异步回调通知我们支付结果,如何确保通知不丢失且不被伪造?
解决方案:
1. 回调接口设计
@app.route('/payment/callback/alipay', methods=['POST'])
def alipay_callback():
"""
支付宝异步回调接口
"""
# 1. 获取回调参数
params = request.form.to_dict()
# 2. 验证签名(防伪造)
sign = params.pop('sign')
if not verify_alipay_signature(params, sign):
logger.warning(f"签名验证失败: {params}")
return "fail" # 返回fail,支付宝会重试
# 3. 提取关键信息
out_trade_no = params['out_trade_no'] # 我方订单号
trade_status = params['trade_status']
total_amount = Decimal(params['total_amount'])
# 4. 幂等性检查(防止重复处理)
if is_callback_processed(out_trade_no, params['trade_no']):
return "success" # 已处理过,直接返回成功
# 5. 根据状态更新订单
if trade_status == 'TRADE_SUCCESS':
update_order_status(out_trade_no, 'SUCCESS', params)
elif trade_status == 'TRADE_CLOSED':
update_order_status(out_trade_no, 'FAILED', params)
# 6. 记录回调日志
log_callback(params)
# 7. 返回success,告诉支付宝不要再重试
return "success"
def verify_alipay_signature(params, sign):
"""
验证支付宝签名
"""
# 1. 参数按key排序
sorted_params = sorted(params.items())
# 2. 拼接成字符串: key1=value1&key2=value2
unsigned_string = '&'.join([f"{k}={v}" for k, v in sorted_params])
# 3. 用支付宝公钥验证签名
from Crypto.PublicKey import RSA
from Crypto.Signature import PKCS1_v1_5
from Crypto.Hash import SHA256
import base64
key = RSA.import_key(ALIPAY_PUBLIC_KEY)
h = SHA256.new(unsigned_string.encode('utf-8'))
verifier = PKCS1_v1_5.new(key)
return verifier.verify(h, base64.b64decode(sign))
2. 主动查询兜底
# 定时任务:每5分钟检查PENDING超过10分钟的订单
@celery.task
def check_pending_orders():
pending_orders = db.query(
"SELECT order_id, third_party_txn_id "
"FROM payment_order "
"WHERE status = 'PENDING' "
"AND created_at < NOW() - INTERVAL 10 MINUTE"
).fetchall()
for order in pending_orders:
# 主动查询第三方支付结果
result = third_party_api.query_order(
order['third_party_txn_id']
)
if result['status'] == 'SUCCESS':
# 补单
update_order_status(order['order_id'], 'SUCCESS', result)
elif result['status'] == 'FAILED':
update_order_status(order['order_id'], 'FAILED', result)
3. 通知重试策略
第三方支付的重试策略(以支付宝为例):
- 第1次: 立即
- 第2次: 2分钟后
- 第3次: 10分钟后
- 第4次: 1小时后
- 第5次: 2小时后
...
- 第N次: 24小时后停止
我们的应对:
1. 每次回调都记录日志
2. 幂等性保证多次回调不会重复处理
3. 如果24小时内未收到回调,主动查询
⚖️ 五、技术选型与 Trade-off 讨论
1. 数据库选型: SQL vs NoSQL
| 维度 | SQL (MySQL/PostgreSQL) | NoSQL (MongoDB/Cassandra) |
|---|---|---|
| 优点 | • ACID 事务保证 • 强一致性 • 成熟的金融场景经验 • 复杂查询(JOIN)支持好 | • 高吞吐写入 • 水平扩展容易 • Schema 灵活 |
| 缺点 | • 垂直扩展有瓶颈 • 分库分表复杂 | • 事务支持弱 • 一致性难保证 • 金融场景不推荐 |
| 适用场景 | 核心账户、订单表 | 日志、用户行为分析 |
| 本题推荐 | ✅ MySQL (InnoDB) | ❌ 不适合核心支付逻辑 |
| 理由 | 支付系统对一致性要求极高,钱的问题容不得半点差错,必须使用支持ACID的关系型数据库。可以通过分库分表(Sharding)解决扩展性问题。 |
2. 一致性模型: 强一致性 vs 最终一致性
| 维度 | 强一致性 (Strong Consistency) | 最终一致性 (Eventual Consistency) |
|---|---|---|
| 优点 | • 数据绝对准确 • 业务逻辑简单 | • 高可用性 • 高性能 • 易扩展 |
| 缺点 | • 性能开销大 • 可用性受影响(CAP定理) | • 短期数据不一致 • 需要补偿机制 |
| 实现方式 | • 分布式事务(2PC/3PC) • Paxos/Raft | • 消息队列 • Saga模式 • 定时对账 |
| 本题推荐 | ✅ 核心账户扣款用强一致性 | ✅ 订单创建、通知等用最终一致性 |
| 混合策略 | 账户余额变动(扣款/退款) 必须强一致,用数据库事务 + 行锁保证。 周边系统(库存、积分、优惠券) 可以接受最终一致,通过消息队列异步同步,失败后人工介入或自动重试。 |
3. 同步 vs 异步支付
| 维度 | 同步支付 (Synchronous) | 异步支付 (Asynchronous) |
|---|---|---|
| 流程 | 用户点击支付 → 调用第三方 → 等待返回 → 展示结果 | 用户点击支付 → 返回"处理中" → 第三方回调 → 推送结果 |
| 优点 | • 用户体验直观 • 逻辑简单 | • 不阻塞用户 • 支持长时间处理(银行转账) |
| 缺点 | • 超时问题(第三方慢) • 用户等待焦虑 | • 需要轮询或推送 • 状态同步复杂 |
| 适用场景 | 扫码支付(微信/支付宝) | 银行转账、跨境支付 |
| 本题推荐 | ✅ 两种都支持 | 快捷支付用同步,大额转账用异步 |
4. 推模式 vs 拉模式 (订单状态查询)
| 维度 | 推模式 (Push) | 拉模式 (Pull) |
|---|---|---|
| 描述 | 第三方主动回调我们的接口 | 我们定时查询第三方接口 |
| 优点 | • 实时性好 • 不消耗查询额度 | • 主动可控 • 不依赖回调可达性 |
| 缺点 | • 回调可能丢失(网络) • 需要公网暴露接口 | • 有延迟 • 增加第三方压力 |
| 本题推荐 | ✅ 推+拉结合 | 主要依赖推(回调),拉作为兜底(定时查询PENDING订单) |
5. 单体 vs 微服务
| 维度 | 单体架构 (Monolithic) | 微服务架构 (Microservices) |
|---|---|---|
| 优点 | • 部署简单 • 事务容易保证 • 调试方便 | • 独立扩展 • 技术栈灵活 • 故障隔离 |
| 缺点 | • 扩展受限 • 单点故障 • 代码耦合 | • 分布式事务难 • 运维复杂 • 网络开销 |
| 本题推荐 | ❌ 大型支付系统不推荐 | ✅ 微服务 |
| 拆分建议 | • 支付网关服务: 接收请求,路由到具体支付渠道 • 账户服务: 管理用户余额,强一致性 • 订单服务: 订单生命周期管理 • 对账服务: 每日对账任务 • 风控服务: 实时风险检测 • 通知服务: 异步消息推送 |
6. 缓存策略: Cache-Aside vs Write-Through
| 维度 | Cache-Aside (旁路缓存) | Write-Through (写穿) |
|---|---|---|
| 读流程 | 先查缓存,miss则查DB并回写缓存 | 同左 |
| 写流程 | 先写DB,再删除缓存 | 先写缓存,缓存再写DB |
| 优点 | • 数据以DB为准,一致性好 | • 写操作快 |
| 缺点 | • 缓存miss时延迟高 | • 数据可能丢失(缓存挂了) |
| 本题推荐 | ✅ Cache-Aside | ❌ 支付场景不能接受数据丢失 |
| 缓存内容 | • 用户余额(TTL 10秒,读多写少) • 订单状态(TTL 60秒) • 不缓存: 正在处理的支付请求(一致性要求高) |
❓ 六、常见面试追问点
1. 如何防止用户重复点击导致重复扣款?
回答思路:
- 前端: 按钮点击后立即disable,防止用户连点
- 后端幂等性: 客户端生成
idempotency_key,服务端用Redis缓存处理结果。相同key的请求直接返回缓存结果 - 数据库唯一约束: 订单表的
(user_id, order_id)建唯一索引,防止同一订单被创建多次 - 分布式锁: 对同一用户的支付请求加锁(Redis SETNX),同一时刻只能处理一个
2. 支付过程中服务器宕机怎么办?
回答思路:
这就是为什么需要状态机 + 异步补偿:
- 订单状态设计: PENDING → PROCESSING → SUCCESS/FAILED,每个状态都持久化到数据库
- 定时任务兜底: 每5分钟扫描 PENDING 状态超过10分钟的订单,主动查询第三方支付结果
- 幂等性保证: 即使重复查询或回调,也不会重复扣款
- WAL日志: 关键操作先写日志,崩溃恢复时replay
3. 如何保证账户余额的一致性(多个请求同时扣款)?
回答思路:
这是经典的并发写问题,三种方案:
方案A: 悲观锁(推荐金融场景)
-- 查询时加行锁
SELECT balance FROM account WHERE user_id = 123 FOR UPDATE;
-- 扣款
UPDATE account SET balance = balance - 100 WHERE user_id = 123;
COMMIT;
方案B: 乐观锁(高并发场景)
-- 查询时记录version
SELECT balance, version FROM account WHERE user_id = 123;
-- 更新时检查version
UPDATE account SET balance = balance - 100, version = version + 1
WHERE user_id = 123 AND version = 5;
-- 如果affected_rows = 0,说明有并发冲突,重试
方案C: 分布式锁(跨服务)
redis_lock = redis.set("lock:account:123", "1", nx=True, ex=5)
if redis_lock:
# 执行扣款
deduct_balance(123, 100)
redis.delete("lock:account:123")
推荐: 数据库层面用悲观锁(FOR UPDATE),Redis分布式锁作为第二道防线。
4. 第三方支付回调丢失怎么处理?
回答思路:
多重保障机制:
- 第三方会重试: 支付宝/微信的回调有重试机制(24小时内多次重试)
- 我们主动查询: 定时任务每5分钟扫描 PENDING 订单,主动调用第三方查询接口
- 用户主动刷新: 前端轮询订单状态接口,触发后端查询
- 人工介入: 24小时后仍未收到回调,生成工单人工处理
代码示例(主动查询):
@celery.beat.schedule(crontab(minute='*/5')) # 每5分钟
def check_timeout_orders():
orders = db.query(
"SELECT * FROM payment_order "
"WHERE status = 'PENDING' AND created_at < NOW() - INTERVAL 10 MINUTE"
)
for order in orders:
result = alipay.query(order['third_party_txn_id'])
if result['status'] == 'SUCCESS':
update_order_status(order['id'], 'SUCCESS')
5. 如何设计对账系统?
回答思路:
对账三方数据对比:
- 我方订单表: payment_order
- 我方流水表: transaction_log
- 第三方账单: 每日凌晨下载支付宝/微信的对账文件
对账流程:
每日凌晨2点
↓
下载第三方对账文件(CSV)
↓
逐条对比: 订单号、金额、状态
↓
发现差异 → 记录到 diff 表
↓
差异类型:
- 长款(第三方有,我们无) → 补单
- 短款(我们有,第三方无) → 退款
- 金额不一致 → 人工审核
关键点:
- 对账单独服务,不影响主流程
- 差异自动处理 + 人工审核双保险
- 保留所有对账记录用于审计
6. 如何防止恶意刷单?
回答思路(风控体系):
实时规则引擎:
# 规则1: 频率限制(1分钟3笔)
if count_recent_txn(user_id, 60) > 3:
return REJECT
# 规则2: 金额异常(突然大额支付)
if amount > user_avg_amount * 10:
return MANUAL_REVIEW
# 规则3: 设备指纹(同一设备多账户)
if count_accounts_on_device(device_id) > 5:
return REJECT
# 规则4: IP黑名单
if is_blacklisted_ip(ip):
return REJECT
离线机器学习模型:
- 特征: 交易金额、时间、地理位置、设备、历史行为
- 模型: XGBoost/LSTM预测欺诈概率
- 高风险交易 → 人工审核
7. 支付系统的QPS瓶颈在哪里?如何优化?
回答思路:
瓶颈分析:
- 数据库写入: 500 TPS已接近单机MySQL极限
- 第三方接口: 调用支付宝/微信的网络延迟(100-300ms)
- 分布式锁: Redis单点故障会导致所有支付失败
优化方案:
| 瓶颈点 | 优化方案 |
|---|---|
| 数据库写入 | • 分库分表(按user_id hash) • 批量写入流水日志 • 读写分离(从库查询) |
| 第三方调用 | • 异步化(消息队列) • 超时控制(3秒) • 熔断降级 |
| 分布式锁 | • Redis集群(哨兵模式) • Redlock算法 • 乐观锁替代悲观锁 |
最终目标: 支持 5000+ TPS
8. 如何处理跨境支付的汇率问题?
回答思路:
汇率获取:
- 实时汇率API(如 Fixer.io, XE.com)
- 缓存汇率(每分钟更新一次,TTL 60秒)
- 汇率表记录历史汇率(用于对账)
计算时机:
# 订单创建时锁定汇率
order = {
"amount_usd": 100,
"exchange_rate": 7.2, # 创建时的汇率
"amount_cny": 720,
"currency": "CNY"
}
# 用户支付时,按锁定汇率计算
# 避免汇率波动导致用户多付或少付
汇率风险:
- 平台承担小幅波动(±0.5%)
- 超过阈值要求用户重新下单
9. 数据库分库分表如何设计?
回答思路:
分片策略(Sharding Key选择):
- 按 user_id 分片: 查询单个用户的订单很快,但某些大V会数据倾斜
- 按 order_id 分片: 数据均匀,但查询用户所有订单需要扫描所有分片
- 按时间分片: 适合归档,但热数据查询慢
推荐方案: user_id + 一致性哈希
# 假设8个分片
shard_id = hash(user_id) % 8
# 查询: 单个用户订单
SELECT * FROM payment_order_shard_{shard_id} WHERE user_id = 123;
# 查询: 所有订单(需要聚合)
results = []
for shard in range(8):
results += query(f"payment_order_shard_{shard}", "SELECT ...")
分库分表中间件: ShardingSphere, Vitess
10. 如果让你从零设计一个支付系统,你会分几期迭代?
回答思路(产品思维):
第一期(MVP - 2个月)
- 核心功能: 余额支付,单一支付方式(如支付宝)
- 订单、账户、流水三张表
- 同步支付,无退款
第二期(完善 - 1个月)
- 新增: 退款、多支付方式(微信、银行卡)
- 异步回调处理
- 基础对账
第三期(稳定性 - 1个月)
- 幂等性保证
- 分布式事务(Saga)
- 主动查询兜底
- 风控规则(频率限制)
第四期(规模化 - 持续)
- 分库分表
- 多机房部署
- 机器学习风控
- 性能优化(缓存、异步)
原则: 先保证功能可用,再追求高可用和高性能。钱的问题,宁可慢也不能错。
🗺️ 七、学习路线图
延伸阅读方向
1. 分布式事务深入
- 论文: Google Spanner (TrueTime API)
- 论文: Amazon DynamoDB (最终一致性)
- 开源项目: Seata (阿里分布式事务框架)
- 博客类型: “分布式事务的七种解决方案”
2. 金融系统架构
- 书籍: 《支付系统设计与实现》
- 开源项目: Killbill (订阅计费系统)
- 案例: Stripe API 设计哲学
- 论文: “Double-Entry Bookkeeping in Financial Systems”(复式记账)
3. 高并发系统优化
- 博客: “淘宝双11支付系统技术演进”
- 论文: Google Bigtable (海量数据存储)
- 工具: JMeter/Gatling 压测实战
- 课程: MIT 6.824 分布式系统
知识复用矩阵
这道题的核心知识可以复用到以下系统设计题:
| 系统设计题 | 复用知识点 |
|---|---|
| 设计秒杀系统 | • 高并发写入 • 库存扣减(类似余额扣减) • 幂等性设计 • 分布式锁 |
| 设计电商订单系统 | • 分布式事务(订单+支付+库存) • 状态机设计 • 对账机制 |
| 设计钱包系统 | • 账户余额管理 • 流水日志 • 充值/提现(类似支付/退款) |
| 设计积分系统 | • 账户变动记录 • 幂等性 • 对账(积分发放) |
| 设计转账系统 | • 双向流水 • 强一致性 • 风控检测 |
| 设计优惠券系统 | • 库存扣减 • 分布式锁 • 高并发领取 |
更多推荐
8
0- 0
已为社区贡献3条内容
所有评论(0)