用 Feature Flag 控制后端接口灰度发布的实现方式

Feature Flag 这玩意儿听起来挺唬人，好像一上就能搞出什么高级发布体系。其实放到后端接口灰度这个场景里，核心就一句话：同一个接口，新旧两套逻辑都留着，请求进来的时候根据某个开关决定走哪套。别急着上规则引擎，先把这句话落地，后面就好办了。

先别把 Feature Flag 写成满地 if

我见过比较常见的做法是，在业务代码里到处写：

if get_flag("order_api_v2"):
    return order_v2()
else:
    return order_v1()

写一两个还行，写多了就完蛋。过两个月没人知道哪些接口挂了 flag，也没人敢删。我的习惯是把 flag 的评估收口到接口层，业务 handler 只关心自己是新逻辑还是旧逻辑，别让它知道自己是被灰度选中的。Node、Python、Java 都一样，中间件、装饰器、拦截器，随便挑一个顺手的。

数据模型够用就行

一个 flag 配置我一般存这些东西：key、enabled、default_variant、percentage、whitelist、rules、salt、version。enabled 是总闸，出问题直接关；whitelist 放测试账号和内部用户；rules 可以按 app 版本、地区、租户做定向；percentage 是百分比放量；salt 用来重新洗牌分桶。别一上来设计几十个字段，真用不到。

配置放 Redis 或者数据库都行，关键是要有本地缓存。每个请求都去查一次 Redis，接口还没灰度完，Redis 先扛不住了。我的做法是本地缓存 2 到 5 秒，配置更新后通过 Redis pub/sub 或消息队列通知各实例刷新。缓存时间别太长，不然回滚的时候你以为关了，实际上旧实例还在按老配置跑。

分桶稳定比随机重要

百分比灰度最怕什么？同一个用户上一秒走新接口，下一秒走旧接口。下单走新逻辑，查订单走旧逻辑，数据对不上就热闹了。所以分桶一定要稳定，优先用 user_id，匿名场景用 device_id，别用 IP，也别用 request_id。IP 会飘，request_id 每次都不一样，等于纯随机。

我一般这么算：

import hashlib

def bucket(flag_key, subject, salt=""):
    raw = f"{flag_key}:{subject}:{salt}".encode("utf-8")
    digest = hashlib.md5(raw).hexdigest()
    return int(digest[:8], 16) % 100

bucket(...) < percentage 就走新逻辑。加 salt 是为了以后想重新分桶时不用改代码，改配置就行。别用 Python 内置 hash()，这玩意儿在不同进程里可能不一样，分桶会乱。

在接口层做切换

下面给一个最小可用的 Python 示例，FastAPI 风格，其他框架照着改就行。先写 flag 评估：

def evaluate(flag_key, ctx):
    cfg = get_flag_config(flag_key)  # 带本地缓存
    if not cfg or not cfg["enabled"]:
        return cfg.get("default_variant", "old") if cfg else "old"

    uid = ctx.get("user_id") or ctx.get("device_id") or ""
    if uid and uid in cfg.get("whitelist", []):
        return "new"

    min_version = cfg.get("rules", {}).get("app_version_gte")
    if min_version and ctx.get("app_version", "") < min_version:
        return "old"

    if uid and bucket(flag_key, uid, cfg.get("salt", "")) < cfg.get("percentage", 0):
        return "new"

    return "old"

然后写个装饰器，把新 handler 和旧 handler 绑在一起：

from functools import wraps

def with_flag(flag_key, fallback):
    def deco(new_handler):
        @wraps(new_handler)
        async def wrapper(request, *args, **kwargs):
            ctx = build_context(request)  # 从 header、登录态里取 user_id 等
            variant = evaluate(flag_key, ctx)
            request.state.flag_variant = variant
            if variant == "new":
                return await new_handler(request, *args, **kwargs)
            return await fallback(request, *args, **kwargs)
        return wrapper
    return deco

接口上就这么用：

@with_flag("order_api_v2", fallback=order_v1)
async def order_v2(request):
    return await create_order_new(request)

记得在响应头或者日志里带上 X-Feature-Flag: order_api_v2=new。线上排查问题时，这个信息比“我觉得应该走了新逻辑”有用得多。评估 flag 的时候也要加 try/except，异常了默认走旧逻辑，别让一个开关把主流程搞挂。

发布节奏和回滚

我的习惯是：flag 先默认关闭上线，确认配置能读到；然后加内部白名单，自己人先点一遍；没问题开 1%，盯错误率、P95、业务指标；再 5%、20%、50%、100%。每一步别急着加量，至少观察半小时到一天，看业务复杂度。回滚就是关 enabled 或者把 percentage 改成 0，别等发版。全量跑一周后，把旧代码和 flag 一起删掉，不然这东西会越积越多。

几个实际踩过的坑

第一，新老逻辑共享数据库时，写操作要兼容，最好幂等。第二，别在 flag 评估里做重操作，它会在每个请求上跑。第三，测试要覆盖两个 variant，别只测新逻辑。第四，flag 变更要有审计记录，谁在什么时候把 1% 改成 100% 的，得查得到。第五，客户端如果缓存了接口响应，后端灰度了但前端还拿旧数据，这种要单独考虑。

说到底，Feature Flag 是降低发布风险的工具，不是替代测试的魔法。团队小的话，一个白名单加百分比分桶就够用了，别上来就搞复杂规则引擎。先跑通，再慢慢加。