从路由策略到代码实现的完整指南，含流程图、核心代码和应用场景。
支付路由是支付系统的"调度中心"。一笔支付请求进来后，路由系统会根据成功率、成本、渠道稳定性、商户配置、用户属性等维度，决定把这笔交易交给哪个支付渠道（微信、支付宝、银联、外卡等）处理。
简单说：支付路由 = 在多个支付渠道中选择最合适的那一个。
没有路由时，系统通常只能固定走一个渠道。一旦这个渠道：
整个支付体验就会直接崩盘。
有了路由系统后，可以实现：
设计路由策略时，通常会考虑以下几个维度：
| 维度 | 说明 | 举例 |
|---|---|---|
| 成功率 | 渠道近期支付成功率 | 微信支付 99.2%，支付宝 98.5% |
| 成本 | 渠道费率 | 银联 0.38%，支付宝 0.55% |
| 响应速度 | 渠道平均耗时 | 微信 120ms，外卡 800ms |
| 渠道状态 | 是否可用/维护 | 通道 A 正在维护 |
| 商户配置 | 商户指定渠道 | 某商户只开微信 |
| 用户属性 | 用户使用的支付方式 | 用户选择支付宝 |
| 卡种/地区 | 银行卡类型、国家地区 | 境外卡需走外卡通道 |
| 限额 | 单笔/日累计限额 | 通道 A 单笔上限 5 万 |
这些维度可以抽象成路由规则和评分权重。
symoon笔记
flowchart LR
A[商户/用户发起支付] --> B[支付路由引擎]
B --> C[规则匹配]
C --> D[渠道评分]
D --> E[选择最优渠道]
E --> F[调用支付渠道]
F --> G[渠道返回结果]
G --> H[更新渠道状态/统计]
H --> B
subgraph 数据源
I[渠道配置表]
J[成功率监控]
K[成本配置]
L[熔断状态]
end
I --> B
J --> B
K --> B
L --> B
架构核心组件：
flowchart TD
Start([支付请求]) --> CheckRules{规则匹配}
CheckRules -->|命中硬规则| Filter[过滤可用渠道]
CheckRules -->|未命中| ReturnDefault[返回默认渠道]
Filter --> Score[多维度评分]
Score --> Rank[排序]
Rank --> Select[选择Top1]
Select --> Call[调用渠道]
Call --> Result{结果}
Result -->|成功| Success[返回成功]
Result -->|失败| Retry{是否重试?}
Retry -->|是| RetryNext[选择下一个渠道]
RetryNext --> Call
Retry -->|否| Fail[返回失败]
Result -->|异常| UpdateCircuit[更新熔断状态]
UpdateCircuit --> RetryNext
Success --> End([结束])
Fail --> End
流程说明：
按固定优先级选择渠道，适合业务初期或合规要求明确的场景。
# 优先级路由示例
def priority_route(channels, order):
for ch in order:
if ch in channels and channels[ch].is_available:
return ch
return None
选择费率最低的可用渠道，适合对成本敏感的场景。
# 成本最优路由示例
def cost_optimal_route(channels):
available = [c for c in channels if c.is_available]
return min(available, key=lambda c: c.fee_rate)
选择近期成功率最高的渠道，适合对支付成功率敏感的场景。
把多个维度加权打分，得到综合评分，再排序选择。这是最常用的策略。
评分公式：
score = 成功率 × w1 + (1 - 成本占比) × w2 + 响应速度得分 × w3 + 渠道稳定性 × w4
其中权重根据业务目标调整：
下面是一个简化但可直接运行的 Python 版支付路由引擎。
from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum
class ChannelStatus(Enum):
AVAILABLE = "available"
DEGRADED = "degraded"
CIRCUIT_OPEN = "circuit_open" # 熔断
@dataclass
class PaymentChannel:
channel_id: str # 渠道ID，如 wechat、alipay
name: str # 渠道名称
fee_rate: float # 费率，如 0.0038
success_rate: float # 近5分钟成功率 0.0-1.0
avg_latency_ms: int # 平均响应耗时
status: ChannelStatus
max_amount: float # 单笔限额
priority: int = 100 # 优先级，数字越小越优先
@dataclass
class PaymentRequest:
amount: float
currency: str
user_id: str
merchant_id: str
pay_method: str # 用户选择的支付方式
card_region: str = "CN" # 卡/账户地区
def filter_channels(
channels: List[PaymentChannel],
request: PaymentRequest,
merchant_config: Dict[str, List[str]]
) -> List[PaymentChannel]:
"""
过滤掉不可用、不满足商户配置、超限额的渠道
"""
allowed = merchant_config.get(request.merchant_id, [])
result = []
for ch in channels:
# 1. 渠道必须可用
if ch.status == ChannelStatus.CIRCUIT_OPEN:
continue
# 2. 商户白名单
if allowed and ch.channel_id not in allowed:
continue
# 3. 支付方式匹配
if ch.channel_id != request.pay_method and request.pay_method != "auto":
continue
# 4. 单笔限额
if request.amount > ch.max_amount:
continue
result.append(ch)
return result
def score_channel(ch: PaymentChannel) -> float:
"""
综合评分：成功率40%、成本30%、响应速度20%、稳定性10%
"""
success_score = ch.success_rate * 100 * 0.4
# 成本越低得分越高，这里用 1 - 费率/0.01 做归一化
cost_score = max(0, (1 - ch.fee_rate / 0.01)) * 100 * 0.3
# 响应速度：100ms以内满分，超过1s得0分
latency_score = max(0, (1000 - ch.avg_latency_ms) / 1000) * 100 * 0.2
# 稳定性：degraded状态扣一半分
stability_score = 100 * 0.1 if ch.status == ChannelStatus.AVAILABLE else 50 * 0.1
return success_score + cost_score + latency_score + stability_score
class PaymentRouter:
def __init__(self, channels: List[PaymentChannel], merchant_config: Dict):
self.channels = channels
self.merchant_config = merchant_config
def route(self, request: PaymentRequest) -> Optional[PaymentChannel]:
# 1. 过滤
candidates = filter_channels(self.channels, request, self.merchant_config)
if not candidates:
return None
# 2. 评分排序
candidates.sort(key=score_channel, reverse=True)
# 3. 返回Top1
return candidates[0]
def route_with_fallback(self, request: PaymentRequest) -> List[PaymentChannel]:
"""
返回一个有序列表，用于主渠道失败时重试
"""
candidates = filter_channels(self.channels, request, self.merchant_config)
candidates.sort(key=score_channel, reverse=True)
return candidates
import time
from collections import deque
class CircuitBreaker:
def __init__(self, threshold=5, timeout=60):
self.threshold = threshold # 连续失败阈值
self.timeout = timeout # 熔断后恢复时间（秒）
self.failures = 0
self.last_failure_time = 0
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
def record_success(self):
self.failures = 0
self.state = "CLOSED"
def record_failure(self):
self.failures += 1
self.last_failure_time = time.time()
if self.failures >= self.threshold:
self.state = "OPEN"
def allow_request(self):
if self.state == "CLOSED":
return True
if self.state == "OPEN":
if time.time() - self.last_failure_time > self.timeout:
self.state = "HALF_OPEN"
return True
return False
return True # HALF_OPEN 允许试探
if __name__ == "__main__":
channels = [
PaymentChannel("wechat", "微信支付", 0.0038, 0.992, 120, ChannelStatus.AVAILABLE, 50000),
PaymentChannel("alipay", "支付宝", 0.0055, 0.985, 150, ChannelStatus.AVAILABLE, 50000),
PaymentChannel("unionpay", "银联", 0.0028, 0.978, 300, ChannelStatus.DEGRADED, 100000),
]
merchant_config = {
"m_1001": ["wechat", "alipay"], # 该商户只开通微信和支付宝
}
router = PaymentRouter(channels, merchant_config)
req = PaymentRequest(
amount=3000.0,
currency="CNY",
user_id="u_123",
merchant_id="m_1001",
pay_method="auto",
)
selected = router.route(req)
if selected:
print(f"路由选择: {selected.name} (score={score_channel(selected):.2f})")
else:
print("无可用渠道")
运行结果大致：
路由选择: 微信支付 (score=45.68)
双 11、618 期间流量激增，单一渠道容易被打满或限流。路由系统可以：
不同国家/地区支持的支付方式差异很大：
路由系统根据用户地区、卡组织、币种选择最合适的渠道。
聚合支付平台接入了几十个渠道，需要把每笔交易路由到最优渠道：
对于交易量大、毛利薄的业务，路由系统可以根据费率选择成本最低的渠道：
支付渠道偶尔会出现网络波动或接口异常。路由系统：
支付路由不是简单的"选一个渠道"，而是一个综合考虑成功率、成本、稳定性、商户需求、用户体验的决策系统。
一个好的支付路由系统应该具备：
从简单静态路由，到动态加权评分，再到智能机器学习路由，可以逐步演进。初期建议先用规则过滤 + 加权评分 + 熔断重试，已经能满足绝大多数业务场景。
写在最后：支付系统无小事，路由设计要在稳定、成本、体验之间找到平衡，切勿为了过度优化而牺牲可靠性。