..

如何编写低耦合可维护的 Python 代码

经常听到一个论调:设计模式本质为弥补编程语言自身的缺陷,例如单例模式的存在,是因为 Java 本身不提供单例对象创建,而 Python 中原生的 import 就是 singleton 的天然实现。

但个人观点这句话仅看到了表象,设计模式初衷还是帮助我们编写更加优雅的代码,背后的目的与语言本身无关。

这篇文章以大型项目痛苦之源 Python 语言为例,通过近期工作中的两个实践案例,尝试分享如何通过常见设计模式,编写可扩展可维护代码的一点点经验。

背景

近期公司内部发起告警阈值托管项目,故名思义通过历史时序数据,自动学习并清洗合理的告警规则+阈值。解放运维人员双手的同时,通过程序自动化的方式保证更高的准确率/召回率。

为了规避智能算法黑盒的弊端:用户难以理解告警触发逻辑,进而一步无法调节业务预期 :(

所以项目初期针对不同类型的时序指标,基于人的专家经验,提炼了一系列告警规则模版,例如成功量下跌、成功量跌零、失败数上涨、历史新增异常等等)。

理想效果:用户输入一周预计告警个数后,清洗模块自动输出可理解的静态告警规则..

1. 阈值清洗流水线抽象 & 扩展

清洗模块代码落地的过程中,发现不同“模版”清洗的流程大同小异,无非 1)拉取历史数据,2)数据预处理,3)特征提取, 4)上下限阈值计算等。

自然而然地将整理清洗流程,抽象固定为 pipeline 基类,不同“告警模版”理论只需实现对应的「特征提取」逻辑即可,但如何进一步兼顾清洗流程的扩展性呢?

例如:

  1. 历史数据完整性检测:数据预处理前,检查拉取的历史数据,例如判断是否至少满足一天,否则中断流程
  2. 阈值结果人为干预:阈值完成清洗后,人工微调告警规则,例如调整夜间触发告警的持续时间
  3. 阈值持久化保存:阈值完成清洗后,进一步执行阈值的持久化(测试用例、实时服务等场景不需要)

这时曾拜读的 tomcat 源码 Lifecycle 突然跃入脑中!

有没有可能将上述的“个性化”处理逻辑,以事件的方式动态注入至 pipeline 中,最终在不同阶段的“埋点”触发:

类图参考:

改造后的 pipeline 片段:
p.s. 完整代码参考:https://github.com/daya0576/python_design_patterns/blob/master/observers/pipeline.py#L7

from abc import ABC, abstractmethod
from observers.events import LifecycleListener
from observers.lifecycle import EventType, Lifecycle
from observers.utils import LifecycleSupport


class BasePipeline(Lifecycle, ABC):
    def __init__(self):
        self.lifecycle = LifecycleSupport(self)

    @abstractmethod
    def _do_load_data(self):
        ...

    @abstractmethod
    def _do_process(self):
        ...

    def start(self):
        self.lifecycle.fire_lifecycle_listener(EventType.START_EVENT)

    def stop(self):
        self.lifecycle.fire_lifecycle_listener(EventType.STOP_EVENT)

    def add_listener(self, listener: LifecycleListener):
        self.lifecycle.add_lifecycle_listener(listener)

    def process(self):
        print("Start..")
        self.start()
        self._do_load_data()
        self._do_process()
        self.stop()
        print("Stop")


class SimplePipeline(BasePipeline):
    def _do_load_data(self):
        print("Do load data")

    def _do_process(self):
        print("Do process")

最终执行效果:

pipeline = SimplePipeline()
pipeline.add_listener(FooLifecycleListener())
pipeline.add_listener(BarLifecycleListener())

pipeline.process()

# ---------------------
# > Start..
# > Foo event fired...
# > Do load data
# > Do process
# > Bar event fired...
# > Stop

如上 Lifecycle demo,本质上为“观察者设计模式”的一种实现。清爽解耦的代码,只有在自由灵活新增代码逻辑的时候,才能懂得它的好。

2. 告警规则覆盖策略

上文清洗逻辑成功落地后,如何针对不同监控项,覆盖正确的「告警规则模版」成为了新的难题。

不难理解若监控包含耗时/失败数指标,直接覆盖耗时上涨/失败数上涨告警规则即可。

但假如判断有误,针对失败量监控覆盖「跌零规则模版」,最终告警显示 PROCESS_FAIL 错误码 10m 跌零,请立即处理,那真的要笑掉大牙了。

那除了人工标定,还有其他“策略”可以自动分析吗?
部分解法:

  1. 指标类型:例如明确的失败量、耗时等指标
  2. 故障等级:若关联 P1P2 故障等极的规则,必然是业务量级监控
  3. 告警规则:甚至可以分析现有监控告警规则,判断量级下降 or 上升代表异常。

模拟如上不同场景的策略,是否可以编写直白的伪代码?

success_count_risk = DefaultSuccessCountRisk()
time_cost_risk = DefaultTimecostRisk()

strategies = [
    # 覆盖 P1P2 业务,SPM 监控类型,成功量指标的风险:成功量下跌/跌零
    P1P2(High(SPM(risk))),
    # 覆盖 P3P4 业务,白天时间段,耗时指标的风险
    P3P4(Daytime(time_cost_risk)),
]

参考装饰器设计模式,一个更加恰当咖啡制作的例子,通过不同类的灵活“装饰”,最终获取这杯咖啡的描述与价格:

// 来一杯 Espresso
beverage = Espresso()
print(beverage)
// Espresso$1.99

// 来一杯调料为豆浆,摩卡,奶泡的 HouseBlend 咖啡
beverage = HouseBlend()
beverage = Soy(Mocha(Whip(beverage)))
print(beverage)
// HouseBlend, Whip, Mocha, Soy$1.39

类图参考:

代码参考:https://github.com/daya0576/python_design_patterns/blob/master/wrapper/decorator.py

总结

上文两个实战案例,对应观察者模式(Observer)、装饰器模式(Wrapper)两种设计模式,本质上都是“开闭原则”的一种最佳实践:对于扩展,类应该是「开放」的;对于修改,类应该是「封闭」的。

简而言之,编写代码时需要区分程序中的 易变稳定 部分。对于未来可预见的新增需求,尽可能不修改原有代码,而是通过简单组合的方式快速扩展。

参考:

  1. Python 不需要设计模式?:https://changchen.me/blog/20201114/why-u-dont-need-design-pattern-in-python/
  2. 《深入剖析 Tomcat》:https://book.douban.com/subject/10426640/
  3. 《Head First 设计模式》学习笔记:https://changchen.me/blog/20200613/design-pattern/