解释器模式实战：实现自定义的告警规则功能
共 19031字，需浏览 39分钟
·
2021-06-25 04:14
大家好，我是征哥，今天分享一种设计模式，解释器模式。
先来看一个需求：
在告警系统中，有很多规则的配置，如果配置的规则被触发，监控系统就通过短信、微信、邮件等方式发送告警给开发者。比如，每分钟 API 总出错数超过 100 或者每分钟 API 总调用数超过 10000 就触发告警。配置的规则如下：
api_error_per_minute > 9 || api_count_per_minute > 10000
在监控系统中，告警模块只判断是否触发告警。至于每分钟 API 接口出错数、每分钟接口调用数等统计数据的计算，是由其他模块来负责的。其他模块将统计数据放到一个 dict 中，数据的格式如下所示：
apiStat = {}
apiStat["api_error_per_minute"] = 10
apiStat["api_count_per_minute"] = 987
接下来，编写程序，输入是一个字典，代表统计数据 apiStat，和一个字符串，代表告警规则 "api_error_per_minute > 9 || api_count_per_minute > 10000"，输出：True 或 False，True 表述满足告警规则，False 表示不满足。
为了简化代码实现，我们假设自定义的告警规则只包含“||、&&、>、<、==”这五个运算符，其中，“>、<、==”运算符的优先级高于“||、&&”运算符，“&&”运算符优先级高于“||”。在表达式中，任意元素之间需要通过空格来分隔。
除此之外，用户可以自定义要监控的 key，比如前面的 api_error_per_minute、api_count_per_minute。
那么如何写代码实现呢？
更具体一点，请将以下 pass 语句替换成可以执行的代码，其中 rule 字符串是可以自由变化的：
class AlertRuleInterpreter:
def __init__(self, ruleExpression: str):
pass
def interpret(self, stats: dict) -> bool:
pass
if __name__ == "__main__":
rule = "key1 > 100 && key2 < 30 || key3 < 100 || key4 == 88"
interpreter = AlertRuleInterpreter(rule)
stats = {}
stats["key1"] = 101
stats["key3"] = 121
stats["key4"] = 88
alert = interpreter.interpret(stats)
print(alert)
你可以先暂停，思考下怎么写，然后和我这里对比一下：
基础版本
思路：先把字符串按照 || 分割成子串，每一个字串内部的逻辑关系就是 &&，再将字串按照 && 分割成子子串，每一个子子串的内部逻辑关系就三种：>、<、==，这是不是很像一个树？
编码实现每一个字串、子子串对应处理逻辑 Expression 类，这里为统一格式，用到了抽象基类，每一种 Expression 类必须包含 interpret 方法。具体代码如下：
from abc import ABCMeta
from numbers import Real
import re
class Expression(metaclass=ABCMeta):
def interpret(self, stats: dict) -> bool:
pass
class GreaterExpression(Expression):
def __init__(self, express: str = None, key: str = None, value: Real = None):
if express:
elements = re.split(r"\s+", express)
if len(elements) == 3 and elements[1] == ">":
self.key = elements[0]
self.value = float(elements[2])
else:
raise Exception("Invalid GreaterExpression")
elif key and value:
self.key = key
self.value = value
else:
raise Exception("GreaterExpression init error")
def interpret(self, stats: dict) -> bool:
if self.key in stats:
return stats[self.key] > self.value
return False
class LessExpression(Expression):
def __init__(self, express: str = None, key: str = None, value: Real = None):
if express:
elements = re.split(r"\s+", express)
if len(elements) == 3 and elements[1] == "<":
self.key = elements[0]
self.value = float(elements[2])
else:
raise Exception("Invalid LessExpression")
elif key and value:
self.key = key
self.value = value
else:
raise Exception("LessExpression init error")
def interpret(self, stats: dict) -> bool:
if self.key in stats:
return stats[self.key] < self.value
return False
class EqualExpression(Expression):
def __init__(self, express: str = None, key: str = None, value: Real = None):
if express:
elements = re.split(r"\s+", express)
if len(elements) == 3 and elements[1] == "==":
self.key = elements[0]
self.value = float(elements[2])
else:
raise Exception("Invalid EqualExpression")
elif key and value:
self.key = key
self.value = value
else:
raise Exception("EqualExpression init error")
def interpret(self, stats: dict) -> bool:
if self.key in stats:
return stats[self.key] == self.value
return False
class AndExpression(Expression):
def __init__(self, express: str):
self.express_list = []
strExpressions = re.split(r"\s+&&\s+", express)
for express in strExpressions:
if ">" in express:
self.express_list.append(GreaterExpression(express))
elif "<" in express:
self.express_list.append(LessExpression(express))
elif "==" in express:
self.express_list.append(EqualExpression(express))
elif "True" == express or "False" == express:
self.express_list.append(BoolExpression(express))
def interpret(self, stats: dict) -> bool:
for expression in self.express_list:
if expression.interpret(stats) == False:
return False
return True
class OrExpression(Expression):
def __init__(self, express: str):
self.express_list = []
strExpressions = re.split(r"\s+\|\|\s+", express)
for express in strExpressions:
self.express_list.append(AndExpression(express))
def interpret(self, stats: dict) -> bool:
for expression in self.express_list:
if expression.interpret(stats) == True:
return True
return False
class AlertRuleInterpreter:
def __init__(self, ruleExpression: str):
self.expression = OrExpression(ruleExpression)
def interpret(self, stats: dict) -> bool:
return self.expression.interpret(stats)
if __name__ == "__main__":
rule = "key1 > 100 && key2 < 30 || key3 < 100 || key4 == 88"
interpreter = AlertRuleInterpreter(rule)
stats = {}
stats["key1"] = 120
stats["key3"] = 121
stats["key4"] = 88
alert = interpreter.interpret(stats)
print(alert)
这样的设计代码的模式，就叫做解释器模式，英文翻译是 Interpreter Design Pattern。在 GoF 的《设计模式》一书中，它是这样定义的：
Interpreter pattern is used to defines a grammatical representation for a language and provides an interpreter to deal with this grammar.
解释器模式为某个语言定义它的语法（或者叫文法）表示，并定义一个解释器用来处理这个语法。它属于行为型模式。这种模式被用在 SQL 解析、符号处理引擎等。这里的语言并不是我们说的中文和英文，而是任意一个信息的载体，比如本文中的告警规则。
加点难度
现在，我们给他增加点难度，比如说支持小括号，小括号内的表达式会作为一个整体，与 || 或 && 平级，小括号内还嵌套小括号。
与基础版本相比，只需要增加一个 ComplexAlertRuleInterpreter 类，利用栈来先计算括号内的值，要么是 True，要么是 False，然后写回表达式，已达到去除括号的目的，最后在用基础版本的套路来实现即可。
class ComplexAlertRuleInterpreter:
def __init__(self, ruleExpression: str):
self.expression = ruleExpression
def interpret(self, stats: dict) -> bool:
stack = deque()
for express in re.split(r"\s+", self.expression):
if express == "(":
stack.append(express)
continue
elif express == ")":
# 取出括号内容，并计算
tmp_express = deque()
while len(stack) > 0:
tmp = stack.pop()
if tmp == "(":
break
else:
tmp_express.appendleft(tmp)
# 计算结果
result = AlertRuleInterpreter(" ".join(tmp_express)).interpret(stats)
stack.append(str(result))
continue
else:
stack.append(express)
return AlertRuleInterpreter(" ".join(stack)).interpret(stats)
括号去除后，表达式多了'True' 或者 'False' 这样的文本，因此就需要一个 BoolExpression 来处理 'True' 或者 'False' ：
class BoolExpression(Expression):
def __init__(self, express: str):
self.express = express
def interpret(self, stats: dict) -> bool:
if self.express == "True":
return True
return False
最后修改 AndExpression 的初始化部分，增加对 'True' 或者 'False' 表达式的处理：
class AndExpression(Expression):
def __init__(self, express: str):
self.express_list = []
strExpressions = re.split(r"\s+&&\s+", express)
for express in strExpressions:
if ">" in express:
self.express_list.append(GreaterExpression(express))
elif "<" in express:
self.express_list.append(LessExpression(express))
elif "==" in express:
self.express_list.append(EqualExpression(express))
elif "True" == express or "False" == express:
self.express_list.append(BoolExpression(express))
def interpret(self, stats: dict) -> bool:
for expression in self.express_list:
if expression.interpret(stats) == False:
return False
return True
最后， rule 可以增加任意多的括号，main 函数如下：
if __name__ == "__main__":
# rule = "key1 > 100 && key2 < 30"
rule = "key1 > 101 && ( ( key2 < 30 || ( key3 < 100 || key4 == 88 ) ) || False ) && True"
interpreter = ComplexAlertRuleInterpreter(rule)
stats = {}
stats["key1"] = 120
stats["key3"] = 121
stats["key4"] = 88
alert = interpreter.interpret(stats)
print(alert)
完整代码
https://github.com/somenzz/geekbang/blob/master/design-pattern/72-interpreter-demo.py
解释器模式应用场景：
简单来说，程序需要对字符串进行解释的，就像编程语言对代码的解释一样，这种情况下，就需要用到解释器模式。比如说：
需要解释的字符串可以表示为一个抽象的语法树
一个重复出现的问题可以用一种简单的语言来表达
现在比较流行的规则引擎系统
最后的话
以上就是解释器模式的编程实战应用，学习自极客时间的专栏《设计模式之美》，我学完之后，真是受益匪浅，再看自己之前写的代码，真叫一个垃圾。如果你也想提升代码水平，这个专栏将会为你打开新世界的大门：
如果从这里扫码购买的，加我微信「somenzz」，抽奖送书，每三人必中一人，6 月 30 前有效。送书如下：