perf: 亲和表分片、记账按驱动并发、访问日志与原始体控量
继续上一轮性能改造的剩余项： - 会话亲和表分 16 片（原来一个全局 mutex，每个请求都要走一遍）， 键改为内联 FNV-1a 的 64 位摘要（原先是拼接字符串 + hex，每请求两次分配）， 请求路径上零分配；新增键稳定性/无分配/分片并发测试。 - 记账落库按驱动自适应并发：Recorder 按用户分片到多个 worker（同一用户恒定 同片，流水顺序不变）。SQLite 只有一个写者，实测 4 worker 会互相抢锁把 p99 拖到 19–34ms，所以 SQLite 固定 1 个、PostgreSQL 用 4 个。 新增多分片落库与顺序测试（用显式 worker 数在 SQLite 上验证语义）。 - 访问日志：每请求一次 log.Printf 直写 stderr，高 QPS 下是系统调用热点； 改为 64 KiB 缓冲 + 每秒 flush（写/flush 互斥，-race 已验）。 - 原始体控量：LogRaw 落库的请求体截到 256 KiB、响应体截到 1 MiB 并标注原始 体积，避免单条日志几十 MB 拖慢 usage_logs 查询与备份。 - 渠道加权随机选路从 crypto/rand + big.Int 换成 math/rand/v2 的 IntN， 去掉每请求的系统调用与两次分配。 跳过的两项与原因： - 「PG 下把 model 目录查询合并成一次 JOIN」：目录已缓存 10s，冷路径每模型 10 秒才走一次，合并收益接近 0。 - 「每流一个 heartbeat goroutine」：Go 的 ticker 成本很低，千级并发流仍在 可接受范围，改动收益不抵风险。 实测（16 并发、mock 上游、SQLite）：p50 679µs→558µs、p99 2.6ms→2.2ms、 RPS 18.7k→23.9k。验证：go vet、go test ./...、-race（proxy/channel/usage）全绿。
This commit is contained in:
Sakurasan committed
1 parent
8c12207478
commit
f1181150aa
11 files changed
+401
-69
No files matched your search
@@ -13,35 +13,76 @@ import (
|
||||
)
|
||||
|
||||
// Recorder 异步记账器：缓冲队列 + 批量事务落库。
|
||||
//
|
||||
// 多个 worker 按用户分片：同一用户的记录始终交给同一个 worker，落库顺序与扣款
|
||||
// 流水顺序保持一致；不同用户并行落库，PostgreSQL 上不再被单 goroutine 串行化。
|
||||
type Recorder struct {
|
||||
db *gorm.DB
|
||||
ch chan *store.UsageLog
|
||||
wg sync.WaitGroup
|
||||
closed chan struct{}
|
||||
once sync.Once // 保证 Close 只执行一次（main 的 defer Shutdown 与显式 Close 双调用）
|
||||
shards []chan *store.UsageLog
|
||||
// workers 本实例的落库并发度（SQLite 为 1，PG 为 flushWorkersPG）
|
||||
workers int
|
||||
wg sync.WaitGroup
|
||||
closed chan struct{}
|
||||
once sync.Once // 保证 Close 只执行一次（main 的 defer Shutdown 与显式 Close 双调用）
|
||||
|
||||
// mu 保护 stopped：Close 之后 Record 不再进队（ch 永不 close，杜绝 send-on-closed panic），
|
||||
// mu 保护 stopped：Close 之后 Record 不再进队（shard 永不 close，杜绝 send-on-closed panic），
|
||||
// 改走同步落库兜底。
|
||||
mu sync.RWMutex
|
||||
stopped bool
|
||||
}
|
||||
|
||||
const batchSize = 32
|
||||
const (
|
||||
batchSize = 32
|
||||
// flushWorkersPG 落库 worker 数（按用户分片）。SQLite 只有一个写者，
|
||||
// 多 worker 并发写只会互相抢锁、把 p99 拖长（实测 4 worker 时 p99 19–34ms），
|
||||
// 所以按驱动自适应：PG 用多个，SQLite 只用一个。
|
||||
flushWorkersPG = 4
|
||||
)
|
||||
|
||||
// flushRetries 批量落库的瞬时错误重试次数（SQLite busy / PG 连接抖动）。
|
||||
const flushRetries = 3
|
||||
|
||||
func NewRecorder(db *gorm.DB) *Recorder {
|
||||
return newRecorder(db, workersFor(db))
|
||||
}
|
||||
|
||||
// workersFor 按驱动决定落库并发度：SQLite 只有一个写者，多 worker 只会抢锁。
|
||||
func workersFor(db *gorm.DB) int {
|
||||
if db != nil && db.Dialector != nil && db.Dialector.Name() == "postgres" {
|
||||
return flushWorkersPG
|
||||
}
|
||||
return 1
|
||||
}
|
||||
|
||||
// newRecorder 指定并发度构造（测试用：在 SQLite 上也能验证多分片语义）。
|
||||
func newRecorder(db *gorm.DB, workers int) *Recorder {
|
||||
if workers < 1 {
|
||||
workers = 1
|
||||
}
|
||||
r := &Recorder{
|
||||
db: db,
|
||||
ch: make(chan *store.UsageLog, 512),
|
||||
closed: make(chan struct{}),
|
||||
db: db,
|
||||
closed: make(chan struct{}),
|
||||
workers: workers,
|
||||
shards: make([]chan *store.UsageLog, workers),
|
||||
}
|
||||
for i := range r.shards {
|
||||
// 每片容量与原来的单队列等效（512 总量按片均分，至少 64）
|
||||
size := 512 / workers
|
||||
if size < 64 {
|
||||
size = 64
|
||||
}
|
||||
r.shards[i] = make(chan *store.UsageLog, size)
|
||||
r.wg.Add(1)
|
||||
go r.run(i)
|
||||
}
|
||||
r.wg.Add(1)
|
||||
go r.run()
|
||||
return r
|
||||
}
|
||||
|
||||
// shardOf 用户 → 队列分片：同一用户恒定落在同一片，保证其记账顺序。
|
||||
func (r *Recorder) shardOf(userID uint64) chan *store.UsageLog {
|
||||
return r.shards[userID%uint64(len(r.shards))]
|
||||
}
|
||||
|
||||
// Record 提交一条用量（非阻塞；队列满或已停止接收时同步写入，保证不丢账）。
|
||||
func (r *Recorder) Record(l *store.UsageLog) {
|
||||
r.mu.RLock()
|
||||
@@ -50,8 +91,9 @@ func (r *Recorder) Record(l *store.UsageLog) {
|
||||
r.flushRetry([]*store.UsageLog{l})
|
||||
return
|
||||
}
|
||||
ch := r.shardOf(l.UserID)
|
||||
select {
|
||||
case r.ch <- l:
|
||||
case ch <- l:
|
||||
r.mu.RUnlock()
|
||||
default:
|
||||
r.mu.RUnlock()
|
||||
@@ -71,8 +113,9 @@ func (r *Recorder) Close() {
|
||||
})
|
||||
}
|
||||
|
||||
func (r *Recorder) run() {
|
||||
func (r *Recorder) run(shard int) {
|
||||
defer r.wg.Done()
|
||||
ch := r.shards[shard]
|
||||
buf := make([]*store.UsageLog, 0, batchSize)
|
||||
flushBuf := func() {
|
||||
if len(buf) > 0 {
|
||||
@@ -84,7 +127,7 @@ func (r *Recorder) run() {
|
||||
defer tick.Stop()
|
||||
for {
|
||||
select {
|
||||
case l := <-r.ch:
|
||||
case l := <-ch:
|
||||
buf = append(buf, l)
|
||||
if len(buf) >= batchSize {
|
||||
flushBuf()
|
||||
@@ -94,7 +137,7 @@ func (r *Recorder) run() {
|
||||
// Record 已拿到 RLock 还没写进 channel，这里等它落地再收尾。
|
||||
for {
|
||||
select {
|
||||
case l := <-r.ch:
|
||||
case l := <-ch:
|
||||
buf = append(buf, l)
|
||||
if len(buf) >= batchSize {
|
||||
flushBuf()
|
||||
|
||||
@@ -218,3 +218,68 @@ func TestBillingAggregatesBatch(t *testing.T) {
|
||||
t.Fatalf("最后一条流水 balance_after = %v，实际余额 %v", last.BalanceAfter, user.Balance)
|
||||
}
|
||||
}
|
||||
|
||||
// 多 worker 落库：同一用户恒定落在同一分片（流水顺序与提交顺序一致），
|
||||
// 不同用户并行落库，最终一条不丢。
|
||||
func TestRecorderShardedByUser(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
var users []uint64
|
||||
for i := 0; i < 3; i++ {
|
||||
u := store.User{Username: fmt.Sprintf("shard-u%d", i), Email: fmt.Sprintf("shard%d@x.com", i),
|
||||
PasswordHash: "x", Role: store.RoleUser, Balance: 100, Status: store.UserStatusActive}
|
||||
if err := db.Create(&u).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
users = append(users, u.ID)
|
||||
}
|
||||
// 显式用 3 个分片：SQLite 生产配置是单 worker，但分片语义本身仍要验证
|
||||
rec := newRecorder(db, 3)
|
||||
const per = 40
|
||||
for i := 0; i < per; i++ {
|
||||
for _, u := range users {
|
||||
rec.Record(&store.UsageLog{
|
||||
TraceID: fmt.Sprintf("u%d-%d", u, i), UserID: u, ModelName: "m1",
|
||||
Cost: 0.001, Status: store.UsageStatusSuccess, CreatedAt: time.Now().UTC(),
|
||||
})
|
||||
}
|
||||
}
|
||||
rec.Close()
|
||||
|
||||
var n int64
|
||||
db.Model(&store.UsageLog{}).Count(&n)
|
||||
if n != int64(per*len(users)) {
|
||||
t.Fatalf("落库 %d 条，期望 %d（一条不丢）", n, per*len(users))
|
||||
}
|
||||
// 每个用户的分片一致 → 流水按 id 递增的顺序应与 trace 序号一致
|
||||
for _, u := range users {
|
||||
var logs []store.BalanceLog
|
||||
db.Where("user_id = ?", u).Order("id ASC").Find(&logs)
|
||||
if len(logs) != per {
|
||||
t.Fatalf("用户 %d 流水 %d 条，期望 %d", u, len(logs), per)
|
||||
}
|
||||
for i := 1; i < len(logs); i++ {
|
||||
if logs[i].BalanceAfter > logs[i-1].BalanceAfter {
|
||||
t.Fatalf("用户 %d 的流水余额应单调递减（顺序错乱）：%v → %v",
|
||||
u, logs[i-1].BalanceAfter, logs[i].BalanceAfter)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 分片函数：同一用户恒定同片，且覆盖所有分片。
|
||||
func TestShardOfStable(t *testing.T) {
|
||||
rec := NewRecorder(openTestDB(t))
|
||||
defer rec.Close()
|
||||
for _, u := range []uint64{1, 2, 3, 17, 1 << 40} {
|
||||
if rec.shardOf(u) != rec.shardOf(u) {
|
||||
t.Fatalf("用户 %d 的分片不稳定", u)
|
||||
}
|
||||
}
|
||||
seen := map[chan *store.UsageLog]bool{}
|
||||
for u := uint64(0); u < 64; u++ {
|
||||
seen[rec.shardOf(u)] = true
|
||||
}
|
||||
if len(seen) != rec.workers {
|
||||
t.Fatalf("只用到 %d 个分片，期望 %d", len(seen), rec.workers)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user