最近接了个历史项目的维护任务,第一眼看到的是一段接近两百行的判断逻辑:订单状态、用户类型、支付渠道、优惠券、会员等级、用户标签全塞在一个方法里,改一个条件恨不得牵动全局。我盘算了一下,与其继续在这个 if-else 迷宫里打补丁,不如自己写一个可配置、可测试、可随时热更新的小引擎,代号就叫 rea(Rules Engine Architecture)。这篇内容不是劝所有人都去自研规则引擎,而是把我自己造轮子的完整思路、核心设计和踩坑记录分享出来。如果你现在正被一堆复杂分支折磨,或者正在纠结要不要在项目里引入规则引擎,这篇文章会比较有参考价值。
1. if-else 失控:我为什么非要自己写规则引擎
1.1 业务代码里的条件地狱长什么样
先说那个让我动手的场景。一个订单状态流转方法里,贴着十几个业务规则:什么“充值订单且金额大于 100 才走赠送逻辑”“企业用户且来源是渠道 A 才允许改价”“会员等级不低于 2 且不是黑名单用户才能参与活动”……每条规则两三个条件,组合起来就是一张密密麻麻的判定网。
这种代码的问题,不是单个 if 有多难懂,而是组合之后根本没法读。你以为自己在加一个“用户不是内部账号”的条件,实际却要理解整条链路的上下文。更要命的是,这些规则经常变,产品侧一个月调三五次门槛、人群、渠道范围,改一次就发布一次。发版倒不累,累的是回归测试:你永远不知道某一层条件的改动,会不会在另一个分支里触发意想不到的结果。
我总结了一下,条件地狱通常有三个典型症状:
- 改一点,动全身。条件之间有隐含依赖,一个判断挪了位置,后面的所有分支都可能受影响。
- 测试覆盖率上不去。组合条件一多,用例数量指数级膨胀,根本写不完。
- 决策链路不可见。线上出了问题,你很难回答“这个用户为什么命中了这条规则”,因为逻辑全在代码里,没有地方可以查。
这就是我要写 rea 的起点:把“条件判断”这件事从业务代码里抽出来,变成一份可读、可查、可改的规则配置。
1.2 为什么不用现成的规则引擎
决定动手之前,我当然先盘了一圈现成方案。市面上确实有功能非常完整的规则引擎,但放进我这个项目的上下文,反而处处别扭。我当时的判断标准很简单:团队能快速上手、部署不增加额外依赖、规则能配置化,就够了。
我整理过一个对比表,贴在这里供参考:
| 方案 | 优点 | 在我这里的障碍 |
|---|---|---|
| 重量级规则引擎 | 功能全面,有复杂事件处理能力 | 概念多、学习曲线陡,几百行的项目没人愿意为了它看几天文档 |
| 云上规则平台 | 托管、界面配置方便 | 业务数据要出内网,安全和合规过不了 |
| 基于表达式组件 | 灵活、性能好 | 底层依赖较重,表达式语法需要团队重新学习 |
| 自研轻量规则引擎 | 代码量小、完全可控、可定制 | 需要自己维护边界和测试 |
对比下来,我选择了自研。因为我的核心诉求非常朴素:能用 JSON 描述条件,能按优先级匹配,能返回规则命中的结果。这些东西用成熟引擎当然能做,但对于一个核心逻辑只有几百行的模块,重量级框架带来的更多是负担而不是收益。
提示:如果你所在的团队已经有成熟的规则平台或表达式组件,并且大家都用熟了,没必要为了“整洁”再造一个轮子。我自研的前提是,项目里确实没有合适的基础设施,且规则复杂度可控。
2. rea 的第一层设计:规则长什么样,数据怎么进
2.1 先把规则拆成“条件 + 行动”
规则引擎的第一步,不是写代码,而是定义规则的数据结构。我反复调过几版,最后稳定下来的规则模型非常简单,就两个部分组成:条件(condition)和行动(action)。条件告诉你“什么时候生效”,行动告诉你“生效之后做什么”。
一个典型的规则大概是这样的:
{ "rule_id": "R0001", "name": "大额订单赠送优惠券", "priority": 10, "condition": { "all": [ { "field": "order.amount", "op": "gt", "value": 100 }, { "field": "user.level", "op": "gte", "value": 2 }, { "field": "order.channel", "op": "in", "value": ["app", "h5"] } ] }, "action": { "type": "grant_coupon", "params": { "coupon_id": "C_200", "amount": 20 } } }我用all表示“同时满足”,对应的还有any表示“任意满足”。这个设计是从日常口语里来的:业务方说“满足 A 和 B,或者只满足 C 也行”,翻译过来就是any: [{all: [A, B]}, C]。嵌套关系在 JSON 里很自然,不需要发明新的语法。
2.2 因子抽取:让规则和业务字段解耦
有一件事如果不在第一版设计里想清楚,后面一定会返工:规则里引用的字段,不能直接绑死业务对象。
比如订单规则里写order.amount,如果判断代码直接去拿Order对象的amount字段,那规则就跟你这个实体类绑死了。以后订单对象重构,或者换个项目想复用这套引擎,规则全部作废。
rea 的做法是引入一个“因子层”。业务侧在调用引擎之前,先把关键数据整理成一个扁平的、用点号做层级分隔的字典,我管它叫factors:
factors = { "order.amount": 168, "order.channel": "app", "user.level": 3, "user.tags": ["vip", "new_user"], "risk.blacklist": False }规则里的field就是这个字典的键。这样做有三个好处:
- 业务对象随便重构,只要 factors 的键不变,规则就不用动。
- 规则可以跨项目复用,因为依赖的是一份中立的数据结构,而不是某个具体类。
- 排查起来直白,规则命不中,看一眼 factors 就清楚是哪个字段没达到条件。
注意:因子抽取这个环节看起来简单,但一定要在项目初期定好规范。我见过最难受的写法是有的地方传
amount,有的地方传order_amount,同一个含义两种键,规则写出来全是坑。
3. rea 的第二层设计:匹配到底怎么跑
3.1 条件表达式求值:从 JSON 到布尔结果
规则数据是死的人是活的,再好看的 JSON 也要翻译成程序能执行的逻辑。rea 的条件表达式只支持一组限定操作符,因为实际操作中你会发现,绝大多数规则需要的判断并不多:
| 操作符 | 含义 | 示例 |
|---|---|---|
eq/neq | 等于 / 不等于 | user.level eq 3 |
gt/gte | 大于 / 大于等于 | order.amount gte 100 |
lt/lte | 小于 / 小于等于 | inventory.stock lt 5 |
in/not_in | 在列表中 / 不在列表中 | order.channel in ["app", "h5"] |
contains/not_contains | 包含 / 不包含(用于列表或字符串) | user.tags contains "vip" |
is_null/is_not_null | 是否为空 | coupon.used_at is_null |
操作符的实现是用一张操作符映射表,把字符串映射到对应的判断函数。这比写一堆if op == "gt": ...干净得多,而且以后想加一个startswith之类的操作符,只需要往表里注册一个函数就行。
单个条件求值的过程很简单:从 factors 里取出field对应的值,调用映射表里的函数,传值和配置里的value比较,得到一个布尔结果。复杂的是条件组合的求值顺序,这里要处理短路逻辑。
3.2 批量匹配与首次命中
规则不可能一条条手动调,引擎要提供一个“喂一把数据,吐出所有命中”的能力。rea 在做匹配的时候,会按优先级把规则排好序,然后根据调用方的意图分两种模式:
- 首个命中模式:适合“只走一个分支”的场景,比如流量分流。优先级最高的规则一旦命中,后面的规则不再判断。
- 全部命中模式:适合“所有结果都要”的场景,比如风控标记。每条命中的规则都会留下结果,方便后续聚合。
这个设计看起来简单,但我在早期版本里漏了“首个命中”这个模式,结果用在分流场景时,配置里还必须手动给规则之间加互斥条件,特别别扭。后来补上模式参数之后,规则配置才真正清爽起来。
匹配过程中的一个重要细节是条件组的求值顺序。对于all组,一旦发现有一个子条件失败,整个组就可以直接返回 False,不用继续往下求值。这样做不仅仅是为了性能,更是为了避免拿到不存在的字段时抛异常——如果前面的条件已经让你的规则没机会命中了,后面再报错反而是噪音。
4. 核心实现:两百行 Python 怎么撑起一个可用引擎
4.1 规则容器与匹配器的骨架
理论说得再多,不如直接看代码。rea 的第一版我用纯 Python 实现,完整功能不到三百行。核心部分分三个模块:操作符表、条件求值器、引擎主类。我先贴一个精简版,去掉日志和边界防护,保留主链路。
# rea_core.py from typing import Any, Dict, List, Optional # ---------- 操作符映射表 ---------- def _op_eq(a, b): return a == b def _op_neq(a, b): return a != b def _op_gt(a, b): return a > b def _op_gte(a, b): return a >= b def _op_lt(a, b): return a < b def _op_lte(a, b): return a <= b def _op_in(a, b): return a in b def _op_not_in(a, b): return a not in b def _op_contains(a, b): return b in a def _op_not_contains(a, b): return b not in a def _op_is_null(a, b): return a is None def _op_is_not_null(a, b): return a is not None OPERATORS = { "eq": _op_eq, "neq": _op_neq, "gt": _op_gt, "gte": _op_gte, "lt": _op_lt, "lte": _op_lte, "in": _op_in, "not_in": _op_not_in, "contains": _op_contains, "not_contains": _op_not_contains, "is_null": _op_is_null, "is_not_null": _op_is_not_null, }操作符表是整套引擎最底层的地基。把操作符独立成函数而不是内联在条件求值里,最大的好处是单测容易写,而且后续想扩展“自定义操作符”时,只需要往字典里加键值。
4.2 条件求值与规则匹配
def evaluate_condition(cond: Dict, factors: Dict) -> bool: if "all" in cond: return all(evaluate_condition(sub, factors) for sub in cond["all"]) if "any" in cond: return any(evaluate_condition(sub, factors) for sub in cond["any"]) op = OPERATORS.get(cond["op"]) if op is None: raise ValueError(f"unsupported operator: {cond['op']}") value = factors.get(cond["field"]) return op(value, cond.get("value"))这段逻辑关键在递归:all和any可以无限嵌套,深层结构也能正确求值。用generator配合all/any天然实现短路,条件是all时遇到第一个 False 就不再往后走,条件是any时遇到第一个 True 就结束。而且每个子条件递归进去,自身又可能是一个嵌套组,表达能力非常够用。
引擎主体负责排序、过滤和收集结果:
class RuleEngine: def __init__(self, rules: Optional[List[Dict]] = None): self._rules = rules or [] def load_rules(self, rules: List[Dict]) -> None: self._rules = sorted(rules, key=lambda r: r.get("priority", 0), reverse=True) def match(self, factors: Dict, first_only: bool = False) -> List[Dict]: hits = [] for rule in self._rules: if evaluate_condition(rule["condition"], factors): hits.append(rule) if first_only: break return hitsload_rules按 priority 从大到小排序,高优先级在前。match遍历规则,把命中的规则对象收集起来。first_only控制是否命中第一条就停。整个引擎对外暴露的接口就这么两个方法,简单直接,团队成员看一眼就会用。
4.3 对外调用示例
接进业务侧的时候,调用很干净:
rules = load_from_config("order_rules.json") engine = RuleEngine(rules) factors = { "order.amount": 168, "order.channel": "app", "user.level": 3, "user.tags": ["vip", "new_user"], } hit_rules = engine.match(factors, first_only=True) if hit_rules: action = hit_rules[0]["action"] do_action(action, factors)从这里能看出整个设计的一个关键选择:业务侧只负责组装 factors 和执行 action,判断逻辑完全交给 engine。也就是说,以后规则调整,只需要改配置文件或者数据库里的规则数据,服务代码一行都不用动。
5. 真实场景接入:优惠、分流、风控三处落地的差异
5.1 优惠场景:多条件叠加和金额计算
第一个接入 rea 的场景是优惠逻辑。原代码里判断“能不能用优惠券”的逻辑分散在三个方法里,因为有“订单金额门槛”“用户身份限制”“渠道限制”三层条件,后来还要叠加“黑名单用户不参与”。用 rea 之后,规则配置变得非常直观:
{ "rule_id": "R1001", "name": "普通用户满减券门槛", "priority": 5, "condition": { "all": [ { "field": "order.amount", "op": "gte", "value": 99 }, { "field": "user.tags", "op": "not_contains", "value": "internal" }, { "field": "risk.blacklist", "op": "eq", "value": false } ] }, "action": { "type": "apply_coupon", "params": { "coupon_id": "C_99_10", "discount": 10 } } }优惠场景最大的特点是条件经常调。满减门槛从 99 调到 129,新用户标签从new_user改成first_order,这些在原来代码里都需要发版,现在直接改配置就行。而且因为规则是数据,产品同事甚至可以自己在后台维护,研发只负责保证引擎稳定。
5.2 分流场景:优先级和互斥控制
第二个场景是灰度分流。以前做 A/B 测试,要么在网关层配置切流比例,要么在代码里写if user_id % 100 < 20这种埋点。问题是这种埋点逻辑分散在各个服务里,统计口径经常对不上。
rea 在这个场景用first_only模式,给每批流量配一条规则,高优先级先拦截:
[ { "rule_id": "F_10", "name": "灰度包用户", "priority": 100, "condition": { "all": [ { "field": "user.tags", "op": "contains", "value": "gray_pkg" } ] }, "action": { "type": "route", "params": { "group": "canary" } } }, { "rule_id": "F_20", "name": "按尾号分流", "priority": 50, "condition": { "all": [ { "field": "user.id_suffix", "op": "in", "value": ["1", "3", "5", "7", "9"] } ] }, "action": { "type": "route", "params": { "group": "experiment" } } }, { "rule_id": "F_30", "name": "默认组", "priority": 1, "condition": { "all": [] }, "action": { "type": "route", "params": { "group": "control" } } } ]注意最后一条默认组,all: []永远为真,相当于 else 分支。有它在,前面的规则即使全都配错,流量至少不会穿透到空逻辑里。这个写法是我踩过坑之后才加上的:任何用 first_only 模式接入的地方,都应该在最低优先级放一条兜底规则。
5.3 风控场景:所有命中都要拿
第三个场景是风控标记,和前面两个有本质区别。优惠和分流大多只关心“走哪个分支”,风控场景却要把所有命中结果都收集下来,因为一个用户可能同时触发“设备异常”“频次超限”“地址变更”多条规则,每条规则都要产生一个独立的标记。
rea 在这个场景切换成全量命中模式,返回值是一个命中规则数组。业务侧拿到结果之后,会根据规则里的风险等级字段做聚合,比如只要有两条以上level: high的规则命中,就进入人工审核队列。这个模式之所以要单独设计,是因为如果引擎只支持“首个命中”,风控场景的规则配置就得人为拆分,反而破坏了一组相关规则的完整性。
6. 回归与避坑:我在 rea 上栽过的坑和补救办法
6.1 变量缺失时的静默失败
第一个坑是字段缺失。早期版本的求值器直接写factors[cond["field"]],字段不存在就抛 KeyError。结果测试环境什么问题都没有,上了生产就开始报警——因为有几个上游服务返回的数据结构不是每次都带全所有字段,一旦缺少某个键,整条规则求值直接中断。
后来我把取值改成factors.get(cond["field"]),缺失时返回 None。这样操作符表里的eq、is_null这些能正常处理 None 值,但gt这类数值比较就会抛 TypeError。最终方案是加了一个异常捕获,字段缺失时记一条 warning 并让这条规则返回 False:
def evaluate_condition(cond: Dict, factors: Dict) -> bool: if "all" in cond: return all(...) if "any" in cond: return any(...) op = OPERATORS.get(cond["op"]) value = factors.get(cond["field"]) # 关键:缺失时名字存在于 factors 吗?如果字段不存在,这里返回 None # 缺失逻辑:如果字段压根不在 factors 里,说明数据不全,按不命中处理 if cond["field"] not in factors: return False return op(value, cond.get("value"))这个调整的哲学是:规则引擎不应该因为数据不完整而崩溃,但也绝不假装数据完整。不命中是安全的默认行为,同时通过日志把缺失情况暴露出来,让数据链路的问题能被及时看到。
6.2 高频调用下的性能优化
第二个坑是关于性能。第一版实现每次调用match的时候都实时递归求值,压测发现单次 1000 条规则的匹配耗时还能接受,但有些接口单笔调用会反复匹配多次,再加上条件里还挂了contains操作符去扫几百个元素的标签列表,QPS 一上来 CPU 就顶不住了。
我做了两个优化。第一个是规则编译缓存:把规则 JSON 在加载时就编译成内部的表达式对象,而不是每次匹配再重新解析 JSON。第二个是条件预筛:在匹配前先用一两个代价极低的字段做粗过滤,比如大多数规则都要求enable: true,这个条件单独提出来先比较,不符合的规则连条件组都不进入。做完这两个优化,单次匹配耗时降到了原先的十分之一左右。
6.3 规则越来越多,怎么办边界
第三个坑是规则数量膨胀。本来 rea 适合三五十条规则的场景,但有一段时间配置中心里的规则越来越多,某条规则改动后,另一条规则突然不生效了——两个条件的优先级没有梳理清楚,高优先级规则把低优先级规则“吃”掉了。
后来我定了两条硬性规范:
- 任何规则必须写明负责人和上线日期,配置数据里加
owner和created_at字段。 - 每次加载规则时做冲突检测,自动检查有没有两条高优先级规则的条件组存在包含关系,有的话在日志里提醒。
规则引擎本身不复杂,复杂的是规则和规则之间的关系。如果没有人治理,配置化最终会变成另一堆更隐蔽的条件耦合。
6.4 别把所有逻辑都塞进规则
最后想分享一个心态上的坑。写规则引擎容易上瘾,因为配置规则太方便了,会有一种“什么都能用规则解决”的错觉。但实际上,像金额分摊计算、风控模型推理这种复杂的算法逻辑,根本不适合放到配置化的规则里。规则引擎适合放的是决策逻辑,也就是“什么情况下走哪条路”,至于路怎么走,应该是注册在 action 里的普通代码。
我的判断标准很简单:如果一条规则的条件组超过七八个字段,或者 action 参数里塞着一大段计算公式,那就说明这个规则已经超出“配置”的范畴了,该用代码实现还是要用代码实现。规则引擎的价值是让决策可见、可改、可追踪,而不是把整个业务都搬进去。
我把 rea 的完整代码和配置示例整理好之后,团队里第一次接它的人只花了半天就上手了。后来不止一个人问我,为什么不自研一个共享平台把这套东西推广到全公司。我的回答是:规则引擎根本不是一个平台问题,而是一种写代码的思维方式。只要条件判断仍然内嵌在业务逻辑里,换多少框架都一样。真正值钱的是那一层把“决策”从“流程”里剥离出来的设计,至于实现,两百行的 rea 和两万行的商业引擎,在这件事上并没有本质区别。