第6讲：实验平台建设——从零搭建企业级 A/B 测试平台

一、为什么要建实验平台

散装实验 vs 平台化实验
────────────────────────────────────────────────────
散装实验（前几讲的方式） 平台化实验
────────────────────────────────────────────────────
手动配置实验参数 可视化配置界面
实验代码散落在各个服务中 统一的实验 SDK
结果靠脚本汇总 实时仪表盘
权限不分，谁都能改配置 角色权限管理
实验之间互相干扰 正交分层保障
没有审计日志 全链路可追溯

核心目标：让产品经理也能自助开实验，让工程师只需要接入一次 SDK。

二、平台整体架构

┌──────────────────────────────────────────────────────────────┐
│ 实验平台前端 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │实验管理 │ │指标配置 │ │报表查看 │ │系统设置 │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└──────────────────────────┬───────────────────────────────────┘
│
┌──────────────────────────▼───────────────────────────────────┐
│ API 网关 │
└──────────────────────────┬───────────────────────────────────┘
│
┌──────────────────────────▼───────────────────────────────────┐
│ 核心服务层 │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │实验服务 │ │分流服务 │ │指标服务 │ │报表服务 │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└──────────────────────────┬───────────────────────────────────┘
│
┌──────────────────────────▼───────────────────────────────────┐
│ 数据层 │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │MySQL │ │Redis │ │Kafka │ │ClickHouse│ │
│ │(元数据) │ │(缓存) │ │(事件流) │ │(分析) │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└──────────────────────────────────────────────────────────────┘

三、核心数据模型

// platform/models.go
package platform

import (
"time"
)

// ─── 实验定义 ───

// Experiment 实验
type Experiment struct {
ID string `json:"id" gorm:"primaryKey"`
Name string `json:"name"`
Description string `json:"description"`
Creator string `json:"creator"`
Status string `json:"status"` // draft, running, paused, completed
Layer string `json:"layer"` // 所属分层
Traffic int `json:"traffic"` // 流量比例 1-100
Variants []Variant `json:"variants" gorm:"foreignKey:ExpID"`
Metrics []Metric `json:"metrics" gorm:"foreignKey:ExpID"`
Filters []Filter `json:"filters" gorm:"foreignKey:ExpID"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
StartedAt *time.Time `json:"started_at"`
EndedAt *time.Time `json:"ended_at"`
}

// Variant 变体
type Variant struct {
ID string `json:"id" gorm:"primaryKey"`
ExpID string `json:"exp_id"`
Name string `json:"name"`
Description string `json:"description"`
Weight int `json:"weight"` // 权重
Config string `json:"config"` // JSON 配置
IsControl bool `json:"is_control"` // 是否为对照组
}

// Metric 指标
type Metric struct {
ID string `json:"id" gorm:"primaryKey"`
ExpID string `json:"exp_id"`
Name string `json:"name"`
Type string `json:"type"` // counter, gauge, ratio, rate
Unit string `json:"unit"` // %, ms, count, score
Direction string `json:"direction"` // higher_is_better, lower_is_better
Formula string `json:"formula"` // 计算表达式
EventName string `json:"event_name"` // 埋点事件名
}

// Filter 过滤条件
type Filter struct {
ID string `json:"id" gorm:"primaryKey"`
ExpID string `json:"exp_id"`
Field string `json:"field"` // user_level, platform, region
Op string `json:"op"` // eq, neq, in, not_in
Value string `json:"value"` // "vip", "[\\"ios\\",\\"android\\"]"
}

// ─── 实验事件 ───

// ExperimentEvent 实验事件
type ExperimentEvent struct {
ID string `json:"id"`
ExpID string `json:"exp_id"`
VariantID string `json:"variant_id"`
UserID string `json:"user_id"`
SessionID string `json:"session_id"`
EventType string `json:"event_type"` // exposure, click, conversion, custom
EventName string `json:"event_name"`
Properties string `json:"properties"` // JSON
Value float64 `json:"value"`
Timestamp time.Time `json:"timestamp"`
}

// ─── 实验结果 ───

// ExperimentResult 实验结果
type ExperimentResult struct {
ExpID string `json:"exp_id"`
Status string `json:"status"`
Duration string `json:"duration"`
SampleSize int `json:"sample_size"`
Variants []VariantResult `json:"variants"`
Conclusion string `json:"conclusion"` // winner, no_significance, inconclusive
}

type VariantResult struct {
VariantID string `json:"variant_id"`
Name string `json:"name"`
SampleSize int `json:"sample_size"`
Metrics []MetricResult `json:"metrics"`
IsWinner bool `json:"is_winner"`
}

type MetricResult struct {
MetricName string `json:"metric_name"`
ControlVal float64 `json:"control_value"`
VariantVal float64 `json:"variant_value"`
Lift float64 `json:"lift"` // 提升百分比
PValue float64 `json:"p_value"`
Significant bool `json:"significant"`
CI_Lower float64 `json:"ci_lower"` // 置信区间下限
CI_Upper float64 `json:"ci_upper"` // 置信区间上限
}

四、实验管理服务

// platform/service.go
package platform

import (
"fmt"
"sync"
"time"
"gorm.io/gorm"
)

// ─── 实验服务 ───

// ExperimentService 实验服务
type ExperimentService struct {
db *gorm.DB
cache *sync.Map
eventBus chan *ExperimentEvent
}

// NewExperimentService 创建实验服务
func NewExperimentService(db *gorm.DB) *ExperimentService {
svc := &ExperimentService{
db: db,
cache: &sync.Map{},
eventBus: make(chan *ExperimentEvent, 10000),
}
go svc.processEvents()
return svc
}

// CreateExperiment 创建实验
func (s *ExperimentService) CreateExperiment(exp *Experiment) error {
// 1. 校验
if err := s.validateExperiment(exp); err != nil {
return err
}

// 2. 检查分层冲突
if err := s.checkLayerConflict(exp); err != nil {
return err
}

// 3. 保存
exp.Status = "draft"
return s.db.Create(exp).Error
}

// StartExperiment 启动实验
func (s *ExperimentService) StartExperiment(expID string) error {
now := time.Now()

// 1. 更新状态
result := s.db.Model(&Experiment{}).
Where("id = ? AND status = ?", expID, "draft").
Updates(map[string]interface{}{
"status": "running",
"started_at": &now,
})

if result.RowsAffected == 0 {
return fmt.Errorf("实验不存在或状态不允许启动")
}

// 2. 加载到缓存
var exp Experiment
s.db.Preload("Variants").First(&exp, "id = ?", expID)
s.cache.Store("exp:"+expID, &exp)

return nil
}

// StopExperiment 停止实验
func (s *ExperimentService) StopExperiment(expID string) error {
now := time.Now()
return s.db.Model(&Experiment{}).
Where("id = ? AND status = ?", expID, "running").
Updates(map[string]interface{}{
"status": "completed",
"ended_at": &now,
}).Error
}

// GetExperimentResult 获取实验结果
func (s *ExperimentService) GetExperimentResult(expID string) (*ExperimentResult, error) {
var exp Experiment
if err := s.db.Preload("Variants").Preload("Metrics").First(&exp, "id = ?", expID).Error; err != nil {
return nil, err
}

result := &ExperimentResult{
ExpID: expID,
Status: exp.Status,
}

// 统计每个变体的数据
for _, v := range exp.Variants {
vr := VariantResult{
VariantID: v.ID,
Name: v.Name,
}

// 统计样本数
var count int64
s.db.Model(&ExperimentEvent{}).
Where("exp_id = ? AND variant_id = ? AND event_type = ?", expID, v.ID, "exposure").
Count(&count)
vr.SampleSize = int(count)

// 计算每个指标
for _, m := range exp.Metrics {
mr := s.calculateMetric(expID, v.ID, m)
vr.Metrics = append(vr.Metrics, mr)
}

result.Variants = append(result.Variants, vr)
}

// 判断胜者
result.Conclusion = s.determineWinner(result)

return result, nil
}

func (s *ExperimentService) validateExperiment(exp *Experiment) error {
if len(exp.Variants) < 2 {
return fmt.Errorf("实验至少需要 2 个变体")
}

totalWeight := 0
for _, v := range exp.Variants {
totalWeight += v.Weight
}
if totalWeight != 100 {
return fmt.Errorf("变体权重之和必须为 100，当前为 %d", totalWeight)
}

hasControl := false
for _, v := range exp.Variants {
if v.IsControl {
hasControl = true
break
}
}
if !hasControl {
return fmt.Errorf("实验必须有一个对照组")
}

return nil
}

func (s *ExperimentService) checkLayerConflict(exp *Experiment) error {
var runningExps []Experiment
s.db.Where("layer = ? AND status = ?", exp.Layer, "running").Find(&runningExps)

if len(runningExps) > 0 {
return fmt.Errorf("分层 %s 已有正在运行的实验，请等待完成后再创建", exp.Layer)
}
return nil
}

func (s *ExperimentService) calculateMetric(expID, variantID string, metric Metric) MetricResult {
mr := MetricResult{MetricName: metric.Name}

// 根据指标类型计算
switch metric.Type {
case "counter":
var value float64
s.db.Model(&ExperimentEvent{}).
Select("COUNT(*) as cnt").
Where("exp_id = ? AND variant_id = ? AND event_name = ?", expID, variantID, metric.EventName).
Scan(&value)
mr.VariantVal = value

case "rate":
var numerator, denominator float64
s.db.Model(&ExperimentEvent{}).
Select("COUNT(*) as cnt").
Where("exp_id = ? AND variant_id = ? AND event_name = ? AND value > 0", expID, variantID, metric.EventName).
Scan(&numerator)
s.db.Model(&ExperimentEvent{}).
Select("COUNT(*) as cnt").
Where("exp_id = ? AND variant_id = ? AND event_name = ?", expID, variantID, metric.EventName).
Scan(&denominator)
if denominator > 0 {
mr.VariantVal = numerator / denominator
}

case "gauge":
var avg float64
s.db.Model(&ExperimentEvent{}).
Select("AVG(value) as avg_val").
Where("exp_id = ? AND variant_id = ? AND event_name = ?", expID, variantID, metric.EventName).
Scan(&avg)
mr.VariantVal = avg
}

return mr
}

func (s *ExperimentService) determineWinner(result *ExperimentResult) string {
if len(result.Variants) < 2 {
return "inconclusive"
}

// 简单的优胜判定（生产环境需要 T 检验/Z 检验）
control := result.Variants[0]
best := control

for _, v := range result.Variants[1:] {
if len(v.Metrics) > 0 && v.Metrics[0].VariantVal > best.Metrics[0].VariantVal {
best = v
}
}

if best.VariantID != control.VariantID {
return fmt.Sprintf("winner:%s", best.VariantID)
}

return "no_significance"
}

// processEvents 异步处理事件
func (s *ExperimentService) processEvents() {
batch := make([]*ExperimentEvent, 0, 100)
ticker := time.NewTicker(time.Second)

for {
select {
case event := <-s.eventBus:
batch = append(batch, event)
if len(batch) >= 100 {
s.flushEvents(batch)
batch = batch[:0]
}
case <-ticker.C:
if len(batch) > 0 {
s.flushEvents(batch)
batch = batch[:0]
}
}
}
}

func (s *ExperimentService) flushEvents(events []*ExperimentEvent) {
s.db.CreateInBatches(events, 100)
}

五、分流 SDK（客户端接入）

// platform/sdk.go
package platform

import (
"context"
"encoding/json"
"fmt"
"net/http"
"sync"
"time"
)

// ─── 客户端 SDK ───

// ExperimentSDK 实验 SDK
type ExperimentSDK struct {
apiEndpoint string
apiKey string
experiments map[string]*cachedExperiment
mu sync.RWMutex
httpClient *http.Client
}

type cachedExperiment struct {
Experiment
ExpireAt time.Time
}

// NewExperimentSDK 创建 SDK
func NewExperimentSDK(endpoint, apiKey string) *ExperimentSDK {
return &ExperimentSDK{
apiEndpoint: endpoint,
apiKey: apiKey,
experiments: make(map[string]*cachedExperiment),
httpClient: &http.Client{
Timeout: time.Second * 5,
},
}
}

// GetVariant 获取用户所属变体
func (sdk *ExperimentSDK) GetVariant(ctx context.Context, expID, userID string) (*Variant, error) {
// 1. 从本地缓存获取实验配置
exp, err := sdk.getExperiment(expID)
if err != nil {
return nil, err
}

// 2. 检查用户是否命中实验流量
if !sdk.isInTraffic(userID, exp.Traffic) {
return nil, nil // 未命中实验
}

// 3. 检查过滤条件
if !sdk.passFilters(userID, exp.Filters) {
return nil, nil
}

// 4. 分流
variant := sdk.assignVariant(userID, exp.Variants)

// 5. 上报曝光事件
sdk.reportExposure(expID, variant.ID, userID)

return variant, nil
}

// getExperiment 获取实验配置（带本地缓存）
func (sdk *ExperimentSDK) getExperiment(expID string) (*Experiment, error) {
sdk.mu.RLock()
cached, ok := sdk.experiments[expID]
sdk.mu.RUnlock()

if ok && time.Now().Before(cached.ExpireAt) {
return &cached.Experiment, nil
}

// 从服务端拉取
exp, err := sdk.fetchExperiment(expID)
if err != nil {
return nil, err
}

sdk.mu.Lock()
sdk.experiments[expID] = &cachedExperiment{
Experiment: *exp,
ExpireAt: time.Now().Add(time.Minute * 5),
}
sdk.mu.Unlock()

return exp, nil
}

func (sdk *ExperimentSDK) fetchExperiment(expID string) (*Experiment, error) {
url := fmt.Sprintf("%s/api/v1/experiments/%s", sdk.apiEndpoint, expID)
req, _ := http.NewRequest("GET", url, nil)
req.Header.Set("Authorization", "Bearer "+sdk.apiKey)

resp, err := sdk.httpClient.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()

var exp Experiment
json.NewDecoder(resp.Body).Decode(&exp)
return &exp, nil
}

func (sdk *ExperimentSDK) isInTraffic(userID string, traffic int) bool {
// 一致性哈希决定用户是否在实验流量内
hash := hashUserID(userID)
return hash%100 < traffic
}

func (sdk *ExperimentSDK) passFilters(userID string, filters []Filter) bool {
// 简化实现：实际需要查用户画像
return true
}

func (sdk *ExperimentSDK) assignVariant(userID string, variants []Variant) *Variant {
hash := hashUserID(userID)

cumulative := 0
for _, v := range variants {
cumulative += v.Weight
if hash%100 < cumulative {
return &v
}
}

return &variants[0]
}

func (sdk *ExperimentSDK) reportExposure(expID, variantID, userID string) {
event := ExperimentEvent{
ExpID: expID,
VariantID: variantID,
UserID: userID,
EventType: "exposure",
Timestamp: time.Now(),
}

// 异步上报
go func() {
data, _ := json.Marshal(event)
url := fmt.Sprintf("%s/api/v1/events", sdk.apiEndpoint)
http.Post(url, "application/json", bytes.NewReader(data))
}()
}

// ─── HTTP 中间件 ───

// ExperimentMiddleware 实验中间件
func (sdk *ExperimentSDK) Middleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
userID := extractUserID(r)

// 遍历所有活跃实验
exps := sdk.getActiveExperiments()
for _, exp := range exps {
variant, err := sdk.GetVariant(r.Context(), exp.ID, userID)
if err == nil && variant != nil {
// 注入变体信息
ctx := context.WithValue(r.Context(), "exp_"+exp.ID, variant.ID)
r = r.WithContext(ctx)
w.Header().Set("X-Exp-"+exp.ID, variant.ID)
}
}

next.ServeHTTP(w, r)
})
}

六、API 路由

// platform/api.go
package platform

import (
"encoding/json"
"net/http"
)

// ExperimentHandler 实验 API 处理器
type ExperimentHandler struct {
service *ExperimentService
}

// RegisterRoutes 注册路由
func (h *ExperimentHandler) RegisterRoutes(mux *http.ServeMux) {
mux.HandleFunc("/api/v1/experiments", h.handleExperiments)
mux.HandleFunc("/api/v1/experiments/", h.handleExperimentByID)
mux.HandleFunc("/api/v1/events", h.handleEvents)
mux.HandleFunc("/api/v1/reports/", h.handleReport)
}

func (h *ExperimentHandler) handleExperiments(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case "GET":
// 列出实验
var exps []Experiment
h.service.db.Find(&exps)
json.NewEncoder(w).Encode(exps)

case "POST":
// 创建实验
var exp Experiment
json.NewDecoder(r.Body).Decode(&exp)
if err := h.service.CreateExperiment(&exp); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
json.NewEncoder(w).Encode(exp)
}
}

func (h *ExperimentHandler) handleExperimentByID(w http.ResponseWriter, r *http.Request) {
expID := extractPathParam(r.URL.Path, "experiments/")

switch r.Method {
case "GET":
exp, err := h.service.GetExperimentResult(expID)
if err != nil {
http.Error(w, "not found", http.StatusNotFound)
return
}
json.NewEncoder(w).Encode(exp)

case "PUT":
action := r.URL.Query().Get("action")
var err error
switch action {
case "start":
err = h.service.StartExperiment(expID)
case "stop":
err = h.service.StopExperiment(expID)
}
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
w.WriteHeader(http.StatusOK)

case "DELETE":
h.service.db.Delete(&Experiment{}, "id = ?", expID)
w.WriteHeader(http.StatusNoContent)
}
}

func (h *ExperimentHandler) handleEvents(w http.ResponseWriter, r *http.Request) {
if r.Method != "POST" {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}

var events []ExperimentEvent
json.NewDecoder(r.Body).Decode(&events)

for _, e := range events {
h.service.eventBus <- &e
}

w.WriteHeader(http.StatusAccepted)
}

func (h *ExperimentHandler) handleReport(w http.ResponseWriter, r *http.Request) {
expID := extractPathParam(r.URL.Path, "reports/")
result, err := h.service.GetExperimentResult(expID)
if err != nil {
http.Error(w, "not found", http.StatusNotFound)
return
}
json.NewEncoder(w).Encode(result)
}

func extractPathParam(path, prefix string) string {
// 简化实现
return path[len(prefix):]
}

七、数据管道与实时计算

// platform/pipeline.go
package platform

import (
"time"
"sync"
)

// ─── 实时聚合 ───

// RealTimeAggregator 实时聚合器
type RealTimeAggregator struct {
windows map[string]*SlidingWindow // key: expID:variantID:metric
mu sync.Mutex
}

// SlidingWindow 滑动窗口
type SlidingWindow struct {
buckets []*Bucket
duration time.Duration
interval time.Duration
mu sync.Mutex
}

type Bucket struct {
StartTime time.Time
Count int64
Sum float64
}

// NewRealTimeAggregator 创建实时聚合器
func NewRealTimeAggregator(windowDuration, bucketInterval time.Duration) *RealTimeAggregator {
return &RealTimeAggregator{
windows: make(map[string]*SlidingWindow),
}
}

// AddEvent 添加事件
func (rta *RealTimeAggregator) AddEvent(event *ExperimentEvent) {
key := event.ExpID + ":" + event.VariantID + ":" + event.EventName

rta.mu.Lock()
sw, ok := rta.windows[key]
if !ok {
sw = NewSlidingWindow(time.Minute*5, time.Second*10)
rta.windows[key] = sw
}
rta.mu.Unlock()

sw.Add(event.Value)
}

// GetCurrentStats 获取当前统计数据
func (rta *RealTimeAggregator) GetCurrentStats(expID, variantID, eventName string) map[string]float64 {
key := expID + ":" + variantID + ":" + eventName

rta.mu.Lock()
sw, ok := rta.windows[key]
rta.mu.Unlock()

if !ok {
return nil
}

return map[string]float64{
"count": float64(sw.Count()),
"sum": sw.Sum(),
"avg": sw.Avg(),
}
}

func NewSlidingWindow(duration, interval time.Duration) *SlidingWindow {
numBuckets := int(duration / interval)
buckets := make([]*Bucket, numBuckets)
now := time.Now()

for i := 0; i < numBuckets; i++ {
buckets[i] = &Bucket{
StartTime: now.Add(-duration + time.Duration(i)*interval),
}
}

return &SlidingWindow{
buckets: buckets,
duration: duration,
interval: interval,
}
}

func (sw *SlidingWindow) Add(value float64) {
sw.mu.Lock()
defer sw.mu.Unlock()

now := time.Now()
bucketIndex := int(now.Sub(now.Truncate(sw.duration)) / sw.interval)
if bucketIndex >= len(sw.buckets) {
bucketIndex = len(sw.buckets) – 1
}

sw.buckets[bucketIndex].Count++
sw.buckets[bucketIndex].Sum += value
}

func (sw *SlidingWindow) Count() int64 {
sw.mu.Lock()
defer sw.mu.Unlock()

var total int64
for _, b := range sw.buckets {
total += b.Count
}
return total
}

func (sw *SlidingWindow) Sum() float64 {
sw.mu.Lock()
defer sw.mu.Unlock()

var total float64
for _, b := range sw.buckets {
total += b.Sum
}
return total
}

func (sw *SlidingWindow) Avg() float64 {
count := sw.Count()
if count == 0 {
return 0
}
return sw.Sum() / float64(count)
}

八、WebSocket 实时推送

// platform/websocket.go
package platform

import (
"encoding/json"
"net/http"
"sync"
"github.com/gorilla/websocket"
)

// ─── WebSocket Hub ───

// WSHub WebSocket 连接管理器
type WSHub struct {
clients map[*WSClient]bool
broadcast chan []byte
register chan *WSClient
unregister chan *WSClient
mu sync.Mutex
}

// WSClient WebSocket 客户端
type WSClient struct {
hub *WSHub
conn *websocket.Conn
send chan []byte
expIDs []string // 订阅的实验 ID
}

// NewWSHub 创建 WebSocket Hub
func NewWSHub() *WSHub {
return &WSHub{
clients: make(map[*WSClient]bool),
broadcast: make(chan []byte, 256),
register: make(chan *WSClient),
unregister: make(chan *WSClient),
}
}

// Run 启动 Hub
func (hub *WSHub) Run() {
for {
select {
case client := <-hub.register:
hub.mu.Lock()
hub.clients[client] = true
hub.mu.Unlock()

case client := <-hub.unregister:
hub.mu.Lock()
if _, ok := hub.clients[client]; ok {
delete(hub.clients, client)
close(client.send)
}
hub.mu.Unlock()

case message := <-hub.broadcast:
hub.mu.Lock()
for client := range hub.clients {
select {
case client.send <- message:
default:
close(client.send)
delete(hub.clients, client)
}
}
hub.mu.Unlock()
}
}
}

// PushExperimentUpdate 推送实验更新
func (hub *WSHub) PushExperimentUpdate(result *ExperimentResult) {
data, _ := json.Marshal(map[string]interface{}{
"type": "experiment_update",
"data": result,
})
hub.broadcast <- data
}

// ServeWs WebSocket 服务
func ServeWs(hub *WSHub, w http.ResponseWriter, r *http.Request) {
upgrader := websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool { return true },
}

conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
return
}

client := &WSClient{
hub: hub,
conn: conn,
send: make(chan []byte, 256),
}
hub.register <- client

go client.writePump()
go client.readPump()
}

func (c *WSClient) writePump() {
for message := range c.send {
c.conn.WriteMessage(websocket.TextMessage, message)
}
c.conn.Close()
}

func (c *WSClient) readPump() {
defer func() {
c.hub.unregister <- c
c.conn.Close()
}()

for {
_, _, err := c.conn.ReadMessage()
if err != nil {
break
}
}
}

九、启动入口

// cmd/platform.go
package main

import (
"log"
"net/http"
"platform"
"gorm.io/driver/mysql"
"gorm.io/gorm"
)

func main() {
// 1. 数据库
dsn := "root:password@tcp(127.0.0.1:3306)/experiment?charset=utf8mb4"
db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{})
if err != nil {
log.Fatal(err)
}
db.AutoMigrate(&platform.Experiment{}, &platform.Variant{},
&platform.Metric{}, &platform.Filter{}, &platform.ExperimentEvent{})

// 2. 实验服务
svc := platform.NewExperimentService(db)

// 3. WebSocket Hub
hub := platform.NewWSHub()
go hub.Run()

// 4. API 路由
handler := &platform.ExperimentHandler{service: svc}
mux := http.NewServeMux()
handler.RegisterRoutes(mux)
mux.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
platform.ServeWs(hub, w, r)
})

// 5. 启动
log.Println("实验平台启动于 :8080")
log.Fatal(http.ListenAndServe(":8080", mux))
}

十、本讲坑位

#

坑位

现象

解决方案

十一、本讲小结

┌─────────────────────────────────────────────────────────────┐
│ 第6讲 核心收获 │
│ │
│ ✅ 理解了实验平台的整体架构（前端→API→服务→数据） │
│ ✅ 实现了完整的实验数据模型（实验/变体/指标/事件） │
│ ✅ 实现了实验 CRUD 服务（创建/启动/停止/结果查询） │
│ ✅ 实现了客户端 SDK（本地缓存 + 分流 + 曝光上报） │
│ ✅ 实现了 HTTP 中间件，对业务代码无侵入 │
│ ✅ 实现了实时聚合器（滑动窗口统计） │
│ ✅ 实现了 WebSocket 实时推送 │
│ ✅ 知道了 6 个常见坑位及其应对方案 │
│ │
│ 核心理念：实验平台的价值在于让实验变得廉价和可靠 │
└─────────────────────────────────────────────────────────────┘

下一讲预告：《统计学基础——AB 实验中的显著性检验与陷阱》

P 值是什么？什么时候用 T 检验什么时候用 Z 检验？多重比较怎么校正？样本量不够怎么办？不讲公式推导，只讲怎么用。

配套工具推荐

🧰 开发之余的小工具推荐

在调试 RAG 系统时，经常需要检查文档切分效果、对比 Embedding 相似度、分析检索结果的 JSON 结构：

JSON 格式化：zz365.top 的 JSON 工具可以将检索结果展开，直观查看每个 Chunk 的元数据和得分

文本差异对比：对比不同切分策略的效果，看哪种子串保留了更完整的语义

Base64 编解码：处理某些 Embedding 模型返回的 Base64 编码向量

UUID 生成器：为每个文档和 Chunk 分配唯一 ID