Go教程(七)后端进阶 —— 缓存、异步任务、定时任务、实时推送、gRPC、监控

Go 07：后端进阶 —— 缓存、异步任务、定时任务、实时推送、gRPC、监控

06 篇的服务已经能用了。要成为"能上线、能扛量、出问题能查"的产品级后端，还需要下面这些能力。
每一节都在 06 篇的记账服务上增加一个功能，并给出 Python 对应方案方便理解。

go get github.com/redis/go-redis/v9 github.com/hibiken/asynq github.com/robfig/cron/v3 \
       golang.org/x/time/rate golang.org/x/sync/singleflight \
       github.com/prometheus/client_golang/prometheus google.golang.org/grpc
docker run -d -p 6379:6379 redis:7     # 本地起一个 Redis

1. 缓存：月度统计加 Redis（Cache-Aside 模式）

统计接口要做 GROUP BY，访问频繁时可以缓存。

// internal/cache/summary.go
package cache

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"time"

	"example.com/accounting-gin/internal/service"
	"github.com/redis/go-redis/v9"
	"golang.org/x/sync/singleflight"
)

type SummaryCache struct {
	RDB   *redis.Client
	Svc   *service.RecordService
	group singleflight.Group
}

func key(uid uint, month string) string { return fmt.Sprintf("summary:%d:%s", uid, month) }

func (c *SummaryCache) Get(ctx context.Context, uid uint, month string) ([]service.CategorySum, error) {
	k := key(uid, month)

	// 1. 先查缓存
	if b, err := c.RDB.Get(ctx, k).Bytes(); err == nil {
		var out []service.CategorySum
		if json.Unmarshal(b, &out) == nil {
			return out, nil
		}
	} else if !errors.Is(err, redis.Nil) {
		slog.Warn("redis unavailable, fallback to db", "err", err) // Redis 挂了不影响主流程（降级）
	}

	// 2. 未命中：singleflight 保证同一个 key 同时只有一个请求查库（防"缓存击穿"）
	v, err, _ := c.group.Do(k, func() (any, error) {
		out, err := c.Svc.Summary(ctx, uid, month)
		if err != nil {
			return nil, err
		}
		if b, err := json.Marshal(out); err == nil {
			c.RDB.Set(ctx, k, b, 10*time.Minute)
		}
		return out, nil
	})
	if err != nil {
		return nil, err
	}
	return v.([]service.CategorySum), nil
}

// Invalidate 在新增/删除记录后调用，保证数据一致
func (c *SummaryCache) Invalidate(ctx context.Context, uid uint, month string) {
	c.RDB.Del(ctx, key(uid, month))
}

// 初始化
rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
summaryCache := &cache.SummaryCache{RDB: rdb, Svc: recordSvc}

QA 测试点：新增记录后统计是否立即更新（缓存失效）；Redis 停掉后接口是否仍可用（降级）。

2. 异步任务：月度报表邮件（asynq ≈ Celery）

耗时操作（发邮件、生成 PDF、调用第三方）不应阻塞 HTTP 请求。

// internal/tasks/tasks.go
package tasks

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"time"

	"github.com/hibiken/asynq"
)

const TypeMonthlyReport = "report:monthly"

type MonthlyReportPayload struct {
	UserID uint   `json:"userId"`
	Email  string `json:"email"`
	Month  string `json:"month"`
}

// NewMonthlyReportTask ≈ Celery 的 task.delay(...) 之前构造参数
func NewMonthlyReportTask(p MonthlyReportPayload) (*asynq.Task, error) {
	b, err := json.Marshal(p)
	if err != nil {
		return nil, err
	}
	return asynq.NewTask(TypeMonthlyReport, b,
		asynq.MaxRetry(3),             // 失败自动重试（指数退避）
		asynq.Timeout(2*time.Minute),
		asynq.Queue("default"),
	), nil
}

// HandleMonthlyReport 在 worker 进程中执行
func HandleMonthlyReport(ctx context.Context, t *asynq.Task) error {
	var p MonthlyReportPayload
	if err := json.Unmarshal(t.Payload(), &p); err != nil {
		return fmt.Errorf("bad payload: %w", asynq.SkipRetry) // 数据错误无需重试
	}
	slog.Info("generating report", "user", p.UserID, "month", p.Month)
	// 1. 查询统计 2. 渲染 HTML 3. 调用邮件服务（如 EAAS 的发信 API）
	return nil
}

// HTTP 侧：投递任务（handler 中）
client := asynq.NewClient(asynq.RedisClientOpt{Addr: "localhost:6379"})
defer client.Close()

task, _ := tasks.NewMonthlyReportTask(tasks.MonthlyReportPayload{UserID: uid, Email: email, Month: "2026-10"})
info, err := client.EnqueueContext(c.Request.Context(), task)
// 立即返回 202 Accepted + 任务 ID，前端可轮询状态
c.JSON(http.StatusAccepted, gin.H{"taskId": info.ID})

// cmd/worker/main.go —— 独立的 worker 进程（≈ celery -A proj worker）
package main

import (
	"log"

	"example.com/accounting-gin/internal/tasks"
	"github.com/hibiken/asynq"
)

func main() {
	srv := asynq.NewServer(
		asynq.RedisClientOpt{Addr: "localhost:6379"},
		asynq.Config{Concurrency: 10}, // 同时处理 10 个任务（10 个 goroutine）
	)
	mux := asynq.NewServeMux()
	mux.HandleFunc(tasks.TypeMonthlyReport, tasks.HandleMonthlyReport)
	if err := srv.Run(mux); err != nil {
		log.Fatal(err)
	}
}

asynq 自带 Web 监控面板 asynqmon（≈ Celery 的 Flower），可查看队列、失败任务、手动重试。

3. 定时任务

3.1 单实例：robfig/cron（≈ APScheduler）

c := cron.New()
c.AddFunc("0 9 1 * *", func() {           // 每月 1 号 9:00
	slog.Info("enqueue monthly reports")
	// 查询所有用户，逐个投递 MonthlyReport 任务
})
c.AddFunc("@every 10m", cleanExpiredSessions)
c.Start()
defer c.Stop()

3.2 多实例部署：避免重复执行

服务部署 3 个副本时，上面的 cron 会执行 3 次。两种解决方式：

// 方式一：Redis 分布式锁，抢到锁的实例执行
ok, err := rdb.SetNX(ctx, "lock:monthly-report:2026-11", hostname, 10*time.Minute).Result()
if err == nil && ok {
	runMonthlyReport()
}

// 方式二：asynq Scheduler 单独部署一个实例，只负责"按时投递任务"，执行交给 worker 集群
scheduler := asynq.NewScheduler(asynq.RedisClientOpt{Addr: "localhost:6379"}, nil)
task := asynq.NewTask("report:all-users", nil)
scheduler.Register("0 9 1 * *", task)
scheduler.Run()

4. 限流：保护登录接口（防暴力破解）

// internal/middleware/ratelimit.go
package middleware

import (
	"net/http"
	"sync"

	"github.com/gin-gonic/gin"
	"golang.org/x/time/rate"
)

// RateLimit 每个 IP 一个令牌桶：每秒补充 r 个令牌，最多积攒 burst 个
func RateLimit(r rate.Limit, burst int) gin.HandlerFunc {
	var mu sync.Mutex
	limiters := map[string]*rate.Limiter{}

	return func(c *gin.Context) {
		ip := c.ClientIP()
		mu.Lock()
		l, ok := limiters[ip]
		if !ok {
			l = rate.NewLimiter(r, burst)
			limiters[ip] = l
		}
		mu.Unlock()

		if !l.Allow() {
			c.AbortWithStatusJSON(http.StatusTooManyRequests, gin.H{"error": "too many requests"})
			return
		}
		c.Next()
	}
}

// 用法：登录每 IP 每分钟最多约 5 次
// api.POST("/auth/login", middleware.RateLimit(rate.Every(12*time.Second), 5), ah.Login)

生产中 map 需要定期清理过期 IP；多实例时改用 Redis 实现（如 go-redis/redis_rate）。

5. 实时推送：SSE（多设备同步新增记录）

手机上记了一笔，电脑网页实时出现。SSE 比 WebSocket 简单（单向、基于 HTTP），非常适合"服务端通知"。
这正是 03 篇 goroutine + channel 的实战：

// internal/realtime/broker.go
package realtime

import "sync"

type Broker struct {
	mu   sync.Mutex
	subs map[uint]map[chan string]struct{} // userID -> 该用户的所有连接
}

func NewBroker() *Broker { return &Broker{subs: map[uint]map[chan string]struct{}{}} }

func (b *Broker) Subscribe(uid uint) chan string {
	ch := make(chan string, 8)
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.subs[uid] == nil {
		b.subs[uid] = map[chan string]struct{}{}
	}
	b.subs[uid][ch] = struct{}{}
	return ch
}

func (b *Broker) Unsubscribe(uid uint, ch chan string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	delete(b.subs[uid], ch)
	close(ch)
}

func (b *Broker) Publish(uid uint, msg string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	for ch := range b.subs[uid] {
		select {
		case ch <- msg:
		default: // 客户端太慢，丢弃，不阻塞发布者
		}
	}
}

// handler：GET /api/events
func (h *EventsHandler) Stream(c *gin.Context) {
	uid := c.GetUint(middleware.UserIDKey)
	ch := h.Broker.Subscribe(uid)
	defer h.Broker.Unsubscribe(uid, ch)

	c.Stream(func(w io.Writer) bool {
		select {
		case msg := <-ch:
			c.SSEvent("record-created", msg)
			return true // 继续保持连接
		case <-c.Request.Context().Done():
			return false // 客户端断开
		}
	})
}

// 在 RecordHandler.Create 成功后：
// b, _ := json.Marshal(r); h.Broker.Publish(r.UserID, string(b))

// 前端（TS）：浏览器原生 EventSource
const es = new EventSource("/api/events", { withCredentials: true });
es.addEventListener("record-created", (e) => {
  const record = JSON.parse((e as MessageEvent).data);
  records.value.unshift(record);
});

注意：EventSource 不能自定义请求头，鉴权需用 Cookie 或 URL 参数中的短期 token。
多实例部署时，Broker 需换成 Redis Pub/Sub 才能跨实例推送。

6. gRPC：服务间调用

当系统拆分为多个服务（例如"报表服务"单独部署），内部调用常用 gRPC：强类型、高性能、自动生成多语言客户端。
它和 C# 的 WCF / ASP.NET gRPC 思路一致。

// proto/report/v1/report.proto
syntax = "proto3";
package report.v1;
option go_package = "example.com/accounting-gin/proto/report/v1;reportv1";

service ReportService {
  rpc MonthlySummary(MonthlySummaryRequest) returns (MonthlySummaryResponse);
}

message MonthlySummaryRequest {
  uint64 user_id = 1;
  string month = 2;
}

message CategoryTotal {
  string type = 1;
  string category = 2;
  int64 total_cents = 3;
}

message MonthlySummaryResponse {
  repeated CategoryTotal items = 1;
}

brew install protobuf
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest
protoc --go_out=. --go_opt=paths=source_relative \
       --go-grpc_out=. --go-grpc_opt=paths=source_relative \
       proto/report/v1/report.proto

// 服务端
type reportServer struct {
	reportv1.UnimplementedReportServiceServer
	svc *service.RecordService
}

func (s *reportServer) MonthlySummary(ctx context.Context, req *reportv1.MonthlySummaryRequest) (*reportv1.MonthlySummaryResponse, error) {
	if req.GetMonth() == "" {
		return nil, status.Error(codes.InvalidArgument, "month is required")
	}
	sums, err := s.svc.Summary(ctx, uint(req.GetUserId()), req.GetMonth())
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	resp := &reportv1.MonthlySummaryResponse{}
	for _, x := range sums {
		resp.Items = append(resp.Items, &reportv1.CategoryTotal{
			Type: x.Type, Category: x.Category, TotalCents: x.TotalCents,
		})
	}
	return resp, nil
}

func runGRPC(svc *service.RecordService) error {
	lis, err := net.Listen("tcp", ":9090")
	if err != nil {
		return err
	}
	g := grpc.NewServer()
	reportv1.RegisterReportServiceServer(g, &reportServer{svc: svc})
	return g.Serve(lis)
}

// 客户端（另一个服务中）
conn, err := grpc.NewClient("localhost:9090", grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
	log.Fatal(err)
}
defer conn.Close()

client := reportv1.NewReportServiceClient(conn)
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) // 一定要设超时
defer cancel()
resp, err := client.MonthlySummary(ctx, &reportv1.MonthlySummaryRequest{UserId: 1, Month: "2026-11"})

# QA 调试 gRPC：grpcurl（≈ 给 gRPC 用的 curl）
grpcurl -plaintext -d '{"user_id":1,"month":"2026-11"}' localhost:9090 report.v1.ReportService/MonthlySummary

需要在服务端注册 reflection（reflection.Register(g)）grpcurl 才能自动发现接口。

7. 可观测性：指标 + 性能分析

// internal/middleware/metrics.go
package middleware

import (
	"strconv"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
)

var httpDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
	Name:    "http_request_duration_seconds",
	Help:    "HTTP request latency",
	Buckets: prometheus.DefBuckets,
}, []string{"method", "path", "status"})

func Metrics() gin.HandlerFunc {
	return func(c *gin.Context) {
		start := time.Now()
		c.Next()
		httpDuration.WithLabelValues(
			c.Request.Method, c.FullPath(), strconv.Itoa(c.Writer.Status()),
		).Observe(time.Since(start).Seconds())
	}
}

// 路由中：
// r.Use(middleware.Metrics())
// r.GET("/metrics", gin.WrapH(promhttp.Handler()))   // Prometheus 抓取，Grafana 展示

性能分析（Go 的杀手级内置工具，Python 很难做到这么方便）：

import _ "net/http/pprof"

go func() { _ = http.ListenAndServe("localhost:6060", nil) }() // 仅内网/本机开放

go tool pprof -http=:8081 "http://localhost:6060/debug/pprof/profile?seconds=30"   # CPU 火焰图
go tool pprof -http=:8081 http://localhost:6060/debug/pprof/heap                   # 内存
curl http://localhost:6060/debug/pprof/goroutine?debug=1                           # goroutine 泄漏排查

链路追踪：OpenTelemetry（go.opentelemetry.io/otel）+ Jaeger，可追踪一个请求经过的所有服务。

8. 何时拆分微服务？

不要过早拆分。Go 的优势是单体也能跑得很快、很省资源。

9. 练习

1. 给统计接口加 Redis 缓存，写一个测试：新增记录后统计结果立即变化。
2. 用 asynq 实现"导出 CSV"异步任务：投递 → 返回任务 ID → GET /api/tasks/:id 查询状态 → 下载。
3. 给登录接口加限流，用 Go 写一个并发压测小程序（复用 03 篇的并发池）验证 429。
4. 实现 SSE 推送，打开两个浏览器窗口验证实时同步。
5. 用 hey 或 wrk 压测统计接口，用 pprof 找出最耗时的函数。