📚 课程目录第3章规则引擎概念

3. 规则引擎概念：什么是规则引擎

我们先从一个最常见的场景聊起——日志巡检。

你有一堆系统日志，每天几百个G。你关心的是：“有没有登录失败超过5次的IP”、“有没有磁盘使用率超过90%的告警”、“有没有半夜3点还在疯狂刷接口的异常行为”。

传统做法是什么？写脚本，if-else 硬编码。今天改一个阈值，明天加一个规则，就得改代码、重新部署。我早期做运维时就吃过这个亏——改了3行代码，结果有个正则没测好，线上漏报了一整夜的告警。后来我才意识到：业务逻辑和代码不应该绑死，你得把「判断」这件事独立出来，交给一个专门干这个的组件——规则引擎。

说白了，规则引擎就是一个“如果条件满足，就执行动作”的通用框架。它把规则当作数据存起来，运行时加载、匹配、触发。不用动代码，改个配置文件或者界面按钮就能上线。

3.1 规则引擎的核心价值

你想想看，没有规则引擎的日子长什么样？

改阈值 → 拉代码分支 → 改 if 条件 → 走发布流程 → 等上线。
加一个新规则 → 重新理解业务 → 写新函数 → 测试 → 部署。
想看看现在系统里有哪些规则在跑 → 去代码里 grep 搜索 if/elif。

有了规则引擎，这些事情就变成了：

改阈值 → 在规则表里改一条记录，即时生效。
加新规则 → 填一个表单，写条件表达式和动作。
查规则 → 一张表，全部列出来，一目了然。

我在一个日处理量超过200亿条日志的项目里，用规则引擎把告警规则的修改时间从「小时级」降到了「分钟级」。主管当时都觉得不可思议——其实就是把硬编码的 if 挪到了数据库里。嗯，就是这么简单的一个思路。

3.2 规则引擎的核心组件

一个完整的规则引擎，哪怕再复杂，拆开来看也就三个部分：规则定义、规则匹配、动作执行。下面我们一个一个说清楚。

3.2.1 规则定义

规则长什么样？

我习惯用三个要素来描述一条规则：

要素	说明	例子（日志巡检）
条件	触发规则需要满足的逻辑表达式	`error_count > 5 AND level = 'ERROR'`
动作	条件满足时要执行什么	发送邮件 / 写入告警表 / 触发 webhook
优先级	多条规则同时满足时，先执行谁	数字越小优先级越高

在代码里，规则定义通常是一个结构体或类：

class Rule:
    def __init__(self, rule_id: str, condition_expr: str, action: str, priority: int):
        self.rule_id = rule_id
        self.condition_expr = condition_expr   # 比如 "status == 500 and count > 10"
        self.action = action                    # 比如 "send_sms"
        self.priority = priority

我的习惯：条件表达式最好设计成字符串形式，通过解析引擎动态求值。不要直接写 Python 代码存进去——那样不安全。我曾经见过一个项目，规则字段里直接存了 exec("...")，结果运维误操作把服务器文件全删了。用表达式解析器（比如 eval 加白名单，或者用第三方类库）会安全很多。

3.2.2 规则匹配

规则定义好了，那怎么知道一条日志（或者一批日志）该不该触发规则？这就是「匹配」干的事。

匹配的核心：将输入数据（日志事件）和规则条件做布尔运算。

我举个例子。假设有一条规则：

条件：response_time > 5000 AND api_path == '/api/login'
动作：发送告警邮件

现在来了一条日志：

{ "api_path": "/api/login", "response_time": 8200, "status": 200 }

匹配引擎会做这几步：

提取日志中的 response_time=8200 和 api_path='/api/login'。
计算 8200 > 5000 → true。
计算 '/api/login' == '/api/login' → true。
最终条件 true → 执行动作。

匹配引擎的实现方式有很多种：

简单 if-else 链：适合规则少且固定的场景，但扩展性差。
规则表 + 线性扫描：把规则放在列表里，逐条匹配。简单直白，规则量少时够用。
Rete 算法 / 索引匹配：当规则量上千甚至上万时，逐条扫描性能不够。Rete 算法会构建条件网络，避免重复计算。我在处理百万级规则匹配时用过一次，效果很好，但实现难度也大。

  选型建议：日志巡检场景下，规则数一般不会超过1000条。用线性扫描 + 规则分组（比如按日志级别分组）就足够了。不要为了炫技引入复杂算法，维护成本划不来。

3.2.3 动作执行

规则匹配成功了，接下来呢？干正事。

动作执行就是把匹配结果转化成「实际行动」。常见的动作类型：

动作类型	说明	使用场景
告警通知	发送邮件、短信、钉钉/企业微信消息	日志出现 ERROR、业务异常
日志归档	将匹配到的日志写入专门的告警表或 ES 索引	长期审计、回溯分析
自动修复	调用脚本或 API 进行重启、清理、降级	确认阈值后可自动处理的场景
计数累积	将匹配次数累加，达到阈值后触发更高级别的动作	防止短时间大量重复告警（降噪）

在设计动作执行层时，我有几点建议：

动作要可配置：不要写死在代码里。比如“发送短信”这个动作，应该允许用户在规则定义时选择，而不是硬编码在 if 里。
动作要幂等：同样的规则匹配到同一条日志，重复执行动作不应该产生副作用。举个例子，发送告警时最好带上“最近N分钟内只发一次”的去重逻辑。我曾经遇到过某个夜间高峰，同一个错误日志每秒匹配10次，结果短信接口被调用到被封号——惨痛教训。
动作执行失败要可重试：网络抖动、接口超时是常事。异步队列 + 重试机制是标配。

3.3 三者关系——一张图看懂

规则定义、规则匹配、动作执行，这三者不是孤立的。我把它们的协作流程画出来：

从图中能看到一条完整的数据流：

日志事件进入系统。
规则匹配引擎从规则库里读取所有活跃规则，逐一比对条件。
匹配成功的规则触发对应的动作执行器。
动作执行器把结果输出到告警表、通知通道或自动修复脚本。

注意：规则匹配和动作执行之间要解耦。不要在一个线程里既做匹配又发HTTP请求。匹配应该是纯计算，动作应该走异步队列。否则规则量一大，匹配过程被网络I/O阻塞，整个引擎的吞吐量会直线下降。我之前重构过一个项目，核心改动就是把动作执行扔进Celery异步任务队列，匹配速度立刻提升了5倍。

3.4 一个小例子串起来

我们用一个极简的 Python 代码片段，把三个组件串在一起感受一下：

# 规则定义
rules = [
    Rule("001", "error_count > 5", "send_email(admin@company.com)", 10),
    Rule("002", "status == 500 and response_time > 3000", "webhook('dingtalk')", 20),
]

# 日志事件
event = {"error_count": 8, "status": 500, "response_time": 4200}

# 规则匹配（线性扫描）
for rule in sorted(rules, key=lambda r: r.priority):
    # 假设 eval_condition 是一个安全求值的函数
    if eval_condition(rule.condition_expr, event):
        # 动作执行
        execute_action(rule.action)
        break   # 可以用 break 实现优先级最高的规则触发后停止

这个例子虽然简陋，但三个角色都有了：rules 是规则定义，for 循环 + eval_condition 是规则匹配，execute_action 是动作执行。你把它换成配置文件 + 表达式解析器 + 异步任务队列，就是一套工业级规则引擎的雏形。

3.5 容易踩的坑

聊了这么多理论基础，最后我分享几个我在实战中踩过的坑：

规则数量爆增：一开始只有10条规则，轻松搞定。后来业务方不停加，不到一个月变成800条。线性扫描开始慢。我的做法是“规则分组”——先按日志来源（模块名、级别）粗筛，再挨个匹配，性能好了不少。
动作调用失败导致规则丢失：动作抛出异常，匹配循环直接退出了。明明后续还有匹配成功的规则，却一个也没执行。解决方案：每个动作执行单独 try-except，失败记录到死信队列。
规则优先级形同虚设：团队里有人写规则时忘了设置优先级，默认为0。然后有两个规则优先级相同，执行顺序不确定，结果告警时漏了重要的。后来我强制规定优先级必填，并做校验。