用 Trie 实现 API 网关路径前缀路由:Python 最小实现与冲突测试
【摘要】 介绍如何用 Trie 按路径分段管理 API 网关路由,查询时逐段下降并保留最深前缀回退,覆盖 /api/users/* 等规则;给出可运行的 Python 最小实现、冲突规则说明与测试方法。
用 Trie 实现 API 网关的路径前缀路由
问题场景
API 网关常需要根据请求路径选择后端服务。规则既包含精确路径,也包含前缀回退,例如:
/api/users交给用户服务/api/users/*作为用户服务兜底/api/users/123/orders/*交给订单服务
如果每次请求都遍历全部规则并做字符串前缀比较,规则多时维护和排查都不方便。Trie 把路径按 / 分段存储,查询时逐段下降,并在下降过程中记录候选处理器。
路由语义
本文实现以下约定:
- 规则和请求路径必须以
/开头,末尾/会被忽略,连续空段会被合并。 - 只支持末尾
*,例如/api/users/*;中间通配不处理。 - 节点保存
exact与prefix两个处理器:exact只匹配完整路径;prefix匹配该节点及其后代。 - 查询优先返回完整路径命中的
exact;没有时返回下降过程中遇到的最深prefix。 - 同一路径重复插入精确规则或前缀规则会抛异常;需要替换时显式传
override=True。
Python 最小实现
from dataclasses import dataclass, field
from typing import Dict, Optional, List, Tuple
@dataclass
class Node:
children: Dict[str, 'Node'] = field(default_factory=dict)
exact: Optional[str] = None
prefix: Optional[str] = None
class RouteTrie:
def __init__(self) -> None:
self.root = Node()
@staticmethod
def _segments(path: str) -> List[str]:
if not path.startswith('/'):
raise ValueError('path must start with /')
path = path.rstrip('/') or '/'
if path == '/':
return []
return [s for s in path.split('/') if s]
def add(self, rule: str, handler: str, *, override: bool = False) -> 'RouteTrie':
segs = self._segments(rule)
wildcard = bool(segs and segs[-1] == '*')
if wildcard:
segs = segs[:-1]
node = self.root
for seg in segs:
if seg == '*':
raise ValueError('only trailing * is supported')
node = node.children.setdefault(seg, Node())
if wildcard:
if node.prefix is not None and not override:
raise ValueError(f'duplicate prefix rule: {rule}')
node.prefix = handler
else:
if node.exact is not None and not override:
raise ValueError(f'duplicate exact rule: {rule}')
node.exact = handler
return self
def match(self, path: str) -> Optional[str]:
segs = self._segments(path)
node = self.root
fallback = node.prefix
depth = 0
for seg in segs:
nxt = node.children.get(seg)
if nxt is None:
break
node = nxt
depth += 1
if node.prefix is not None:
fallback = node.prefix
if depth == len(segs) and node.exact is not None:
return node.exact
return fallback
运行与测试
把代码保存为 route_trie.py,在文件末尾加入测试并执行 python route_trie.py:
if __name__ == '__main__':
t = RouteTrie()
t.add('/api/users', 'users')
t.add('/api/users/*', 'users_fallback')
t.add('/api/users/123', 'user_123')
t.add('/api/users/123/orders/*', 'orders_fallback')
cases: List[Tuple[str, Optional[str]]] = [
('/api/users', 'users'),
('/api/users/', 'users'),
('/api/users/123', 'user_123'),
('/api/users/123/orders', 'orders_fallback'),
('/api/users/123/orders/456', 'orders_fallback'),
('/api/users/999/orders', 'users_fallback'),
('/api/other', None),
]
for path, expected in cases:
got = t.match(path)
assert got == expected, (path, got, expected)
try:
t.add('/api/users', 'duplicate')
except ValueError:
pass
else:
raise AssertionError('duplicate exact rule should fail')
print('all route tests passed')
冲突规则可以归纳为:同一节点上,完整路径请求优先使用 exact;请求继续向下时,最近的前缀规则覆盖更浅的前缀规则;更具体的字面量节点天然位于更深层,因此优先于浅层 * 回退。重复定义同一规则不会被静默覆盖,便于在配置加载阶段发现错误。
边界与限制
- 仅支持末尾
*,不支持*出现在路径中间,也不支持参数占位符、正则或 HTTP 方法区分。 - 路径按
/分段并过滤空段,因此/api//users与/api/users视为同一路径;如果业务需要保留空段,应调整_segments。 - 单次查询时间复杂度与路径段数相关,不随规则总数线性增长;内存占用取决于规则节点数量。
- 若需要按 HTTP 方法、域名或请求头路由,可在节点处理器中保存更完整的路由配置,而不是只存字符串。
- 规则热更新时应加锁或使用不可变快照,避免读写并发导致匹配到中间状态。
【版权声明】本文为华为云社区用户原创内容,未经允许不得转载,如需转载请自行联系原作者进行授权。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱:
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论(0)