最近接到了一个新的需求。需求本身是一个简单的运营活动,不过这个运营活动并不是长期存在的,需要通过后台设置生效时间。
抽象一下的话就是需要通过开关来控制一个功能是否生效,也就是特性开关(Feature Flags)模式。
Martin Fowler 先生写过一篇特性开关模式的文章,感兴趣的读者可以深入阅读。
针对本次应用场景和日后的类似需求,我用 Redis 作为存储实现了一个简单的特性开关。
数据结构
定义Feature
类,open
属性表示特性开关是否打开,start
和end
代表特性的生效时间(均为 None 表示该特性长期生效),
1 2 3 4 5 6 7 8 9
| from datetime import datetime from typing import Optional
from pydantic import BaseModel class Feature(BaseModel): name: str open: bool start: Optional[datetime] end: Optional[datetime]
|
设置特性开关状态
直接使用Feature
的name
作为 key, 将Feature
对象设置到Redis
缓存中。
1 2 3 4 5 6 7 8 9
| from redis import Redis
client = Redis()
def build_key(name: str): return f'FEATURE:{name}'
def set_feature(feature: Feature, client: Redis) -> None: client.set(build_key(feature.name), feature.json())
|
读取特性开关状态
从Redis
缓存中查询指定name
的Feature
, 根据open
,start
和end
的值来判断该特性是否为开启状态。
1 2 3 4 5 6 7 8 9 10 11
| def get_feature(name: str, client: Redis) -> Feature | None: raw = client.get(build_key(name)) return raw and Feature.parse_raw(raw)
def get_feature_status(name: str, date: datetime, client: Redis) -> bool: feature = get_feature(name, client) if not (feature and feature.open): return False if feature.start and feature.end and not( feature.start < date < feature.end): return False return True
|
我这里的get_status
函数并没有直接判断当前时间是否在特性的生效时间内,而是需要显式的传入date
参数(事实client
参数也一直是显式传入的)。
这样的设计会确保特性开关相关的函数都是纯函数,没有任何副作用,方便编写单元测试,并且使用起来可以更灵活(例如可以切换数据源为其他数据库或直接存在内存对象中)。
使用特性开关
我们可以在代码逻辑中直接根据指定特性的状态来走不同的分支,也可以将相关接口暴露给前端,有前端根据不同的状态控制页面逻辑。
1 2 3 4
| def process() -> None: if get_feature_status(FEATURE_A, client): do_a() do_b()
|
1 2 3 4
| function Component() { const featureA = getFeatureFlag(FEATURE_A); return <>{featureA && <ComponentA />}</>; }
|
使用装饰器
可以将判断特性开关状态的逻辑封装为一个装饰器,使用起来更加灵活,代码也更简洁。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29
| from functools import wraps from typing import Callable
def do_nothing(*args, **kwargs): pass
def check_feature(feature_name: str): status = get_feature_status(feature_name, datetime.now(), client)
def outter(func: Callable):
@wraps(func) def inner(*args, **kwargs): return func(*args, **kwargs)
return inner if status else do_nothing
return outter
@check_feature(FEATURE_A) def try_do_a(): do_a()
def do_a(): print("Do A")
def process(): try_do_a()
|
总结
特性开关是一个不错的软件实践,适用于单分支发布的 SASS 项目,一个显著的优势是可以在功能上线前就将代码集成到主分支中(避免较晚合并代码时的痛苦),在测试环境通过打开特性开关来测试功能,同时不影响线上环境的正常使用。
感谢阅读。