压力测试与性能优化

package benchmark

import (
    "testing"
    "math/rand"
)

func BenchmarkMatchingEngine(b *testing.B) {
    engine := NewMatchingEngine("BTCUSDT")

    // 预先填充订单簿
    for i := 0; i < 1000; i++ {
        engine.AddOrder(&Order{
            OrderID:  fmt.Sprintf("buy_%d", i),
            Side:     "buy",
            Price:    49000 + rand.Float64()*1000,
            Quantity: 0.01,
        })

        engine.AddOrder(&Order{
            OrderID:  fmt.Sprintf("sell_%d", i),
            Side:     "sell",
            Price:    50000 + rand.Float64()*1000,
            Quantity: 0.01,
        })
    }

    b.ResetTimer()

    // 测试性能
    for i := 0; i < b.N; i++ {
        order := &Order{
            OrderID:  fmt.Sprintf("test_%d", i),
            Side:     []string{"buy", "sell"}[i%2],
            Price:    49500 + rand.Float64()*1000,
            Quantity: 0.01,
        }

        engine.AddOrder(order)
    }
}

// 运行：go test -bench=. -benchtime=10s
// 结果示例：
// BenchmarkMatchingEngine-8    1000000    1234 ns/op
// 说明：每秒可处理约 810,000 笔订单

func TestMatchingEngineConcurrency(t *testing.T) {
    engine := NewMatchingEngine("BTCUSDT")

    // 并发数
    concurrency := 100

    // 每个协程发送的订单数
    ordersPerRoutine := 10000

    var wg sync.WaitGroup
    start := time.Now()

    for i := 0; i < concurrency; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()

            for j := 0; j < ordersPerRoutine; j++ {
                order := &Order{
                    OrderID:  fmt.Sprintf("%d_%d", id, j),
                    Side:     []string{"buy", "sell"}[j%2],
                    Price:    49500 + rand.Float64()*1000,
                    Quantity: 0.01,
                }

                engine.AddOrder(order)
            }
        }(i)
    }

    wg.Wait()
    elapsed := time.Since(start)

    totalOrders := concurrency * ordersPerRoutine
    tps := float64(totalOrders) / elapsed.Seconds()

    fmt.Printf("Total orders: %d\n", totalOrders)
    fmt.Printf("Time elapsed: %v\n", elapsed)
    fmt.Printf("TPS: %.2f\n", tps)

    // 示例输出：
    // Total orders: 1000000
    // Time elapsed: 5.2s
    // TPS: 192307.69
}

import "database/sql"

func InitDB(dsn string) (*sql.DB, error) {
    db, err := sql.Open("mysql", dsn)
    if err != nil {
        return nil, err
    }

    // 最大打开连接数
    db.SetMaxOpenConns(100)

    // 最大空闲连接数
    db.SetMaxIdleConns(20)

    // 连接最大生命周期
    db.SetConnMaxLifetime(time.Hour)

    // 连接最大空闲时间
    db.SetConnMaxIdleTime(10 * time.Minute)

    return db, nil
}

// 监控连接池状态
func MonitorDBStats(db *sql.DB) {
    ticker := time.NewTicker(10 * time.Second)
    defer ticker.Stop()

    for range ticker.C {
        stats := db.Stats()

        log.Printf("DB Stats: Open=%d, InUse=%d, Idle=%d, Waiting=%d",
            stats.OpenConnections,
            stats.InUse,
            stats.Idle,
            stats.WaitCount,
        )

        // 如果等待连接数过多，需要增加MaxOpenConns
        if stats.WaitCount > 100 {
            log.Println("Warning: Too many connections waiting")
        }
    }
}

// 慢：逐条插入
func InsertTradesSlow(trades []*Trade) error {
    for _, trade := range trades {
        _, err := db.Exec(`
            INSERT INTO trades (trade_id, symbol, price, quantity, created_at)
            VALUES (?, ?, ?, ?, ?)
        `, trade.TradeID, trade.Symbol, trade.Price, trade.Quantity, trade.CreatedAt)

        if err != nil {
            return err
        }
    }
    return nil
}

// 快：批量插入
func InsertTradesFast(trades []*Trade) error {
    if len(trades) == 0 {
        return nil
    }

    // 构建批量插入SQL
    valueStrings := make([]string, 0, len(trades))
    valueArgs := make([]interface{}, 0, len(trades)*5)

    for _, trade := range trades {
        valueStrings = append(valueStrings, "(?, ?, ?, ?, ?)")
        valueArgs = append(valueArgs, trade.TradeID, trade.Symbol, trade.Price,
            trade.Quantity, trade.CreatedAt)
    }

    query := fmt.Sprintf(`
        INSERT INTO trades (trade_id, symbol, price, quantity, created_at)
        VALUES %s
    `, strings.Join(valueStrings, ","))

    _, err := db.Exec(query, valueArgs...)
    return err
}

// 性能对比：
// 插入10000条记录
// InsertTradesSlow: 8.2s
// InsertTradesFast: 0.3s  (快27倍)

// 慢：逐个SET
func SetTickersSlow(tickers map[string]*Ticker) error {
    for symbol, ticker := range tickers {
        key := fmt.Sprintf("ticker:%s", symbol)
        data, _ := json.Marshal(ticker)
        err := redisClient.Set(ctx, key, data, 5*time.Second).Err()
        if err != nil {
            return err
        }
    }
    return nil
}

// 快：Pipeline批量操作
func SetTickersFast(tickers map[string]*Ticker) error {
    pipe := redisClient.Pipeline()

    for symbol, ticker := range tickers {
        key := fmt.Sprintf("ticker:%s", symbol)
        data, _ := json.Marshal(ticker)
        pipe.Set(ctx, key, data, 5*time.Second)
    }

    _, err := pipe.Exec(ctx)
    return err
}

// 性能对比：
// 更新1000个ticker
// SetTickersSlow: 1.2s (往返1000次)
// SetTickersFast: 0.05s (往返1次，快24倍)

// 使用Lua脚本实现原子操作
var updateTickerScript = redis.NewScript(`
    local key = KEYS[1]
    local lastPrice = tonumber(ARGV[1])
    local volume = tonumber(ARGV[2])

    redis.call('HSET', key, 'lastPrice', lastPrice)
    redis.call('HINCRBY', key, 'volume', volume)

    local high = redis.call('HGET', key, 'highPrice')
    if not high or lastPrice > tonumber(high) then
        redis.call('HSET', key, 'highPrice', lastPrice)
    end

    local low = redis.call('HGET', key, 'lowPrice')
    if not low or lastPrice < tonumber(low) then
        redis.call('HSET', key, 'lowPrice', lastPrice)
    end

    return redis.call('HGETALL', key)
`)

func UpdateTicker(symbol string, lastPrice, volume float64) error {
    key := fmt.Sprintf("ticker:%s", symbol)

    _, err := updateTickerScript.Run(ctx, redisClient,
        []string{key},
        lastPrice, volume,
    ).Result()

    return err
}

type CacheWarmer struct {
    db    *sql.DB
    redis *redis.Client
}

func (cw *CacheWarmer) WarmUpTickers() error {
    // 从数据库加载所有ticker
    rows, err := cw.db.Query(`
        SELECT symbol, last_price, open_price, high_price, low_price, volume
        FROM tickers
    `)
    if err != nil {
        return err
    }
    defer rows.Close()

    pipe := cw.redis.Pipeline()

    for rows.Next() {
        var ticker Ticker
        rows.Scan(&ticker.Symbol, &ticker.LastPrice, &ticker.OpenPrice,
            &ticker.HighPrice, &ticker.LowPrice, &ticker.Volume)

        key := fmt.Sprintf("ticker:%s", ticker.Symbol)
        data, _ := json.Marshal(ticker)

        pipe.Set(context.Background(), key, data, 24*time.Hour)
    }

    _, err = pipe.Exec(context.Background())
    return err
}

// 在服务启动时预热缓存
func main() {
    warmer := &CacheWarmer{db: db, redis: redisClient}

    log.Println("Warming up cache...")
    err := warmer.WarmUpTickers()
    if err != nil {
        log.Fatalf("Cache warmup failed: %v", err)
    }
    log.Println("Cache warmup completed")

    // 启动服务
    // ...
}

import "github.com/gorilla/handlers"

func main() {
    router := mux.NewRouter()

    // 启用Gzip压缩
    handler := handlers.CompressHandler(router)

    http.ListenAndServe(":8080", handler)
}

// 效果：
// 未压缩: 50KB
// Gzip压缩: 8KB (减少84%)

// HTTP客户端配置
var httpClient = &http.Client{
    Transport: &http.Transport{
        MaxIdleConns:        100,
        MaxIdleConnsPerHost: 10,
        IdleConnTimeout:     90 * time.Second,
        DisableKeepAlives:   false,
    },
    Timeout: 10 * time.Second,
}

// 使用单例client，避免每次请求创建新连接
func FetchExternalData(url string) ([]byte, error) {
    resp, err := httpClient.Get(url)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()

    return ioutil.ReadAll(resp.Body)
}

import "golang.org/x/time/rate"

type RateLimitedHandler struct {
    limiter *rate.Limiter
    handler http.Handler
}

func (rl *RateLimitedHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    if !rl.limiter.Allow() {
        http.Error(w, "Rate limit exceeded", http.StatusTooManyRequests)
        return
    }

    rl.handler.ServeHTTP(w, r)
}

func main() {
    router := mux.NewRouter()

    // 限制：每秒1000个请求
    limiter := rate.NewLimiter(rate.Limit(1000), 2000)

    handler := &RateLimitedHandler{
        limiter: limiter,
        handler: router,
    }

    http.ListenAndServe(":8080", handler)
}

import _ "net/http/pprof"

func main() {
    // 启动pprof服务器
    go func() {
        log.Println(http.ListenAndServe("localhost:6060", nil))
    }()

    // 启动应用
    // ...
}

import "runtime/trace"

func main() {
    f, _ := os.Create("trace.out")
    defer f.Close()

    trace.Start(f)
    defer trace.Stop()

    // 运行代码
    benchmarkMatchingEngine()
}

// 分析trace
// go tool trace trace.out

func BenchmarkOrderBook(b *testing.B) {
    ob := NewOrderBook()

    // 预热
    for i := 0; i < 1000; i++ {
        ob.AddOrder(&Order{
            OrderID:  fmt.Sprintf("warmup_%d", i),
            Side:     "buy",
            Price:    50000,
            Quantity: 0.01,
        })
    }

    // 重置计时器
    b.ResetTimer()

    // 报告内存分配
    b.ReportAllocs()

    // 并发测试
    b.RunParallel(func(pb *testing.PB) {
        i := 0
        for pb.Next() {
            ob.AddOrder(&Order{
                OrderID:  fmt.Sprintf("bench_%d", i),
                Side:     "buy",
                Price:    50000 + float64(i%1000),
                Quantity: 0.01,
            })
            i++
        }
    })
}

// 运行：
// go test -bench=. -benchmem -cpuprofile=cpu.prof -memprofile=mem.prof

// 结果示例：
// BenchmarkOrderBook-8    5000000    287 ns/op    128 B/op    2 allocs/op