chore: code review and fix race conditions in redis module (by AI)
This commit is contained in:
AI Engineer
parent
69324d566e
commit
067e74d46d
16
AI.md
Normal file
16
AI.md
Normal file
@ -0,0 +1,16 @@
|
||||
# AI 调用规则 - redis
|
||||
|
||||
## 当前版本
|
||||
v1.0.1
|
||||
|
||||
## AI 准则
|
||||
- **类型安全**: 优先使用 `any` 替代 `interface{}`。
|
||||
- **泛型优先**: 结果转换应优先推荐使用 `To[T]`。
|
||||
- **防御性编程**: 在执行 `Do` 或 `Subscribe` 前必须检查 `pool` 是否初始化。
|
||||
- **线程安全**: 所有对 `subs` 和 `subConn` 的操作必须持有 `subLock`。
|
||||
- **内存安全**: 避免在日志或错误信息中直接打印 Redis 密码，使用 `conf.Dsn()` 提供的脱敏版本。
|
||||
|
||||
## 核心 API 路径
|
||||
- `GetRedis`: 入口函数，建议使用单例模式获取。
|
||||
- `Result.To`: 底层反序列化逻辑。
|
||||
- `IdMaker.makeSecIndex`: 分布式 ID 预取逻辑，步长固定为 100。
|
||||
@ -1,5 +1,14 @@
|
||||
# CHANGELOG - redis
|
||||
|
||||
## v1.0.1 (2026-05-03)
|
||||
- **Security & Stability**:
|
||||
- 修复了 `Pub/Sub` 模块中 `subs` map 和 `subConn` 的并发访问竞争风险（Race Condition），引入 `subLock` 互斥锁。
|
||||
- **Cleanup**:
|
||||
- 移除了 `Redis` 结构体中冗余的 `ReadTimeout` 字段，统一由 `Config` 管理。
|
||||
- **Testing**:
|
||||
- 新增 `bench_test.go`，建立性能基准测试。
|
||||
- 新增 `TEST.md` 记录测试覆盖场景与 Benchmark 结果。
|
||||
|
||||
## v1.0.0 (2026-05-03)
|
||||
- **Repo Migration**: 从 `@ssgo/redis` 迁移至 `apigo.cc/go/redis`。
|
||||
- **Standard Realignment**:
|
||||
|
||||
25
TEST.md
Normal file
25
TEST.md
Normal file
@ -0,0 +1,25 @@
|
||||
# redis 模块测试报告
|
||||
|
||||
## 测试场景
|
||||
1. **基础操作 (TestBase)**:
|
||||
- 验证 `GET`, `SET`, `DEL`, `EXISTS`, `GETSET` 等基本命令。
|
||||
- 验证 `EXPIRE` 自动过期功能。
|
||||
- 验证结构体自动序列化与反序列化。
|
||||
- 验证 `MSET`, `MGET` 批量操作。
|
||||
2. **泛型支持 (TestGenerics)**:
|
||||
- 验证 `To[T]` 泛型函数对结果的反序列化。
|
||||
3. **发布订阅 (TestSub)**:
|
||||
- 验证 `Subscribe`, `Unsubscribe`, `PUBLISH` 功能。
|
||||
- 验证并发订阅与取消订阅的稳定性。
|
||||
|
||||
## Benchmark 结果
|
||||
```bash
|
||||
goos: darwin
|
||||
goarch: amd64
|
||||
pkg: apigo.cc/go/redis
|
||||
cpu: Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
|
||||
BenchmarkGetSet-16 4057 283694 ns/op
|
||||
BenchmarkIdMaker-16 338056 3377 ns/op
|
||||
```
|
||||
- `BenchmarkGetSet`: 每次 GET+SET 耗时约 283微秒（受本地 Redis 响应速度影响）。
|
||||
- `BenchmarkIdMaker`: 每次获取分布式 ID 耗时约 3.4微秒（得益于 100 步长的预取机制）。
|
||||
36
bench_test.go
Normal file
36
bench_test.go
Normal file
@ -0,0 +1,36 @@
|
||||
package redis_test
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"apigo.cc/go/config"
|
||||
"apigo.cc/go/redis"
|
||||
)
|
||||
|
||||
func BenchmarkGetSet(b *testing.B) {
|
||||
os.Setenv("REDIS_TEST", "redis://:@localhost:6379/2?timeout=10ms&logSlow=10us")
|
||||
_ = config.Load("redis", nil)
|
||||
rd := redis.GetRedis("test", nil)
|
||||
rd.DEL("bench_key")
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
rd.SET("bench_key", "bench_value")
|
||||
_ = rd.GET("bench_key").String()
|
||||
}
|
||||
b.StopTimer()
|
||||
rd.DEL("bench_key")
|
||||
}
|
||||
|
||||
func BenchmarkIdMaker(b *testing.B) {
|
||||
os.Setenv("REDIS_TEST", "redis://:@localhost:6379/2?timeout=10ms&logSlow=10us")
|
||||
_ = config.Load("redis", nil)
|
||||
rd := redis.GetRedis("test", nil)
|
||||
im := redis.NewIdMaker(rd)
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
im.Get(16)
|
||||
}
|
||||
}
|
||||
4
redis.go
4
redis.go
@ -22,11 +22,11 @@ import (
|
||||
type Redis struct {
|
||||
name string
|
||||
pool *redis.Pool
|
||||
ReadTimeout int
|
||||
Config *Config
|
||||
logger *log.Logger
|
||||
Error error
|
||||
subConn *redis.PubSubConn
|
||||
subLock sync.RWMutex
|
||||
subStopChan chan bool
|
||||
subs map[string]*SubCallbacks
|
||||
SubRunning bool
|
||||
@ -139,7 +139,6 @@ func NewRedis(conf *Config, logger *log.Logger) *Redis {
|
||||
}
|
||||
|
||||
rd := new(Redis)
|
||||
rd.ReadTimeout = int(conf.ReadTimeout / time.Millisecond)
|
||||
rd.pool = conn
|
||||
rd.Config = conf
|
||||
rd.logger = logger
|
||||
@ -150,7 +149,6 @@ func NewRedis(conf *Config, logger *log.Logger) *Redis {
|
||||
func (rd *Redis) CopyByLogger(logger *log.Logger) *Redis {
|
||||
newRedis := new(Redis)
|
||||
newRedis.name = rd.name
|
||||
newRedis.ReadTimeout = rd.ReadTimeout
|
||||
newRedis.pool = rd.pool
|
||||
newRedis.subConn = rd.subConn
|
||||
newRedis.subs = rd.subs
|
||||
|
||||
42
subscribe.go
42
subscribe.go
@ -8,6 +8,9 @@ import (
|
||||
)
|
||||
|
||||
func (rd *Redis) Subscribe(name string, reset func(), received func([]byte)) bool {
|
||||
rd.subLock.Lock()
|
||||
defer rd.subLock.Unlock()
|
||||
|
||||
if rd.subs == nil {
|
||||
rd.subs = make(map[string]*SubCallbacks)
|
||||
}
|
||||
@ -24,6 +27,9 @@ func (rd *Redis) Subscribe(name string, reset func(), received func([]byte)) boo
|
||||
}
|
||||
|
||||
func (rd *Redis) Unsubscribe(name string) bool {
|
||||
rd.subLock.Lock()
|
||||
defer rd.subLock.Unlock()
|
||||
|
||||
if rd.subs != nil {
|
||||
delete(rd.subs, name)
|
||||
}
|
||||
@ -39,10 +45,13 @@ func (rd *Redis) Unsubscribe(name string) bool {
|
||||
}
|
||||
|
||||
func (rd *Redis) Start() {
|
||||
rd.subLock.Lock()
|
||||
if rd.subs == nil {
|
||||
rd.subs = make(map[string]*SubCallbacks)
|
||||
}
|
||||
rd.SubRunning = true
|
||||
rd.subLock.Unlock()
|
||||
|
||||
subStartChan := make(chan bool)
|
||||
go rd.receiveSub(subStartChan)
|
||||
<-subStartChan
|
||||
@ -50,14 +59,19 @@ func (rd *Redis) Start() {
|
||||
|
||||
func (rd *Redis) receiveSub(subStartChan chan bool) {
|
||||
for {
|
||||
if !rd.SubRunning {
|
||||
rd.subLock.RLock()
|
||||
running := rd.SubRunning
|
||||
rd.subLock.RUnlock()
|
||||
if !running {
|
||||
break
|
||||
}
|
||||
|
||||
// 开始接收订阅数据
|
||||
rd.subLock.Lock()
|
||||
if rd.subConn == nil {
|
||||
conn, err := rd.GetConnection()
|
||||
if err != nil {
|
||||
rd.subLock.Unlock()
|
||||
time.Sleep(time.Second)
|
||||
continue
|
||||
}
|
||||
@ -72,6 +86,7 @@ func (rd *Redis) receiveSub(subStartChan chan bool) {
|
||||
if err != nil {
|
||||
_ = rd.subConn.Close()
|
||||
rd.subConn = nil
|
||||
rd.subLock.Unlock()
|
||||
time.Sleep(time.Second)
|
||||
continue
|
||||
}
|
||||
@ -83,6 +98,7 @@ func (rd *Redis) receiveSub(subStartChan chan bool) {
|
||||
}
|
||||
}
|
||||
}
|
||||
rd.subLock.Unlock()
|
||||
|
||||
if subStartChan != nil {
|
||||
subStartChan <- true
|
||||
@ -91,10 +107,20 @@ func (rd *Redis) receiveSub(subStartChan chan bool) {
|
||||
|
||||
for {
|
||||
isErr := false
|
||||
receiveObj := rd.subConn.Receive()
|
||||
rd.subLock.RLock()
|
||||
subConn := rd.subConn
|
||||
rd.subLock.RUnlock()
|
||||
|
||||
if subConn == nil {
|
||||
break
|
||||
}
|
||||
|
||||
receiveObj := subConn.Receive()
|
||||
switch v := receiveObj.(type) {
|
||||
case redis.Message:
|
||||
rd.subLock.RLock()
|
||||
callback := rd.subs[v.Channel]
|
||||
rd.subLock.RUnlock()
|
||||
if callback != nil && callback.received != nil {
|
||||
callback.received(v.Data)
|
||||
}
|
||||
@ -107,13 +133,19 @@ func (rd *Redis) receiveSub(subStartChan chan bool) {
|
||||
if !strings.Contains(v.Error(), "connection closed") && !strings.Contains(v.Error(), "use of closed network connection") {
|
||||
rd.logger.Error(v.Error(), "type", "redis", "action", "receiveSub")
|
||||
}
|
||||
rd.subLock.Lock()
|
||||
if rd.subConn != nil {
|
||||
_ = rd.subConn.Close()
|
||||
rd.subConn = nil
|
||||
}
|
||||
rd.subLock.Unlock()
|
||||
isErr = true
|
||||
}
|
||||
if isErr || !rd.SubRunning {
|
||||
|
||||
rd.subLock.RLock()
|
||||
running = rd.SubRunning
|
||||
rd.subLock.RUnlock()
|
||||
if isErr || !running {
|
||||
break
|
||||
}
|
||||
}
|
||||
@ -124,6 +156,7 @@ func (rd *Redis) receiveSub(subStartChan chan bool) {
|
||||
}
|
||||
|
||||
func (rd *Redis) Stop() {
|
||||
rd.subLock.Lock()
|
||||
if rd.SubRunning {
|
||||
rd.subStopChan = make(chan bool)
|
||||
rd.SubRunning = false
|
||||
@ -137,8 +170,11 @@ func (rd *Redis) Stop() {
|
||||
_ = rd.subConn.Close()
|
||||
rd.subConn = nil
|
||||
}
|
||||
rd.subLock.Unlock()
|
||||
<-rd.subStopChan
|
||||
rd.subStopChan = nil
|
||||
} else {
|
||||
rd.subLock.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user