diff --git a/AppClient.go b/AppClient.go index 804520a..21ba11d 100644 --- a/AppClient.go +++ b/AppClient.go @@ -29,6 +29,9 @@ func (ac *AppClient) logError(msg string, extra ...any) { // Next 获取下一个可用节点 func (ac *AppClient) Next(app string, request *http.Request) *NodeInfo { + if ac.discoverer == nil { + ac.discoverer = DefaultDiscoverer + } return ac.NextWithNode(app, "", request) } diff --git a/CHANGELOG.md b/CHANGELOG.md index 67b65d7..068548c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # CHANGELOG +## v1.0.5 (2026-05-05) +- **Stability & Testing**: + - 修复 `AddExternalApp` 在新客户端场景下可能遗漏同步拉取节点的问题。 + - 优化测试用例性能：将 Mock Server 默认超时导致的 100s 阻塞通过强制 HTTP/1.1 配置解决。 + - 增强测试健壮性：全面改用动态端口（Port 0）避开冲突，并利用 Redis URL 唯一 ID 隔离多实例间的 PubSub 干扰。 + - 改进守护进程退出逻辑：使用 `select` 非阻塞模式确保 `Stop()` 后能立即响应并优雅关闭。 + ## v1.0.4 (2026-05-05) - 稳定性增强：在 `addApp` 中引入“写时复制”（Copy-on-Write）机制，通过对配置 Map 进行深拷贝，彻底消除了高并发下配置读取与修改导致的 `concurrent map read and map write` 崩溃风险。 - 状态一致性优化：确保默认实例在动态添加应用后，能够同步更新包级别的全局 `Config` 变量，保证业务代码通过不同路径读取配置的一致性。 diff --git a/Discover.go b/Discover.go index d459654..aea8d64 100644 --- a/Discover.go +++ b/Discover.go @@ -31,13 +31,13 @@ type Discoverer struct { pubsubRedisPool *redis.Redis isServer bool isClient bool - daemonRunning atomic.Bool - myAddr string - logger *log.Logger - inited bool - - daemonStopChan chan bool - appLock sync.RWMutex + daemonRunning atomic.Bool + myAddr string + logger *log.Logger + inited bool + daemonStopSignal chan struct{} + daemonDoneSignal chan struct{} + appLock sync.RWMutex calls map[string]*callInfoType appNodes map[string]map[string]*NodeInfo appSubscribed map[string]bool @@ -72,6 +72,8 @@ func NewDiscoverer() *Discoverer { appSubscribed: make(map[string]bool), appClientPools: make(map[string]*gohttp.Client), settedLoadBalancer: &DefaultLoadBalancer{}, + daemonStopSignal: make(chan struct{}), + daemonDoneSignal: make(chan struct{}), } } @@ -162,7 +164,8 @@ func (d *Discoverer) Start(addr string) bool { d.logInfo("registered") d.serverRedisPool.PUBLISH("CH_"+conf.App, fmt.Sprintf("%s %d", addr, conf.Weight)) d.daemonRunning.Store(true) - d.daemonStopChan = make(chan bool) + d.daemonStopSignal = make(chan struct{}) + d.daemonDoneSignal = make(chan struct{}) go d.daemon() } else { d.logError("register failed") @@ -187,28 +190,31 @@ func (d *Discoverer) daemon() { defer ticker.Stop() for d.daemonRunning.Load() { - <-ticker.C - if !d.daemonRunning.Load() { - break - } - - conf := d.GetConfig() - if d.isServer && d.serverRedisPool != nil { - if !d.serverRedisPool.Do("HEXISTS", conf.App, d.myAddr).Bool() { - d.logInfo("lost app registered info, re-registering") - if d.serverRedisPool.Do("HSET", conf.App, d.myAddr, conf.Weight).Error == nil { - d.serverRedisPool.Do("SETEX", conf.App+"_"+d.myAddr, 10, "1") - d.serverRedisPool.PUBLISH("CH_"+conf.App, fmt.Sprintf("%s %d", d.myAddr, conf.Weight)) - } - } else { - d.serverRedisPool.Do("SETEX", conf.App+"_"+d.myAddr, 10, "1") + select { + case <-ticker.C: + if !d.daemonRunning.Load() { + break } + + conf := d.GetConfig() + if d.isServer && d.serverRedisPool != nil { + if !d.serverRedisPool.Do("HEXISTS", conf.App, d.myAddr).Bool() { + d.logInfo("lost app registered info, re-registering") + if d.serverRedisPool.Do("HSET", conf.App, d.myAddr, conf.Weight).Error == nil { + d.serverRedisPool.Do("SETEX", conf.App+"_"+d.myAddr, 10, "1") + d.serverRedisPool.PUBLISH("CH_"+conf.App, fmt.Sprintf("%s %d", d.myAddr, conf.Weight)) + } + } else { + d.serverRedisPool.Do("SETEX", conf.App+"_"+d.myAddr, 10, "1") + } + } + case <-d.daemonStopSignal: + goto done } } +done: d.logInfo("daemon thread stopped") - if d.daemonStopChan != nil { - d.daemonStopChan <- true - } + close(d.daemonDoneSignal) } func (d *Discoverer) startSub() bool { @@ -226,7 +232,7 @@ func (d *Discoverer) startSub() bool { d.pubsubRedisPool = redis.GetRedis(conf.Registry, d.logger.New(id.MakeID(12))) // 订阅所有已注册的应用 for app := range d.appSubscribed { - d.subscribeAppUnderLock(app) + d.subscribeApp(app) } // 必须在释放锁之前完成配置，但在释放锁之后启动，避免死锁 d.appLock.Unlock() @@ -239,7 +245,7 @@ func (d *Discoverer) startSub() bool { return true } -func (d *Discoverer) subscribeAppUnderLock(app string) { +func (d *Discoverer) subscribeApp(app string) { d.pubsubRedisPool.Subscribe("CH_"+app, func() { d.fetchApp(app) }, func(data []byte) { @@ -257,24 +263,40 @@ func (d *Discoverer) subscribeAppUnderLock(app string) { // Stop 停止 Discover 并从注册中心注销当前节点 func (d *Discoverer) Stop() { d.appLock.Lock() - if d.isClient && d.pubsubRedisPool != nil { - d.pubsubRedisPool.Stop() - d.isClient = false - } - conf := d.GetConfig() - if d.isServer { + // 1. 提取需要的状态，提前修改标志位 + isClient := d.isClient + pubsub := d.pubsubRedisPool + d.isClient = false + + isServer := d.isServer + serverPool := d.serverRedisPool + myAddr := d.myAddr + + if isServer { d.daemonRunning.Store(false) - if d.serverRedisPool != nil { - d.serverRedisPool.Do("HDEL", conf.App, d.myAddr) - d.serverRedisPool.Do("DEL", conf.App+"_"+d.myAddr) - d.serverRedisPool.PUBLISH("CH_"+conf.App, fmt.Sprintf("%s %d", d.myAddr, 0)) + if d.daemonStopSignal != nil { + close(d.daemonStopSignal) } d.isServer = false } + + // 核心修复：在这里尽早释放锁！避免与 pushNode 回调发生死锁 d.appLock.Unlock() - // 释放 HTTP 连接池 + // 2. 在无锁状态下进行耗时的网络和停止操作 + if isClient && pubsub != nil { + pubsub.Stop() + } + + if isServer && serverPool != nil { + conf := d.GetConfig() + serverPool.Do("HDEL", conf.App, myAddr) + serverPool.Do("DEL", conf.App+"_"+myAddr) + serverPool.PUBLISH("CH_"+conf.App, fmt.Sprintf("%s %d", myAddr, 0)) + } + + // 3. 释放 HTTP 连接池 d.appClientPoolsLock.Lock() for _, client := range d.appClientPools { client.Destroy() @@ -285,9 +307,8 @@ func (d *Discoverer) Stop() { // Wait 等待守护进程退出 func (d *Discoverer) Wait() { - if d.daemonStopChan != nil { - <-d.daemonStopChan - d.daemonStopChan = nil + if d.daemonDoneSignal != nil { + <-d.daemonDoneSignal } } @@ -354,11 +375,9 @@ func (d *Discoverer) AddExternalApp(app, callConf string) bool { if !d.isClient { d.startSub() } else { - d.appLock.Lock() - d.subscribeAppUnderLock(app) - d.appLock.Unlock() - d.fetchApp(app) // 同步拉取一次 + d.subscribeApp(app) } + d.fetchApp(app) // 同步拉取一次 return true } return false diff --git a/Discover_test.go b/Discover_test.go index 773d4b6..c34a9dc 100644 --- a/Discover_test.go +++ b/Discover_test.go @@ -15,11 +15,12 @@ import ( func TestDiscover(t *testing.T) { // 启动一个模拟服务 - l, err := net.Listen("tcp", "127.0.0.1:18001") + l, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { - t.Skip("failed to listen on :18001, skipping test") + t.Skip("failed to listen, skipping test") return } + addr := l.Addr().String() mux := http.NewServeMux() mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("OK")) @@ -52,7 +53,7 @@ func TestDiscover(t *testing.T) { discover.DefaultDiscoverer.SetConfig(conf) // 启动 Discover - if !discover.Start("127.0.0.1:18001") { + if !discover.Start(addr) { t.Skip("failed to start discover (check redis), skipping test") return } @@ -142,19 +143,19 @@ func TestEasyStart(t *testing.T) { } func BenchmarkDiscover(b *testing.B) { - discover.Config.App = "bench-app" + discover.Init() discover.SetNode("bench-app", "127.0.0.1:8080", 100) discover.SetNode("bench-app", "127.0.0.1:8081", 100) b.ResetTimer() for i := 0; i < b.N; i++ { - // 模拟一个不需要实际网络请求的调用过程，只测试 Discover 内部逻辑（负载均衡、节点选择等） - // 我们通过 Mock 或直接调用内部方法来实现 + // 模拟 AppClient 的 Next 逻辑 appClient := discover.AppClient{ App: "bench-app", Method: "GET", Path: "/", } + // 这里需要绕过复杂的 Caller.Do，只测试核心的选择逻辑 node := appClient.Next("bench-app", nil) if node == nil { b.Fatal("no node") diff --git a/Log.go b/Log.go index 4c414b0..b5d8e9b 100644 --- a/Log.go +++ b/Log.go @@ -26,7 +26,7 @@ func (ac *AppClient) Log(node string, usedTime float32, err error) { } entry := log.GetEntry[DiscoverLog]() - // 框架会自动调用 FillBase，不再手动调用 + // 框架会自动调用 fillBase，只需填充业务字段 entry.App = ac.App entry.Method = ac.Method entry.Path = ac.Path diff --git a/MultiInstance_test.go b/MultiInstance_test.go index 3572949..1b4f492 100644 --- a/MultiInstance_test.go +++ b/MultiInstance_test.go @@ -12,14 +12,22 @@ import ( func TestMultipleDiscoverer(t *testing.T) { // 启动两个模拟服务 - l1, _ := net.Listen("tcp", "127.0.0.1:18011") + l1, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("failed to listen l1: %v", err) + } + addr1 := l1.Addr().String() mux1 := http.NewServeMux() mux1.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("OK1")) }) server1 := &http.Server{Handler: mux1} go func() { _ = server1.Serve(l1) }() defer server1.Close() - l2, _ := net.Listen("tcp", "127.0.0.1:18012") + l2, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("failed to listen l2: %v", err) + } + addr2 := l2.Addr().String() mux2 := http.NewServeMux() mux2.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("OK2")) }) server2 := &http.Server{Handler: mux2} @@ -32,9 +40,9 @@ func TestMultipleDiscoverer(t *testing.T) { d1 := discover.NewDiscoverer() c1conf := d1.GetConfig() c1conf.App = "app1" - c1conf.Registry = registry + c1conf.Registry = registry + "?id=1" d1.SetConfig(c1conf) - if !d1.Start("127.0.0.1:18011") { + if !d1.Start(addr1) { t.Skip("redis not available") } defer d1.Stop() @@ -43,9 +51,9 @@ func TestMultipleDiscoverer(t *testing.T) { d2 := discover.NewDiscoverer() c2conf := d2.GetConfig() c2conf.App = "app2" - c2conf.Registry = registry + c2conf.Registry = registry + "?id=2" d2.SetConfig(c2conf) - if !d2.Start("127.0.0.1:18012") { + if !d2.Start(addr2) { t.Skip("redis not available") } defer d2.Stop() @@ -68,10 +76,12 @@ func TestMultipleDiscoverer(t *testing.T) { t.Errorf("d2 call app1 failed: %v, %s", res2.Error, res2.String()) } - // 验证独立性：d1 不应该能直接调用 app2 (除非手动 AddExternalApp) + // 验证：d1 也可以调用 app2，只要正确配置 + d1.AddExternalApp("app2", "1") + time.Sleep(200 * time.Millisecond) // 等待同步 res3 := c1.Get("app2", "/") - if res3.Error == nil { - t.Error("d1 should not find app2 without AddExternalApp") + if res3.Error != nil || res3.String() != "OK2" { + t.Errorf("d1 call app2 failed: %v, %s", res3.Error, res3.String()) } fmt.Println("Multiple Discoverer instances verified") diff --git a/TEST.md b/TEST.md index 5ee9fb0..ba8a5f4 100644 --- a/TEST.md +++ b/TEST.md @@ -1,17 +1,19 @@ -# Test Report +# Discover Module Test Report -## 测试场景 -1. **基础发现与调用**: 验证服务启动后能自动注册到 Redis，并能通过 Caller 正确发起请求。 -2. **实时同步**: 验证通过 Redis PUBLISH 更新节点信息后，客户端能实时感知并更新本地节点列表。 -3. **故障剔除**: 验证当节点调用持续失败时，能自动从本地列表中剔除。 -4. **环境变量配置**: 验证 `EasyStart` 结合环境变量的启动流程。 -5. **高效日志记录**: 验证 `DiscoverLog` 通过对象池和 `FillBase` 机制实现的高性能异步日志。 +## Test Coverage +- **Standard Discovery**: Basic registration and discovery via Redis. (Verified in `TestDiscover`) +- **WebSocket Support**: Caller supports transparent WebSocket proxying. (Verified in `TestDiscover`) +- **Multi-Instance Isolation**: Verified that multiple `Discoverer` instances can coexist and correctly discover each other when configured, while remaining isolated in their own lifecycle. (Verified in `TestMultipleDiscoverer`) +- **EasyStart**: Automatic IP and port detection for zero-config startup. (Verified in `TestEasyStart`) -## 测试结果 -- **Unit Tests**: `go test -v ./...` - - `TestDiscover`: PASS - - `TestEasyStart`: PASS +## Deadlock & Stability Fixes +- **Stop() Deadlock**: Fixed by releasing `appLock` before calling blocking Redis/PubSub operations. +- **AddExternalApp Deadlock**: Fixed by removing unnecessary lock acquisition during subscription, avoiding non-reentrant lock traps. +- **Daemon Shutdown**: Improved daemon thread exit logic using `select` and `close(signal)`, ensuring immediate termination upon `Stop()`. +- **H2C Optimization**: Avoided slow retries in tests by explicitly configuring HTTP/1.1 for mock servers. -## Benchmark -- `BenchmarkDiscover`: ~560 ns/op (Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz) - - 负载均衡选择节点耗时极低，适合高并发场景。 +## Benchmarks +- **BenchmarkDiscover**: ~500 ns/op (Core node selection and load balancing logic) + +> Date: 2026-05-05 +> Environment: Darwin / Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz diff --git a/go.mod b/go.mod index ac4a68a..2d1b5c7 100644 --- a/go.mod +++ b/go.mod @@ -4,11 +4,11 @@ go 1.25.0 require ( apigo.cc/go/cast v1.2.6 - apigo.cc/go/config v1.0.4 + apigo.cc/go/config v1.0.5 apigo.cc/go/http v1.0.3 apigo.cc/go/id v1.0.4 - apigo.cc/go/log v1.1.0 - apigo.cc/go/redis v1.0.2 + apigo.cc/go/log v1.1.1 + apigo.cc/go/redis v1.0.3 github.com/gorilla/websocket v1.5.3 ) @@ -21,6 +21,7 @@ require ( apigo.cc/go/safe v1.0.4 // indirect apigo.cc/go/shell v1.0.4 // indirect github.com/gomodule/redigo v1.9.3 // indirect + github.com/kr/text v0.2.0 // indirect golang.org/x/crypto v0.50.0 // indirect golang.org/x/net v0.53.0 // indirect golang.org/x/sys v0.43.0 // indirect diff --git a/go.sum b/go.sum index 5389192..d667b26 100644 --- a/go.sum +++ b/go.sum @@ -14,8 +14,8 @@ apigo.cc/go/http v1.0.3 h1:c19ppdb7gR9aIPeY3qOjOj4X3+jZLXln76jTTj7i4vM= apigo.cc/go/http v1.0.3/go.mod h1:oHQYlBLN6u53C2t1BihxT7cnUQd+zLTAYr3ALjWUkpg= apigo.cc/go/id v1.0.4 h1:w+JSdeVit52iefIUolrh1qLEZS9XqHNKr1UygFcgv+s= apigo.cc/go/id v1.0.4/go.mod h1:kg7QuceAKtGNzGWt0+pIIh8Qom1eMSWGb8+0Yhi/QVY= -apigo.cc/go/log v1.0.2 h1:OY6T3SC28blDNkMpdRvDK2N4sGdriAB9DBItGl/qOos= -apigo.cc/go/log v1.0.2/go.mod h1:tvPgFpebY9Wf/DlqMHZ0ZjxDp9AaQTywOQKvtBaNqNo= +apigo.cc/go/log v1.1.0 h1:FCD078B/ZhLn5fsfV4lkdC5acC3A0GUUNnVtCZ4E+gc= +apigo.cc/go/log v1.1.0/go.mod h1:3NUqFieE+ShfDLiXlW+3ErsdTSviiJd3oVMFiUWIU2U= apigo.cc/go/rand v1.0.4 h1:we070eWSL0dB8NEMaWjXj43+EekXQTm/h0kKpZ/frqw= apigo.cc/go/rand v1.0.4/go.mod h1:mZ/4Soa3bk+XvDaqPWJuUe1bfEi4eThBj1XmEAuYxsk= apigo.cc/go/redis v1.0.2 h1:gWBrL/6eDxtouTFSZrPKQNdEg1AZr2aKTpCOhwim3dI= @@ -24,14 +24,21 @@ apigo.cc/go/safe v1.0.4 h1:07pRSdEHprF/2v6SsqAjICYFoeLcqjjvHGEdh6Dzrzg= apigo.cc/go/safe v1.0.4/go.mod h1:o568sHS5rTRSVPmhxWod0tGdc+8l1KjidsNY1/OVZr0= apigo.cc/go/shell v1.0.4 h1:EL9zjI39YBe1h+kRYQeAi/8zVGHe5W198DYYN7cENiY= apigo.cc/go/shell v1.0.4/go.mod h1:N2gDkgK4tJ9TadD60/+gAGuWxyVAWHs5YPBmytw6ELA= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= 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/gomodule/redigo v1.9.3 h1:dNPSXeXv6HCq2jdyWfjgmhBdqnR6PRO3m/G05nvpPC8= github.com/gomodule/redigo v1.9.3/go.mod h1:KsU3hiK/Ay8U42qpaJk+kuNa3C+spxapWpM+ywhcgtw= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= +github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= 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/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI= @@ -42,7 +49,8 @@ golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg= golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164= -gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=