为什么需要专业的分流系统？

刚接触AB测试(Bucket Testing)时，很多同学会用最朴素的if-else随机分流。实际生产中会遇到三大致命问题：

• 流量倾斜：简单取模会导致某些实验组流量超分配（比如用户ID尾号不均匀）
• 实验干扰：同一个用户在多次请求中被分到不同组，导致行为数据失真
• 无法回溯：临时改代码调整流量比例会破坏实验连续性

分桶算法选型

1. 随机数分层（Random Stratification）

• 时间复杂度：O(1)
• 适合场景：快速验证、临时活动
• 缺点：无法保证用户始终落入同组

2. 一致性哈希分桶（Consistent Hashing Bucket）

• 时间复杂度：O(log n)
• 适合场景：长期实验、需要用户粘性
• 优势：支持动态调整权重且影响面最小

Python核心实现

# 一致性哈希分桶实现
class ABTestBucket:
    def __init__(self, buckets: Dict[str, float]):
        """
        :param buckets: {"A": 0.3, "B": 0.7} 表示30%流量分到A组
        """
        self.ring = []
        self.total = sum(buckets.values())

        # 构建哈希环
        current = 0
        for name, weight in buckets.items():
            upper_bound = current + weight/self.total
            self.ring.append((upper_bound, name))
            current = upper_bound

    def get_bucket(self, user_id: str) -> str:
        """根据用户ID哈希值确定分桶"""
        hash_val = zlib.adler32(user_id.encode()) % 10000 / 10000

        # 二分查找对应区间
        left, right = 0, len(self.ring)
        while left < right:
            mid = (left + right) // 2
            if hash_val > self.ring[mid][0]:
                left = mid + 1
            else:
                right = mid
        return self.ring[left][1]

# 单元测试示例
def test_bucket_distribution():
    bucket = ABTestBucket({"A": 1, "B": 3})
    counter = Counter()
    for i in range(10000):
        counter[bucket.get_bucket(f"user_{i}")] += 1

    assert abs(counter["A"]/10000 - 0.25) < 0.02  # 允许2%误差

Django中间件实现

# middleware.py
class ABTestMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.bucket = ABTestBucket({"control": 0.5, "variant": 0.5})

    def __call__(self, request):
        # 优先读取已分配的分组（防篡改）
        bucket_id = request.COOKIES.get('ab_test_bucket')
        if not bucket_id or bucket_id not in ['control', 'variant']:
            bucket_id = self.bucket.get_bucket(request.user.id if request.user.is_authenticated else request.session.session_key)
            request.set_cookie('ab_test_bucket', bucket_id, max_age=30*24*3600)

        request.ab_test_bucket = bucket_id
        return self.get_response(request)

监控看板搭建

推荐使用Prometheus + Grafana组合：

1. 指标采集：
2. ab_test_requests_total{experiment="首页改版", bucket="A"}

3. ab_test_conversion_rate{experiment="注册流程", bucket="B"}

4. 关键监控项：

5. 分流均匀性（卡方检验）
6. 转化率标准差（Z检验）
7. 置信区间可视化

避坑指南

冷启动问题

• 使用样本量计算器
• 初期建议每组至少500UV

样本污染处理

• 新用户实验：按注册时间过滤历史数据
• 老用户实验：设置足够长的冷却期（如7天无访问）

扩展思考

这套架构稍加改造就能支持灰度发布(Canary Release)： 1. 增加基于设备/地区的分桶规则 2. 添加异常熔断机制 3. 结合Feature Flag管理系统

实际使用半年后，我们团队发现这套系统最棒的特点是：当你想临时加个实验组时，再也不需要半夜上线改代码了。通过动态调整分桶权重，可以随时控制流量比例，这大概就是工程化带来的幸福感吧！