diff --git a/CLAUDE.md b/CLAUDE.md index 6d87aa6..7246938 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -44,6 +44,7 @@ - 禁止直接用裸 `go` 启动并行工作负载,一律走 `common` 的池(池清单与默认值见 README 配置说明) - **防死锁**:等待链单向「主 → A池 → B池」,被等待池的任务内不得再等待任何池(会饿死 worker);池无 Wait 方法,等待用调用方 `sync.WaitGroup`,任务结果经 buffered channel 回主 goroutine - **共享状态安全(内存)**:池内任务并发执行,共享实例(service 单例、model 句柄等)只允许**只读**访问;可变字段必须在**提交池之前**由主 goroutine 一次性预置,任务内禁止写共享字段——Go map 并发写直接 `fatal error: concurrent map writes`,无锁、无降级、不可恢复,只能崩溃重启。需要可变共享状态时按优先级:① 无共享(任务内新建、buffered channel 传递结果)② 锁(`sync.Mutex`/`RWMutex`,锁内只做内存操作,LLM/DB 等 IO 放锁外)③ `sync/atomic`(仅限 int 类标量计数/标志,如 `atomic.AddInt64`,并发计数禁用普通 `++`;复合结构不要用 atomic,指针 CAS 属例外) +- **锁的使用**:互斥场景唯一入口是 `common.WithLock[T any](ctx, key, expire, retries, retryInterval, fn func() (T, error)) (T, error)`——泛型回调,业务返回值经 T 原样透出给下游;内部按 config.yml 自动选择锁实现(配置了 `redis` 节点 → redis 锁,跨实例互斥,SET NX EX + token 对比删除防误删他人锁;未配置 → gcache 内存锁,单实例互斥),禁止直接用 gcache/gredis 自己实现加锁。拿不到锁(被占用,`ErrLockHeld`)最多重试 `retries` 次、每次间隔 `retryInterval`(`retries=0` 立即失败;ctx 取消/超时同样终止等待);中间件故障不重试直接返回。锁自动释放:无论 fn 成功、失败还是 panic,defer 释放。`expire` 必须 > 0(进程崩溃兜底不死锁),fn 耗时必须在 expire 前完成,fn 内禁止长耗时 IO(LLM/DB 调用);锁粒度按业务唯一键尽量小 - **裸 `go` 允许的例外**:`go func(){ wg.Wait(); close(ch) }()` 收尾惯用法、SSE 心跳、流式管道(Stream 读写)等长生命周期/非工作负载协程 ## 文档职责（三文档体系） @@ -65,6 +66,7 @@ - **缓存一致性**:DAO 查询走缓存(TTL 来自 `database.cache.ttl`),写操作后必须清对应缓存 - **批处理 SQL**:批量写入用 `InsertAll` 类方法,批量删除用 `IN` 子句,禁止循环单条 INSERT/DELETE - **配置即使用**:config.yml 中出现 redis / mq 等中间件配置时,代码必须实际接入使用,禁止"配置了但代码不用"或"代码写死但配置缺失" +- **消息/回调幂等**:接入 MQ / Webhook 时,消费与回调处理必须幂等——MQ 至少一次语义、webhook 失败重试都可能重复投递同一事件,禁止依赖"只投一次"假设。幂等手段:以业务唯一键(如 `任务ID + 事件类型`)先查重或建唯一约束再落库,重复事件直接忽略;重试与补偿逻辑同样要防重复执行 ## 运维部署规范(硬性要求) diff --git a/common/util.go b/common/util.go index 414827a..d986e74 100644 --- a/common/util.go +++ b/common/util.go @@ -1,11 +1,24 @@ package common import ( + "context" "crypto/rand" "crypto/sha256" "encoding/hex" + "fmt" + "time" + + "github.com/gogf/gf/v2/database/gredis" + "github.com/gogf/gf/v2/errors/gerror" + "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/os/gcache" + + _ "github.com/gogf/gf/contrib/nosql/redis/v2" // 注册 go-redis 适配器（gredis.New 需要），无 redis 配置时闲置 ) +// unlockLuaScript 对比 token 再删除，防止锁过期后误删他人持有的锁 +const unlockLuaScript = `if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end` + // RandomToken 生成 n 字节随机数的十六进制字符串（2n 位） func RandomToken(n int) string { b := make([]byte, n) @@ -20,3 +33,76 @@ func TokenFingerprint(token string) string { sum := sha256.Sum256([]byte(token)) return hex.EncodeToString(sum[:8]) } + +// ErrLockHeld 锁被他人持有。WithLock 据此重试；中间件故障等其它错误不重试，直接返回。 +var ErrLockHeld = gerror.New("lock held by another holder") + +// WithLock 是互斥锁的唯一封装（util.go 仅此一个锁方法，禁止业务代码直接操作 gcache/gredis 自己实现）： +// 获取互斥锁后执行 fn，fn 结束（成功、失败或 panic）时自动释放锁（defer）。 +// 锁实现按 config.yml 自动选择：配置了 redis 节点 → redis 锁（SET NX EX，跨实例互斥）； +// 未配置 → gcache 内存锁（单实例互斥）。 +// +// 拿不到锁（被占用）时最多重试 retries 次（含首次共 retries+1 次尝试），每次间隔 retryInterval； +// retries=0 表示拿不到锁立即失败。ctx 取消/超时同样终止等待。中间件故障不重试，直接返回。 +// +// fn 为泛型回调，返回 (T, error)：T 由闭包返回类型推断，业务返回值原样透出给下游； +// 方法参数/数量无约束，闭包捕获即可。仅需 error 的场景返回 (nil, err)（T 推断为 any）。 +// +// doc, err := common.WithLock(ctx, "task:"+id, 30*time.Second, 3, 200*time.Millisecond, func() (string, error) { +// return s.process(ctx, id, mode) // 返回值透出给下游 +// }) +// +// expire 必须 > 0：锁自动过期兜底（进程崩溃不死锁），fn 耗时必须在 expire 前完成，fn 内禁止长耗时 IO。 +func WithLock[T any](ctx context.Context, key string, expire time.Duration, retries int, retryInterval time.Duration, fn func() (T, error)) (T, error) { + var zero T + if expire <= 0 { + return zero, gerror.Newf("with lock %s: expire must be positive", key) + } + if retries > 0 && retryInterval <= 0 { + return zero, gerror.Newf("with lock %s: retryInterval must be positive when retries > 0", key) + } + for attempt := 0; ; attempt++ { + if err := ctx.Err(); err != nil { + return zero, gerror.Wrapf(err, "with lock %s", key) + } + var unlock func() + if !g.Cfg().MustGet(ctx, "redis", nil).IsNil() { + client := g.Redis() + token := RandomToken(16) + secs := int64(expire / time.Second) + if secs < 1 { + secs = 1 + } + v, err := client.Set(ctx, key, token, gredis.SetOption{TTLOption: gredis.TTLOption{EX: &secs}, NX: true}) + if err != nil { + return zero, gerror.Wrap(err, "redis lock acquire failed") + } + if !v.IsEmpty() { + unlock = func() { + if _, err := client.Eval(ctx, unlockLuaScript, 1, []string{key}, []any{token}); err != nil { + fmt.Printf("redis lock unlock failed (%s): %v\n", key, err) + } + } + } + } else { + ok, err := gcache.SetIfNotExist(ctx, key, 1, expire) + if err != nil { + return zero, err + } + if ok { + unlock = func() { gcache.Remove(ctx, key) } + } + } + if unlock != nil { + defer unlock() + return fn() + } + if attempt >= retries { + return zero, ErrLockHeld + } + select { + case <-ctx.Done(): + case <-time.After(retryInterval): + } + } +} diff --git a/data/business.db b/data/business.db index 828a3b6..77a9571 100644 Binary files a/data/business.db and b/data/business.db differ diff --git a/go.mod b/go.mod index ef0afcd..569c241 100644 --- a/go.mod +++ b/go.mod @@ -23,9 +23,11 @@ require ( github.com/bytedance/gopkg v0.1.3 // indirect github.com/bytedance/sonic v1.15.0 // indirect github.com/bytedance/sonic/loader v0.5.0 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/clbanning/mxj/v2 v2.7.0 // indirect github.com/clipperhouse/uax29/v2 v2.7.0 // indirect github.com/cloudwego/base64x v0.1.6 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/eino-contrib/jsonschema v1.0.3 // indirect github.com/emirpasic/gods/v2 v2.0.0-alpha // indirect @@ -34,6 +36,7 @@ require ( github.com/glebarez/go-sqlite v1.21.2 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect + github.com/gogf/gf/contrib/nosql/redis/v2 v2.10.2 // indirect github.com/google/uuid v1.6.0 // indirect github.com/goph/emperror v0.17.2 // indirect github.com/gorilla/websocket v1.5.3 // indirect @@ -55,6 +58,7 @@ require ( github.com/olekukonko/tablewriter v1.1.0 // indirect github.com/pelletier/go-toml/v2 v2.0.9 // indirect github.com/pkg/errors v0.9.1 // indirect + github.com/redis/go-redis/v9 v9.12.1 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/rivo/uniseg v0.2.0 // indirect github.com/sirupsen/logrus v1.9.3 // indirect diff --git a/go.sum b/go.sum index 1a5e65d..3b253f3 100644 --- a/go.sum +++ b/go.sum @@ -20,6 +20,8 @@ github.com/bytedance/sonic v1.15.0/go.mod h1:tFkWrPz0/CUCLEF4ri4UkHekCIcdnkqXw9V github.com/bytedance/sonic/loader v0.5.0 h1:gXH3KVnatgY7loH5/TkeVyXPfESoqSBSBEiDd5VjlgE= github.com/bytedance/sonic/loader v0.5.0/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo= github.com/certifi/gocertifi v0.0.0-20190105021004-abcd57078448/go.mod h1:GJKEexRPVJrBSOjoqN5VNOIKJ5Q3RViH6eu3puDRwx4= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/clbanning/mxj/v2 v2.7.0 h1:WA/La7UGCanFe5NpHF0Q3DNtnCsVoxbPKuyBNHWRyME= github.com/clbanning/mxj/v2 v2.7.0/go.mod h1:hNiWqW14h+kc+MdF9C6/YoRfjEJoR3ou6tn/Qo+ve2s= github.com/clipperhouse/uax29/v2 v2.7.0 h1:+gs4oBZ2gPfVrKPthwbMzWZDaAFPGYK72F0NJv2v7Vk= @@ -35,6 +37,8 @@ github.com/cloudwego/eino-ext/components/document/transformer/splitter/semantic github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/eino-contrib/jsonschema v1.0.3 h1:2Kfsm1xlMV0ssY2nuxshS4AwbLFuqmPmzIjLVJ1Fsp0= @@ -61,6 +65,8 @@ github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM= github.com/gogf/gf/contrib/drivers/sqlite/v2 v2.10.2 h1:KLS68SWS2W749x7e+eCCOO3UD2Sbw+bIbLEPR8o1FXw= github.com/gogf/gf/contrib/drivers/sqlite/v2 v2.10.2/go.mod h1:uLcsu73PfpyhRc0Jq0gGAWQjN1tyGU9iBRrYgt/lu7g= +github.com/gogf/gf/contrib/nosql/redis/v2 v2.10.2 h1:iTQegT+lEg/wDKvj2mi3W1wrdrwFarjokf88EXVVgu4= +github.com/gogf/gf/contrib/nosql/redis/v2 v2.10.2/go.mod h1:ZRw3GNz5cq4uYrW4TPSVyrYWaoqzujKdWro/AOcGBaE= github.com/gogf/gf/v2 v2.10.2 h1:46IO0Uc8e85/FqdftJFskfDejJLBL0JBnGS5qOftUu8= github.com/gogf/gf/v2 v2.10.2/go.mod h1:Svl1N+E8G/QshU2DUbh/3J/AJauqCgUnxHurXWR4Qx0= github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= @@ -143,6 +149,8 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/redis/go-redis/v9 v9.12.1 h1:k5iquqv27aBtnTm2tIkROUDp8JBXhXZIVu1InSgvovg= +github.com/redis/go-redis/v9 v9.12.1/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY=