短链服务进阶：Prometheus 监控 + 批量写入优化 + 布隆过滤器

上一篇文章实现了 Redis 限流和多级缓存。这篇文章记录三个重要改进：Prometheus + Grafana 监控体系、批量写入优化、布隆过滤器防穿透。

第一部分：Prometheus + Grafana 监控

为什么需要监控

之前虽然加了 Prometheus 指标，但从来没看过——因为不知道怎么看。这次把完整的监控体系搭起来。

Docker Compose 配置

在 docker-compose.yml 中添加 Prometheus 和 Grafana：

1
prometheus:
2
image: prom/prometheus:v2.54.0
3
ports:
4
    - "9090:9090"
5
volumes:
6
    - ./prometheus.yml:/etc/prometheus/prometheus.yml
7
    - prometheus_data:/prometheus
8
command:
9
    - '--config.file=/etc/prometheus/prometheus.yml'
10
    - '--storage.tsdb.path=/prometheus'
11
    - '--storage.tsdb.retention.time=15d'
12

13
grafana:
14
image: grafana/grafana:11.0.0
15
ports:
16
    - "3000:3000"
17
environment:
18
    - GF_SECURITY_ADMIN_USER=admin
19
    - GF_SECURITY_ADMIN_PASSWORD=admin
20
volumes:
21
    - grafana_data:/var/lib/grafana
22
    - ./grafana/provisioning:/etc/grafana/provisioning
23
depends_on:
24
    - prometheus

Prometheus 配置

创建 prometheus.yml：

1
global:
2
  scrape_interval: 15s
3

4
scrape_configs:
5
  - job_name: 'prometheus'
6
    static_configs:
7
      - targets: ['localhost:9090']
8

9
  - job_name: 'shortlink'
10
    static_configs:
11
      # Windows Docker Desktop 用 host.docker.internal 访问宿主机
12
      - targets: ['host.docker.internal:6060']

Grafana 自动配置数据源

创建 grafana/provisioning/datasources/datasources.yml：

1
apiVersion: 1
2
datasources:
3
  - name: Prometheus
4
    type: prometheus
5
    access: proxy
6
    url: http://prometheus:9090
7
    isDefault: true

添加业务指标

除了基础的 HTTP 指标，还需要业务指标来了解系统运行状况。

在 internal/platform/metrics/metrics.go 中添加：

1
// 缓存指标
2
CacheOperations = prometheus.NewCounterVec(
3
    prometheus.CounterOpts{
4
        Name: "shortlink_cache_operations_total",
5
        Help: "缓存操作计数",
6
    },
7
    []string{"level", "result"},  // level: l1/l2, result: hit/miss/hit_negative
8
)
9

10
// 短链业务指标
11
ShortlinkCreated = prometheus.NewCounter(
12
    prometheus.CounterOpts{
13
        Name: "shortlink_created_total",
14
        Help: "创建的短链总数",
15
    },
16
)
17

18
ShortlinkRedirects = prometheus.NewCounter(
19
    prometheus.CounterOpts{
20
        Name: "shortlink_redirects_total",
21
        Help: "短链跳转总数",
22
    },
23
)
24

25
// 数据库指标
26
DBQueryDuration = prometheus.NewHistogramVec(
27
    prometheus.HistogramOpts{
28
        Name:    "shortlink_db_query_duration_seconds",
29
        Help:    "数据库查询耗时分布",
30
        Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1},
31
    },
32
    []string{"operation"},
33
)

在代码中埋点

缓存层埋点（cache/shortlink.go）：

1
func (c *ShortlinkCache) Get(ctx context.Context, code string) (string, error) {
2
    // L1: 本地缓存
3
    if c.local != nil {
4
        if url, ok := c.local.Get(code); ok {
5
            if url == notFoundSentinel {
6
                metrics.CacheOperations.WithLabelValues("l1", "hit_negative").Inc()
7
            } else {
8
                metrics.CacheOperations.WithLabelValues("l1", "hit").Inc()
9
            }
10
            return url, nil
11
        }
12
        metrics.CacheOperations.WithLabelValues("l1", "miss").Inc()
13
    }
14
    // L2 Redis 类似...
15
}

Dashboard 效果

配置完成后，Grafana Dashboard 可以展示：

面板	PromQL
总 QPS	`sum(rate(http_request_total[1m]))`
5xx 错误率	`sum(rate(http_request_total{status=~"5.."}[5m])) / sum(rate(http_request_total[5m]))`
P95 延迟	`histogram_quantile(0.95, sum by (le) (rate(http_request_duration_seconds_bucket[5m])))`
缓存命中率	`sum(shortlink_cache_operations_total{result="hit"}) / sum(shortlink_cache_operations_total)`

第二部分：批量写入优化

问题分析

之前的点击统计写入代码：

1
func (c *Consumer) flush(batch []ClickEvent) {
2
    for _, e := range batch {
3
        tx.Exec(ctx, `INSERT INTO click_stats ...`)  // 100 次
4
        tx.Exec(ctx, `UPDATE shortlinks SET click_count = click_count + 1 ...`)  // 又 100 次
5
    }
6
}

100 条数据 = 200 次 SQL 执行，每次都有网络往返延迟。

优化方案：CopyFrom + unnest

使用 PostgreSQL 的 COPY 协议批量插入，用 unnest 聚合更新：

1
func (c *Consumer) flush(batch []ClickEvent) {
2
    if len(batch) == 0 {
3
        return
4
    }
5

6
    start := time.Now()
7
    defer func() {
8
        metrics.StatsFlushDuration.Observe(time.Since(start).Seconds())
9
        metrics.StatsFlushSize.Observe(float64(len(batch)))
10
    }()
11

12
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
13
    defer cancel()
14

15
    tx, err := c.db.Begin(ctx)
16
    if err != nil {
17
        slog.Error("click stats: begin tx failed", "err", err)
18
        return
19
    }
20
    defer tx.Rollback(context.Background())
21

22
    // 1. 使用 CopyFrom 批量插入
23
    rows := make([][]any, len(batch))
24
    for i, e := range batch {
25
        rows[i] = []any{e.Code, e.ClickedAt, e.IP, e.UserAgent, e.Referer}
26
    }
27

28
    _, err = tx.CopyFrom(ctx,
29
        pgx.Identifier{"click_stats"},
30
        []string{"code", "clicked_at", "ip", "user_agent", "referer"},
31
        pgx.CopyFromRows(rows),
32
    )
33
    if err != nil {
34
        slog.Error("click stats: copy failed", "err", err)
35
        return
36
    }
37

38
    // 2. 聚合统计每个 code 的点击数
39
    counts := make(map[string]int)
40
    for _, e := range batch {
41
        counts[e.Code]++
42
    }
43

44
    // 3. 批量更新（使用 unnest）
45
    codes := make([]string, 0, len(counts))
46
    deltas := make([]int, 0, len(counts))
47
    for code, delta := range counts {
48
        codes = append(codes, code)
49
        deltas = append(deltas, delta)
50
    }
51

52
    _, err = tx.Exec(ctx, `
53
        UPDATE shortlinks s
54
        SET click_count = s.click_count + v.delta,
55
            updated_at = now()
56
        FROM unnest($1::text[], $2::int[]) AS v(code, delta)
57
        WHERE s.code = v.code
58
    `, codes, deltas)
59
    if err != nil {
60
        slog.Error("click stats: batch update failed", "err", err)
61
        return
62
    }
63

64
    if err := tx.Commit(ctx); err != nil {
65
        slog.Error("click stats: commit failed", "err", err)
66
    }
67
}

性能对比

添加了 flush 耗时指标后，实测结果：

1
shortlink_stats_flush_duration_seconds_sum 0.0458
2
shortlink_stats_flush_duration_seconds_count 5

5 次 flush，总耗时 0.0458 秒，平均每次 ~9ms。

指标	优化前（估算）	优化后（实测）
100 条数据 SQL 次数	200 次	2 次
100 条数据耗时	~100-200ms	~9ms
提升倍数	-	10-20x

关键点

CopyFrom：使用 PostgreSQL COPY 协议，一次网络往返插入所有数据
聚合 counts：如果 100 条记录中有 80 条是同一个 code，只需要 UPDATE 一次（+80）
unnest：把数组展开成虚拟表，一条 SQL 更新所有 code

第三部分：布隆过滤器防穿透

问题

虽然已经有负缓存，但恶意请求大量不存在的短码时，每个新短码都要查一次 DB 才能写负缓存。

布隆过滤器原理

返回 false → 一定不存在，直接返回
返回 true → 可能存在，继续查缓存和 DB

优势：内存占用小（100 万短码约 1.2MB），查询 O(1)。

实现

创建 internal/app/shortlink/cache/bloom.go：

1
package cache
2

3
import (
4
    "sync"
5
    "github.com/bits-and-blooms/bloom/v3"
6
)
7

8
type BloomFilter struct {
9
    filter *bloom.BloomFilter
10
    mu     sync.RWMutex
11
}
12

13
func NewBloomFilter(expectedItems uint, falsePositiveRate float64) *BloomFilter {
14
    return &BloomFilter{
15
        filter: bloom.NewWithEstimates(expectedItems, falsePositiveRate),
16
    }
17
}
18

19
func (b *BloomFilter) Add(code string) {
20
    b.mu.Lock()
21
    defer b.mu.Unlock()
22
    b.filter.AddString(code)
23
}
24

25
func (b *BloomFilter) MightExist(code string) bool {
26
    b.mu.RLock()
27
    defer b.mu.RUnlock()
28
    return b.filter.TestString(code)
29
}

集成到 Repo

在 NewShortlinksRepo 中初始化：

1
func NewShortlinksRepo(db *pgxpool.Pool, cache *cache.ShortlinkCache, bloom *cache.BloomFilter) *ShortlinksRepo {
2
    repo := &ShortlinksRepo{db: db, cache: cache, bloom: bloom}
3

4
    // 初始化布隆过滤器
5
    if bloom != nil {
6
        ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
7
        defer cancel()
8

9
        rows, err := db.Query(ctx, "SELECT code FROM shortlinks WHERE code IS NOT NULL")
10
        if err != nil {
11
            slog.Error("bloom filter: load codes failed", "err", err)
12
            return repo
13
        }
14
        defer rows.Close()
15

16
        count := 0
17
        for rows.Next() {
18
            var code string
19
            if err := rows.Scan(&code); err != nil {
20
                continue
21
            }
22
            bloom.Add(code)
23
            count++
24
        }
25
        slog.Info("bloom filter initialized", "count", count)
26
    }
27

28
    return repo
29
}

在 Resolve 方法开头检查：

1
func (s *ShortlinksRepo) Resolve(ctx context.Context, code string) string {
2
    // 布隆过滤器快速判断
3
    if s.bloom != nil && !s.bloom.MightExist(code) {
4
        return ""  // 一定不存在
5
    }
6
    // 继续查缓存和 DB...
7
}

创建短链时添加到布隆过滤器：

1
// Create 方法中，commit 成功后
2
if s.bloom != nil && code != "" {
3
    s.bloom.Add(code)
4
}

效果验证

发送 100 个不存在的短码请求：

1
for i in $(seq 1 100); do curl -s -o /dev/null http://localhost:9999/fake_code_$i; done

检查缓存指标：

1
shortlink_cache_operations_total{level="l2",result="hit"} 1

缓存操作数没有增加，说明 100 个请求全部被布隆过滤器拦截，没有穿透到缓存和数据库。

总结

这次改进的三个功能：

功能	效果
Prometheus + Grafana	可视化监控 QPS、延迟、缓存命中率
批量写入优化	性能提升 10-20 倍
布隆过滤器	彻底防止缓存穿透