go/discover
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: go:v1.0.2
Branches Tags
go:master
go:main
go:v1.3.2
go:v1.3.0
go:v1.0.11
go:v1.0.10
go:v1.0.9
go:v1.0.8
go:v1.0.7
go:v1.0.6
go:v1.0.5
go:v1.0.4
go:v1.0.3
go:v1.0.2
go:v1.0.1
..
compare: go:master
Branches Tags
go:main
go:master
go:v1.3.2
go:v1.3.0
go:v1.0.11
go:v1.0.10
go:v1.0.9
go:v1.0.8
go:v1.0.7
go:v1.0.6
go:v1.0.5
go:v1.0.4
go:v1.0.3
go:v1.0.2
go:v1.0.1

No commits in common. "v1.0.2" and "master" have entirely different histories.
v1.0.2 ... master

14 changed files with 274 additions and 539 deletions
Show Stats Expand all files Collapse all files

43
AppClient.go
View File

@ -8,63 +8,58 @@ import (
// AppClient 用于管理单个请求的重试和负载均衡状态
type AppClient struct {
discoverer *Discoverer
excludes map[string]bool // 本次请求已排除的节点
attempts int // 本次请求的重试次数
Logger *log.Logger // 用于日志记录的 Logger
App string // 目标应用名称
Method string // 请求方法
Path string // 请求路径
Data map[string]any // 请求数据
Headers map[string]string // 请求头
excludes map[string]bool
tryTimes int
Logger *log.Logger
App string
Method string
Path string
Data *map[string]any
Headers *map[string]string
}
// logError 记录 Discover 客户端错误
func (ac *AppClient) logError(msg string, extra ...any) {
func (ac *AppClient) logError(error string, extra ...any) {
if ac.Logger == nil {
ac.Logger = log.DefaultLogger
}
ac.Logger.Error("Discover Client: "+msg, extra...)
ac.Logger.Error("Discover Client: "+error, extra...)
}
// Next 获取下一个可用节点
func (ac *AppClient) Next(app string, request *http.Request) *NodeInfo {
return ac.NextWithNode(app, "", request)
}
// CheckApp 检查并尝试添加应用
func (ac *AppClient) CheckApp(app string) bool {
nodes := ac.discoverer.GetAppNodes(app)
nodes := getAppNodes(app)
if nodes == nil {
if !ac.discoverer.AddExternalApp(app, "") {
ac.logError("app not found", "app", app, "calls", ac.discoverer.Config.Calls)
if !addApp(app, "", true) {
ac.logError("app not found", "app", app, "calls", Config.Calls)
return false
}
}
return true
}
// NextWithNode 获取下一个可用节点，支持指定节点
func (ac *AppClient) NextWithNode(app, withNode string, request *http.Request) *NodeInfo {
if ac.excludes == nil {
ac.excludes = make(map[string]bool)
}
allNodes := ac.discoverer.GetAppNodes(app)
allNodes := getAppNodes(app)
if len(allNodes) == 0 {
ac.logError("node not found", "app", app)
return nil
}
ac.attempts++
ac.tryTimes++
if withNode != "" {
ac.excludes[withNode] = true
return allNodes[withNode]
}
readyNodes := make([]*NodeInfo, 0, len(allNodes))
readyNodes := make([]*NodeInfo, 0)
for _, node := range allNodes {
if ac.excludes[node.Addr] || node.FailedTimes.Load() >= int32(ac.discoverer.Config.CallRetryTimes) {
if ac.excludes[node.Addr] || node.FailedTimes >= Config.CallRetryTimes {
continue
}
readyNodes = append(readyNodes, node)
@ -81,14 +76,14 @@ func (ac *AppClient) NextWithNode(app, withNode string, request *http.Request) *
var node *NodeInfo
if len(readyNodes) > 0 {
node = ac.discoverer.settedLoadBalancer.Next(ac, readyNodes, request)
node = settedLoadBalancer.Next(ac, readyNodes, request)
if node != nil {
ac.excludes[node.Addr] = true
}
}
if node == nil {
ac.logError("no available node", "app", app, "attempts", ac.attempts)
ac.logError("no available node", "app", app, "tryTimes", ac.tryTimes)
}
return node

10
CHANGELOG.md
View File

@ -1,15 +1,5 @@
# CHANGELOG
## v1.0.1 (2026-05-05)
- 优化代码规范：修复变量名冲突（Shadowing），改进 `tryTimes` -> `attempts` 等语义命名。
- 性能优化：优化 `AppClient` 中的 `Data` 和 `Headers` 类型（从指针改为直接引用），减少内存寻址开销。
- 性能优化：优化 `NextWithNode` 中的切片分配。
- 架构优化：导出 `log.FillBase`，支持外部模块实现高效自定义日志。
- 功能增强：引入 `DiscoverLog`，实现基于对象池的高性能发现过程日志记录。
- 标准对齐：统一使用 `apigo.cc/go/http` 中定义的 Header 常量。
- 文档完善：为所有导出类型和方法添加详细文档注释。
- 测试增强：添加 `BenchmarkDiscover` 基准测试。
## v1.0.0
- 从 `ssgo/discover` 迁移至 `apigo.cc/go/discover`。
- 采用全新的 `apigo.cc/go` 基础设施（log, redis, http, cast, u）。

73
Caller.go
View File

@ -41,64 +41,48 @@ func getHttpClient(app string, timeout time.Duration, h2c bool) *gohttp.Client {
return c
}
// Caller 用于发起服务间调用
type Caller struct {
discoverer *Discoverer
Request *http.Request // 原始请求，用于透传 Header
NoBody bool // 是否不发送请求体
logger *log.Logger // 用于日志记录的 Logger
Request *http.Request
NoBody bool
logger *log.Logger
}
// NewCaller 创建一个新的调用器
func NewCaller(request *http.Request, logger *log.Logger) *Caller {
return DefaultDiscoverer.NewCaller(request, logger)
return &Caller{Request: request, logger: logger}
}
// NewCaller 创建一个新的调用器实例
func (d *Discoverer) NewCaller(request *http.Request, logger *log.Logger) *Caller {
return &Caller{discoverer: d, Request: request, logger: logger}
}
// logError 记录 Discover 调用器错误
func (c *Caller) logError(msg string, extra ...any) {
func (c *Caller) logError(error string, extra ...any) {
if c.logger == nil {
c.logger = log.DefaultLogger
}
c.logger.Error("Discover Caller: "+msg, extra...)
c.logger.Error("Discover Caller: "+error, extra...)
}
// Get 发起 GET 请求
func (c *Caller) Get(app, path string, headers ...string) *gohttp.Result {
return c.Do("GET", app, path, nil, headers...)
}
// Post 发起 POST 请求
func (c *Caller) Post(app, path string, data any, headers ...string) *gohttp.Result {
return c.Do("POST", app, path, data, headers...)
}
// Put 发起 PUT 请求
func (c *Caller) Put(app, path string, data any, headers ...string) *gohttp.Result {
return c.Do("PUT", app, path, data, headers...)
}
// Delete 发起 DELETE 请求
func (c *Caller) Delete(app, path string, data any, headers ...string) *gohttp.Result {
return c.Do("DELETE", app, path, data, headers...)
}
// Head 发起 HEAD 请求
func (c *Caller) Head(app, path string, headers ...string) *gohttp.Result {
return c.Do("HEAD", app, path, nil, headers...)
}
// Do 发起通用请求
func (c *Caller) Do(method, app, path string, data any, headers ...string) *gohttp.Result {
r, _ := c.DoWithNode(method, app, "", path, data, headers...)
return r
}
// Open 发起 WebSocket 连接
func (c *Caller) Open(app, path string, headers ...string) *websocket.Conn {
r, _ := c.doWithNode(false, "WS", app, "", path, nil, headers...)
if v, ok := r.(*websocket.Conn); ok {
@ -107,7 +91,6 @@ func (c *Caller) Open(app, path string, headers ...string) *websocket.Conn {
return nil
}
// DoWithNode 发起请求并返回结果及节点地址
func (c *Caller) DoWithNode(method, app, withNode, path string, data any, headers ...string) (*gohttp.Result, string) {
r, nodeAddr := c.doWithNode(false, method, app, withNode, path, data, headers...)
if v, ok := r.(*gohttp.Result); ok {
@ -116,7 +99,6 @@ func (c *Caller) DoWithNode(method, app, withNode, path string, data any, header
return nil, nodeAddr
}
// ManualDoWithNode 发起请求（手动处理响应）并返回结果及节点地址
func (c *Caller) ManualDoWithNode(method, app, withNode, path string, data any, headers ...string) (*gohttp.Result, string) {
r, nodeAddr := c.doWithNode(true, method, app, withNode, path, data, headers...)
if v, ok := r.(*gohttp.Result); ok {
@ -125,15 +107,15 @@ func (c *Caller) ManualDoWithNode(method, app, withNode, path string, data any,
return nil, nodeAddr
}
func (c *Caller) doWithNode(manual bool, method, app, withNode, path string, data any, headers ...string) (any, string) {
func (c *Caller) doWithNode(manualDo bool, method, app, withNode, path string, data any, headers ...string) (any, string) {
callerHeaders := make(map[string]string)
for i := 1; i < len(headers); i += 2 {
callerHeaders[headers[i-1]] = headers[i]
}
if c.discoverer.isServer {
callerHeaders[HeaderFromApp] = c.discoverer.Config.App
callerHeaders[HeaderFromNode] = c.discoverer.myAddr
if isServer {
callerHeaders[HeaderFromApp] = Config.App
callerHeaders[HeaderFromNode] = myAddr
}
callData := make(map[string]any)
@ -145,17 +127,16 @@ func (c *Caller) doWithNode(manual bool, method, app, withNode, path string, dat
}
appClient := AppClient{
discoverer: c.discoverer,
Logger: c.logger,
App: app,
Method: method,
Path: path,
Data: callData,
Headers: callerHeaders,
Data: &callData,
Headers: &callerHeaders,
}
if c.discoverer.settedRoute != nil {
c.discoverer.settedRoute(&appClient, c.Request)
if settedRoute != nil {
settedRoute(&appClient, c.Request)
app = appClient.App
method = appClient.Method
path = appClient.Path
@ -165,7 +146,7 @@ func (c *Caller) doWithNode(manual bool, method, app, withNode, path string, dat
return &gohttp.Result{Error: fmt.Errorf("app %s not found", app)}, ""
}
callInfo := c.discoverer.getCallInfo(app)
callInfo := getCallInfo(app)
if callInfo != nil && callInfo.Token != "" {
callerHeaders["Access-Token"] = callInfo.Token
}
@ -181,7 +162,7 @@ func (c *Caller) doWithNode(manual bool, method, app, withNode, path string, dat
break
}
node.UsedTimes.Add(1)
node.UsedTimes++
startTime := time.Now()
scheme := "http"
if callInfo != nil && callInfo.SSL {
@ -213,13 +194,13 @@ func (c *Caller) doWithNode(manual bool, method, app, withNode, path string, dat
res = &gohttp.Result{Error: err, Response: resp}
} else {
if c.Request != nil {
if manual {
if manualDo {
res = hc.ManualDoByRequest(c.Request, method, url, data, settedHeaders...)
} else {
res = hc.DoByRequest(c.Request, method, url, data, settedHeaders...)
}
} else {
if manual {
if manualDo {
res = hc.ManualDo(method, url, data, settedHeaders...)
} else {
res = hc.Do(method, url, data, settedHeaders...)
@ -228,11 +209,10 @@ func (c *Caller) doWithNode(manual bool, method, app, withNode, path string, dat
}
responseTime := time.Since(startTime)
usedTimeMs := float32(responseTime.Nanoseconds()) / 1e6
c.discoverer.settedLoadBalancer.Response(&appClient, node, res.Error, res.Response, responseTime)
settedLoadBalancer.Response(&appClient, node, res.Error, res.Response, responseTime)
if res.Error != nil || (res.Response != nil && res.Response.StatusCode >= 502 && res.Response.StatusCode <= 504) {
node.FailedTimes.Add(1)
node.FailedTimes++
errStr := ""
if res.Error != nil {
errStr = res.Error.Error()
@ -240,17 +220,18 @@ func (c *Caller) doWithNode(manual bool, method, app, withNode, path string, dat
errStr = res.Response.Status
}
c.logError(errStr, "app", app, "node", node.Addr, "path", path, "attempts", appClient.attempts)
appClient.Log(node.Addr, usedTimeMs, fmt.Errorf("%s", errStr))
c.logError(errStr, "app", app, "node", node.Addr, "path", path, "tryTimes", appClient.tryTimes)
if node.FailedTimes.Load() >= int32(c.discoverer.Config.CallRetryTimes) {
c.discoverer.logError("node isolated locally due to high failures", "app", app, "node", node.Addr)
if node.FailedTimes >= Config.CallRetryTimes {
logError("node removed due to high failures", "app", app, "node", node.Addr)
if clientRedisPool != nil {
clientRedisPool.Do("HDEL", app, node.Addr)
clientRedisPool.PUBLISH("CH_"+app, fmt.Sprintf("%s 0", node.Addr))
}
}
continue
}
node.FailedTimes.Store(0)
appClient.Log(node.Addr, usedTimeMs, nil)
if strings.ToUpper(method) == "WS" {
return wsConn, node.Addr
}

9
Config.go
View File

@ -1,17 +1,14 @@
package discover
// ConfigStruct 存储发现服务的配置
type ConfigStruct struct {
// Config 存储发现服务的全局配置
var Config = struct {
Registry string // 注册中心地址，如 redis://:@127.0.0.1:6379/15
App string // 当前应用名称
Weight int // 权重，默认为 100
Calls map[string]string // 调用的应用列表及其配置
CallRetryTimes int // 调用重试次数
IpPrefix string // 指定使用的 IP 网段
}
// Config 存储发现服务的全局配置（兼容旧代码）
var Config = ConfigStruct{
}{
Weight: 100,
CallRetryTimes: 10,
}

44
Constants.go
View File

@ -1,27 +1,35 @@
package discover
import (
gohttp "apigo.cc/go/http"
)
const (
HeaderFromApp = gohttp.HeaderFromApp
HeaderFromNode = gohttp.HeaderFromNode
HeaderFromApp = "X-Discover-From-App"
HeaderFromNode = "X-Discover-From-Node"
HeaderClientIP = gohttp.HeaderClientIP
HeaderForwardedFor = gohttp.HeaderForwardedFor
HeaderUserID = gohttp.HeaderUserID
HeaderDeviceID = gohttp.HeaderDeviceID
HeaderClientAppName = gohttp.HeaderClientAppName
HeaderClientAppVersion = gohttp.HeaderClientAppVersion
HeaderSessionID = gohttp.HeaderSessionID
HeaderRequestID = gohttp.HeaderRequestID
HeaderHost = gohttp.HeaderHost
HeaderScheme = gohttp.HeaderScheme
HeaderUserAgent = gohttp.HeaderUserAgent
HeaderClientIp = "X-Client-Ip"
HeaderForwardedFor = "X-Forwarded-For"
HeaderUserId = "X-User-Id"
HeaderDeviceId = "X-Device-Id"
HeaderClientAppName = "X-Client-App-Name"
HeaderClientAppVersion = "X-Client-App-Version"
HeaderSessionId = "X-Session-Id"
HeaderRequestId = "X-Request-Id"
HeaderHost = "X-Host"
HeaderScheme = "X-Scheme"
HeaderUserAgent = "User-Agent"
)
var RelayHeaders = gohttp.RelayHeaders
var RelayHeaders = []string{
HeaderClientIp,
HeaderForwardedFor,
HeaderUserId,
HeaderDeviceId,
HeaderClientAppName,
HeaderClientAppVersion,
HeaderSessionId,
HeaderRequestId,
HeaderHost,
HeaderScheme,
HeaderUserAgent,
}
const DefaultRegistry = "127.0.0.1:6379::15"
const EnvRegistry = "DISCOVER_REGISTRY"

437
Discover.go
View File

@ -19,29 +19,26 @@ import (
"apigo.cc/go/redis"
)
// Discoverer 发现服务实例
type Discoverer struct {
Config ConfigStruct
var (
serverRedisPool *redis.Redis
clientRedisPool *redis.Redis
pubsubRedisPool *redis.Redis
isServer bool
isClient bool
daemonRunning bool
myAddr string
logger *log.Logger
inited bool
isServer = false
isClient = false
daemonRunning = false
myAddr = ""
_logger = log.DefaultLogger
_inited = false
daemonStopChan chan bool
appLock sync.RWMutex
calls map[string]*callInfoType
appNodes map[string]map[string]*NodeInfo
appSubscribed map[string]bool
_calls = map[string]*callInfoType{}
_appNodes = map[string]map[string]*NodeInfo{}
appSubscribed = map[string]bool{}
settedRoute func(*AppClient, *http.Request)
settedLoadBalancer LoadBalancer
}
settedRoute func(*AppClient, *http.Request) = nil
settedLoadBalancer LoadBalancer = &DefaultLoadBalancer{}
)
type callInfoType struct {
Timeout time.Duration
@ -50,168 +47,138 @@ type callInfoType struct {
SSL bool
}
// DefaultDiscoverer 默认的全局发现服务实例
var DefaultDiscoverer = NewDiscoverer()
func IsServer() bool { return isServer }
func IsClient() bool { return isClient }
// NewDiscoverer 创建一个新的发现服务实例
func NewDiscoverer() *Discoverer {
return &Discoverer{
Config: ConfigStruct{
Weight: 100,
CallRetryTimes: 10,
},
logger: log.DefaultLogger,
calls: make(map[string]*callInfoType),
appNodes: make(map[string]map[string]*NodeInfo),
appSubscribed: make(map[string]bool),
settedLoadBalancer: &DefaultLoadBalancer{},
}
func logError(error string, extra ...any) {
_logger.Error("Discover: "+error, append(extra, "app", Config.App, "addr", myAddr)...)
}
// IsServer 返回当前节点是否作为服务端运行
func (d *Discoverer) IsServer() bool { return d.isServer }
// IsClient 返回当前节点是否作为客户端运行
func (d *Discoverer) IsClient() bool { return d.isClient }
func (d *Discoverer) logError(msg string, extra ...any) {
d.logger.Error("Discover: "+msg, append(extra, "app", d.Config.App, "addr", d.myAddr)...)
func logInfo(info string, extra ...any) {
_logger.Info("Discover: "+info, append(extra, "app", Config.App, "addr", myAddr)...)
}
func (d *Discoverer) logInfo(msg string, extra ...any) {
d.logger.Info("Discover: "+msg, append(extra, "app", d.Config.App, "addr", d.myAddr)...)
func SetLogger(logger *log.Logger) {
_logger = logger
}
// SetLogger 设置 Discover 使用的全局 Logger
func (d *Discoverer) SetLogger(logger *log.Logger) {
d.logger = logger
}
// Init 初始化 Discover 配置
func (d *Discoverer) Init() {
d.appLock.Lock()
defer d.appLock.Unlock()
if d.inited {
func Init() {
appLock.Lock()
defer appLock.Unlock()
if _inited {
return
}
d.inited = true
// 如果是默认实例，尝试加载配置
if d == DefaultDiscoverer {
_ = config.Load(&d.Config, "discover")
Config = d.Config // 保持 Config 变量同步
_inited = true
_ = config.Load(&Config, "discover")
if Config.CallRetryTimes <= 0 {
Config.CallRetryTimes = 10
}
if Config.Weight <= 0 {
Config.Weight = 100
}
if Config.Registry == "" {
Config.Registry = DefaultRegistry
}
if d.Config.CallRetryTimes <= 0 {
d.Config.CallRetryTimes = 10
}
if d.Config.Weight <= 0 {
d.Config.Weight = 100
}
if d.Config.Registry == "" {
d.Config.Registry = DefaultRegistry
_logger = log.New(id.MakeID(12))
}
if d.logger == log.DefaultLogger || d.logger == nil {
d.logger = log.New(id.MakeID(12))
}
}
func Start(addr string) bool {
Init()
myAddr = addr
// Start 启动服务发现，指定当前节点的外部访问地址
func (d *Discoverer) Start(addr string) bool {
d.Init()
d.myAddr = addr
d.isServer = d.Config.App != "" && d.Config.Weight > 0
if d.isServer && d.Config.Registry != "" {
d.serverRedisPool = redis.GetRedis(d.Config.Registry, d.logger)
if d.serverRedisPool.Error != nil {
d.logError(d.serverRedisPool.Error.Error())
isServer = Config.App != "" && Config.Weight > 0
if isServer && Config.Registry != "" {
serverRedisPool = redis.GetRedis(Config.Registry, _logger)
if serverRedisPool.Error != nil {
logError(serverRedisPool.Error.Error())
}
// 注册节点
if d.serverRedisPool.Do("HSET", d.Config.App, addr, d.Config.Weight).Error == nil {
d.serverRedisPool.Do("SETEX", d.Config.App+"_"+addr, 10, "1")
d.logInfo("registered")
d.serverRedisPool.PUBLISH("CH_"+d.Config.App, fmt.Sprintf("%s %d", addr, d.Config.Weight))
d.daemonRunning = true
d.daemonStopChan = make(chan bool)
go d.daemon()
if serverRedisPool.Do("HSET", Config.App, addr, Config.Weight).Error == nil {
serverRedisPool.Do("SETEX", Config.App+"_"+addr, 10, "1")
logInfo("registered")
serverRedisPool.PUBLISH("CH_"+Config.App, fmt.Sprintf("%s %d", addr, Config.Weight))
daemonRunning = true
daemonStopChan = make(chan bool)
go daemon()
} else {
d.logError("register failed")
logError("register failed")
}
}
calls := d.getCalls()
calls := getCalls()
if len(calls) > 0 {
for app, conf := range calls {
d.addApp(app, conf, false)
addApp(app, conf, false)
}
if !d.startSub() {
if !startSub() {
return false
}
}
return true
}
func (d *Discoverer) daemon() {
d.logInfo("daemon thread started")
ticker := time.NewTicker(5 * time.Second)
func daemon() {
logInfo("daemon thread started")
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
for d.daemonRunning {
for daemonRunning {
<-ticker.C
if !d.daemonRunning {
if !daemonRunning {
break
}
if d.isServer && d.serverRedisPool != nil {
if !d.serverRedisPool.Do("HEXISTS", d.Config.App, d.myAddr).Bool() {
d.logInfo("lost app registered info, re-registering")
if d.serverRedisPool.Do("HSET", d.Config.App, d.myAddr, d.Config.Weight).Error == nil {
d.serverRedisPool.Do("SETEX", d.Config.App+"_"+d.myAddr, 10, "1")
d.serverRedisPool.PUBLISH("CH_"+d.Config.App, fmt.Sprintf("%s %d", d.myAddr, d.Config.Weight))
if isServer && serverRedisPool != nil {
if !serverRedisPool.Do("HEXISTS", Config.App, myAddr).Bool() {
logInfo("lost app registered info, re-registering")
if serverRedisPool.Do("HSET", Config.App, myAddr, Config.Weight).Error == nil {
serverRedisPool.Do("SETEX", Config.App+"_"+myAddr, 10, "1")
serverRedisPool.PUBLISH("CH_"+Config.App, fmt.Sprintf("%s %d", myAddr, Config.Weight))
}
} else {
d.serverRedisPool.Do("SETEX", d.Config.App+"_"+d.myAddr, 10, "1")
serverRedisPool.Do("SETEX", Config.App+"_"+myAddr, 10, "1")
}
}
}
d.logInfo("daemon thread stopped")
if d.daemonStopChan != nil {
d.daemonStopChan <- true
logInfo("daemon thread stopped")
if daemonStopChan != nil {
daemonStopChan <- true
}
}
func (d *Discoverer) startSub() bool {
if d.Config.Registry == "" {
func startSub() bool {
if Config.Registry == "" {
return true
}
d.appLock.Lock()
if d.clientRedisPool == nil {
d.clientRedisPool = redis.GetRedis(d.Config.Registry, d.logger)
appLock.Lock()
if clientRedisPool == nil {
clientRedisPool = redis.GetRedis(Config.Registry, _logger)
}
if d.pubsubRedisPool == nil {
d.pubsubRedisPool = redis.GetRedis(d.Config.Registry, d.logger.New(id.MakeID(12)))
if pubsubRedisPool == nil {
pubsubRedisPool = redis.GetRedis(Config.Registry, _logger.New(id.MakeID(12)))
// 订阅所有已注册的应用
for app := range d.appSubscribed {
d.subscribeAppUnderLock(app)
for app := range appSubscribed {
subscribeAppUnderLock(app)
}
// 必须在释放锁之前完成配置，但在释放锁之后启动，避免死锁
d.appLock.Unlock()
d.pubsubRedisPool.Start()
d.appLock.Lock()
appLock.Unlock()
pubsubRedisPool.Start()
appLock.Lock()
}
d.isClient = true
d.appLock.Unlock()
isClient = true
appLock.Unlock()
return true
}
func (d *Discoverer) subscribeAppUnderLock(app string) {
d.pubsubRedisPool.Subscribe("CH_"+app, func() {
d.fetchApp(app)
func subscribeAppUnderLock(app string) {
pubsubRedisPool.Subscribe("CH_"+app, func() {
fetchApp(app)
}, func(data []byte) {
a := strings.Split(string(data), " ")
addr := a[0]
@ -219,42 +186,39 @@ func (d *Discoverer) subscribeAppUnderLock(app string) {
if len(a) == 2 {
weight = cast.Int(a[1])
}
d.logInfo("received node update", "app", app, "addr", addr, "weight", weight)
d.pushNode(app, addr, weight)
logInfo("received node update", "app", app, "addr", addr, "weight", weight)
pushNode(app, addr, weight)
})
}
// Stop 停止 Discover 并从注册中心注销当前节点
func (d *Discoverer) Stop() {
d.appLock.Lock()
if d.isClient && d.pubsubRedisPool != nil {
d.pubsubRedisPool.Stop()
d.isClient = false
func Stop() {
appLock.Lock()
if isClient && pubsubRedisPool != nil {
pubsubRedisPool.Stop()
isClient = false
}
if d.isServer {
d.daemonRunning = false
if d.serverRedisPool != nil {
d.serverRedisPool.Do("HDEL", d.Config.App, d.myAddr)
d.serverRedisPool.Do("DEL", d.Config.App+"_"+d.myAddr)
d.serverRedisPool.PUBLISH("CH_"+d.Config.App, fmt.Sprintf("%s %d", d.myAddr, 0))
if isServer {
daemonRunning = false
if serverRedisPool != nil {
serverRedisPool.Do("HDEL", Config.App, myAddr)
serverRedisPool.Do("DEL", Config.App+"_"+myAddr)
serverRedisPool.PUBLISH("CH_"+Config.App, fmt.Sprintf("%s %d", myAddr, 0))
}
d.isServer = false
isServer = false
}
d.appLock.Unlock()
appLock.Unlock()
}
// Wait 等待守护进程退出
func (d *Discoverer) Wait() {
if d.daemonStopChan != nil {
<-d.daemonStopChan
d.daemonStopChan = nil
func Wait() {
if daemonStopChan != nil {
<-daemonStopChan
daemonStopChan = nil
}
}
// EasyStart 自动根据环境变量和本地网卡信息启动 Discover
func (d *Discoverer) EasyStart() (string, int) {
d.Init()
func EasyStart() (string, int) {
Init()
port := 0
if listen := os.Getenv("DISCOVER_LISTEN"); listen != "" {
if _, p, err := net.SplitHostPort(listen); err == nil {
@ -266,7 +230,7 @@ func (d *Discoverer) EasyStart() (string, int) {
ln, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
if err != nil {
d.logError("failed to listen", "err", err)
logError("failed to listen", "err", err)
return "", 0
}
addrInfo := ln.Addr().(*net.TCPAddr)
@ -282,7 +246,7 @@ func (d *Discoverer) EasyStart() (string, int) {
if ip4 == nil || !ip4.IsGlobalUnicast() {
continue
}
if d.Config.IpPrefix != "" && strings.HasPrefix(ip4.String(), d.Config.IpPrefix) {
if Config.IpPrefix != "" && strings.HasPrefix(ip4.String(), Config.IpPrefix) {
ip = ip4
break
}
@ -294,7 +258,7 @@ func (d *Discoverer) EasyStart() (string, int) {
}
addr := fmt.Sprintf("%s:%d", ip.String(), port)
if !d.Start(addr) {
if !Start(addr) {
return "", 0
}
@ -302,51 +266,48 @@ func (d *Discoverer) EasyStart() (string, int) {
signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
go func() {
<-sigChan
d.Stop()
Stop()
}()
return ip.String(), port
}
// AddExternalApp 动态添加需要发现的外部应用
func (d *Discoverer) AddExternalApp(app, callConf string) bool {
if d.addApp(app, callConf, true) {
if !d.isClient {
d.startSub()
func AddExternalApp(app, callConf string) bool {
if addApp(app, callConf, true) {
if !isClient {
startSub()
} else {
d.appLock.Lock()
d.subscribeAppUnderLock(app)
d.appLock.Unlock()
d.fetchApp(app) // 同步拉取一次
appLock.Lock()
subscribeAppUnderLock(app)
appLock.Unlock()
}
return true
}
return false
}
// SetNode 手动设置某个服务的节点信息
func (d *Discoverer) SetNode(app, addr string, weight int) {
d.pushNode(app, addr, weight)
func SetNode(app, addr string, weight int) {
pushNode(app, addr, weight)
}
func (d *Discoverer) getCallInfo(app string) *callInfoType {
d.appLock.RLock()
defer d.appLock.RUnlock()
return d.calls[app]
func getCallInfo(app string) *callInfoType {
appLock.RLock()
defer appLock.RUnlock()
return _calls[app]
}
var numberMatcher = regexp.MustCompile(`^\d+(s|ms|us|µs|ns?)?$`)
func (d *Discoverer) addApp(app, callConf string, fetch bool) bool {
d.appLock.Lock()
if d.Config.Calls == nil {
d.Config.Calls = make(map[string]string)
func addApp(app, callConf string, fetch bool) bool {
appLock.Lock()
if Config.Calls == nil {
Config.Calls = make(map[string]string)
}
if d.Config.Calls[app] == callConf && d.appNodes[app] != nil {
d.appLock.Unlock()
if Config.Calls[app] == callConf && _appNodes[app] != nil {
appLock.Unlock()
return false
}
d.Config.Calls[app] = callConf
Config.Calls[app] = callConf
callInfo := &callInfoType{
Timeout: 10 * time.Second,
@ -378,23 +339,23 @@ func (d *Discoverer) addApp(app, callConf string, fetch bool) bool {
}
}
d.calls[app] = callInfo
if d.appNodes[app] == nil {
d.appNodes[app] = make(map[string]*NodeInfo)
_calls[app] = callInfo
if _appNodes[app] == nil {
_appNodes[app] = make(map[string]*NodeInfo)
}
d.appSubscribed[app] = true
d.appLock.Unlock()
appSubscribed[app] = true
appLock.Unlock()
if fetch && d.isClient {
d.fetchApp(app)
if fetch {
fetchApp(app)
}
return true
}
func (d *Discoverer) fetchApp(app string) {
d.appLock.RLock()
pool := d.clientRedisPool
d.appLock.RUnlock()
func fetchApp(app string) {
appLock.RLock()
pool := clientRedisPool
appLock.RUnlock()
if pool == nil {
return
}
@ -409,132 +370,80 @@ func (d *Discoverer) fetchApp(app string) {
}
}
currentNodes := d.getAppNodes(app)
currentNodes := getAppNodes(app)
if currentNodes != nil {
for addr := range currentNodes {
if _, ok := results[addr]; !ok {
d.pushNode(app, addr, 0)
pushNode(app, addr, 0)
}
}
}
for addr, res := range results {
d.pushNode(app, addr, res.Int())
pushNode(app, addr, res.Int())
}
}
func (d *Discoverer) getAppNodes(app string) map[string]*NodeInfo {
d.appLock.RLock()
defer d.appLock.RUnlock()
if d.appNodes[app] == nil {
func getAppNodes(app string) map[string]*NodeInfo {
appLock.RLock()
defer appLock.RUnlock()
if _appNodes[app] == nil {
return nil
}
nodes := make(map[string]*NodeInfo)
for k, v := range d.appNodes[app] {
for k, v := range _appNodes[app] {
nodes[k] = v
}
return nodes
}
func (d *Discoverer) getCalls() map[string]string {
d.appLock.RLock()
defer d.appLock.RUnlock()
func getCalls() map[string]string {
appLock.RLock()
defer appLock.RUnlock()
calls := make(map[string]string)
for k, v := range d.Config.Calls {
for k, v := range Config.Calls {
calls[k] = v
}
return calls
}
// GetAppNodes 获取某个应用的所有节点列表
func (d *Discoverer) GetAppNodes(app string) map[string]*NodeInfo {
return d.getAppNodes(app)
func GetAppNodes(app string) map[string]*NodeInfo {
return getAppNodes(app)
}
func (d *Discoverer) pushNode(app, addr string, weight int) {
d.appLock.Lock()
defer d.appLock.Unlock()
func pushNode(app, addr string, weight int) {
appLock.Lock()
defer appLock.Unlock()
if weight <= 0 {
if d.appNodes[app] != nil {
delete(d.appNodes[app], addr)
if _appNodes[app] != nil {
delete(_appNodes[app], addr)
}
return
}
if d.appNodes[app] == nil {
d.appNodes[app] = make(map[string]*NodeInfo)
if _appNodes[app] == nil {
_appNodes[app] = make(map[string]*NodeInfo)
}
if node, ok := d.appNodes[app][addr]; ok {
if node, ok := _appNodes[app][addr]; ok {
if node.Weight != weight {
used := node.UsedTimes.Load()
node.UsedTimes.Store(uint64(float64(used) / float64(node.Weight) * float64(weight)))
node.UsedTimes = uint64(float64(node.UsedTimes) / float64(node.Weight) * float64(weight))
node.Weight = weight
}
} else {
var avgUsed uint64 = 0
if len(d.appNodes[app]) > 0 {
if len(_appNodes[app]) > 0 {
var totalScore float64
for _, n := range d.appNodes[app] {
totalScore += float64(n.UsedTimes.Load()) / float64(n.Weight)
for _, n := range _appNodes[app] {
totalScore += float64(n.UsedTimes) / float64(n.Weight)
}
avgUsed = uint64(totalScore / float64(len(d.appNodes[app])) * float64(weight))
avgUsed = uint64(totalScore / float64(len(_appNodes[app])) * float64(weight))
}
node := &NodeInfo{
_appNodes[app][addr] = &NodeInfo{
Addr: addr,
Weight: weight,
}
node.UsedTimes.Store(avgUsed)
d.appNodes[app][addr] = node
UsedTimes: avgUsed,
}
}
// 以下是包级别 API，通过转发给 DefaultDiscoverer 实现兼容性
func IsServer() bool { return DefaultDiscoverer.IsServer() }
func IsClient() bool { return DefaultDiscoverer.IsClient() }
func logError(msg string, extra ...any) {
DefaultDiscoverer.logError(msg, extra...)
}
func logInfo(msg string, extra ...any) {
DefaultDiscoverer.logInfo(msg, extra...)
}
func SetLogger(logger *log.Logger) {
DefaultDiscoverer.SetLogger(logger)
}
func Init() {
DefaultDiscoverer.Init()
}
func Start(addr string) bool {
return DefaultDiscoverer.Start(addr)
}
func Stop() {
DefaultDiscoverer.Stop()
}
func Wait() {
DefaultDiscoverer.Wait()
}
func EasyStart() (string, int) {
return DefaultDiscoverer.EasyStart()
}
func AddExternalApp(app, callConf string) bool {
return DefaultDiscoverer.AddExternalApp(app, callConf)
}
func SetNode(app, addr string, weight int) {
DefaultDiscoverer.SetNode(app, addr, weight)
}
func GetAppNodes(app string) map[string]*NodeInfo {
return DefaultDiscoverer.GetAppNodes(app)
}

25
Discover_test.go
View File

@ -46,8 +46,8 @@ func TestDiscover(t *testing.T) {
defer server.Close()
// 配置 Discover
discover.DefaultDiscoverer.Config.App = "test-app"
discover.DefaultDiscoverer.Config.Registry = "redis://127.0.0.1:6379/15"
discover.Config.App = "test-app"
discover.Config.Registry = "redis://127.0.0.1:6379/15"
// 启动 Discover
if !discover.Start("127.0.0.1:18001") {
@ -138,24 +138,3 @@ func TestEasyStart(t *testing.T) {
fmt.Printf("EasyStart: %s:%d\n", ip, port)
discover.Stop()
}
func BenchmarkDiscover(b *testing.B) {
discover.Config.App = "bench-app"
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 := discover.AppClient{
App: "bench-app",
Method: "GET",
Path: "/",
}
node := appClient.Next("bench-app", nil)
if node == nil {
b.Fatal("no node")
}
}
}

18
LoadBalancer.go
View File

@ -7,12 +7,7 @@ import (
// SetLoadBalancer 设置全局负载均衡策略
func SetLoadBalancer(lb LoadBalancer) {
DefaultDiscoverer.SetLoadBalancer(lb)
}
// SetLoadBalancer 设置负载均衡策略
func (d *Discoverer) SetLoadBalancer(lb LoadBalancer) {
d.settedLoadBalancer = lb
settedLoadBalancer = lb
}
// LoadBalancer 负载均衡接口
@ -27,17 +22,20 @@ type LoadBalancer interface {
// DefaultLoadBalancer 默认负载均衡器（简单权重轮询/得分最小者优先）
type DefaultLoadBalancer struct{}
// Response 在默认负载均衡器中不再执行写操作，减少锁竞争
func (lb *DefaultLoadBalancer) Response(appClient *AppClient, node *NodeInfo, err error, response *http.Response, responseTime time.Duration) {
node.Data.Store("score", float64(node.UsedTimes)/float64(node.Weight))
}
// Next 根据得分（UsedTimes / Weight）选择得分最小的节点
func (lb *DefaultLoadBalancer) Next(appClient *AppClient, nodes []*NodeInfo, request *http.Request) *NodeInfo {
var minScore float64 = -1
var minNode *NodeInfo
for _, node := range nodes {
// 动态计算得分，避免使用 sync.Map 存储，减少内存分配和锁竞争
score := float64(node.UsedTimes.Load()) / float64(node.Weight)
scoreValue, ok := node.Data.Load("score")
if !ok {
scoreValue = float64(node.UsedTimes) / float64(node.Weight)
node.Data.Store("score", scoreValue)
}
score := scoreValue.(float64)
if minNode == nil || score < minScore {
minScore = score
minNode = node

40
Log.go
View File

@ -1,40 +0,0 @@
package discover
import (
"apigo.cc/go/log"
)
const LogTypeDiscover = "discover"
type DiscoverLog struct {
log.BaseLog
App string
Method string
Path string
Node string
Attempts int
UsedTime float32
Error string
}
func (ac *AppClient) Log(node string, usedTime float32, err error) {
if ac.Logger == nil {
ac.Logger = log.DefaultLogger
}
if !ac.Logger.CheckLevel(log.INFO) {
return
}
entry := log.GetEntry[DiscoverLog]()
// 框架会自动调用 FillBase，不再手动调用
entry.App = ac.App
entry.Method = ac.Method
entry.Path = ac.Path
entry.Node = node
entry.Attempts = ac.attempts
entry.UsedTime = usedTime
if err != nil {
entry.Error = err.Error()
}
ac.Logger.Log(entry)
}

74
MultiInstance_test.go
View File

@ -1,74 +0,0 @@
package discover_test
import (
"fmt"
"net"
"net/http"
"testing"
"time"
"apigo.cc/go/discover"
)
func TestMultipleDiscoverer(t *testing.T) {
// 启动两个模拟服务
l1, _ := net.Listen("tcp", "127.0.0.1:18011")
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")
mux2 := http.NewServeMux()
mux2.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("OK2")) })
server2 := &http.Server{Handler: mux2}
go func() { _ = server2.Serve(l2) }()
defer server2.Close()
registry := "redis://127.0.0.1:6379/15"
// 实例 1
d1 := discover.NewDiscoverer()
d1.Config.App = "app1"
d1.Config.Registry = registry
if !d1.Start("127.0.0.1:18011") {
t.Skip("redis not available")
}
defer d1.Stop()
// 实例 2
d2 := discover.NewDiscoverer()
d2.Config.App = "app2"
d2.Config.Registry = registry
if !d2.Start("127.0.0.1:18012") {
t.Skip("redis not available")
}
defer d2.Stop()
// 实例 1 发现并调用自己
d1.AddExternalApp("app1", "1")
time.Sleep(200 * time.Millisecond) // 等待同步
c1 := d1.NewCaller(nil, nil)
res1 := c1.Get("app1", "/")
if res1.Error != nil || res1.String() != "OK1" {
t.Errorf("d1 call app1 failed: %v, %s", res1.Error, res1.String())
}
// 实例 2 发现并调用 实例 1
d2.AddExternalApp("app1", "1")
time.Sleep(200 * time.Millisecond) // 等待同步
c2 := d2.NewCaller(nil, nil)
res2 := c2.Get("app1", "/")
if res2.Error != nil || res2.String() != "OK1" {
t.Errorf("d2 call app1 failed: %v, %s", res2.Error, res2.String())
}
// 验证独立性：d1 不应该能直接调用 app2 (除非手动 AddExternalApp)
res3 := c1.Get("app2", "/")
if res3.Error == nil {
t.Error("d1 should not find app2 without AddExternalApp")
}
fmt.Println("Multiple Discoverer instances verified")
}

5
NodeInfo.go
View File

@ -2,14 +2,13 @@ package discover
import (
"sync"
"sync/atomic"
)
// NodeInfo 存储服务节点信息
type NodeInfo struct {
Addr string // 节点地址
Weight int // 节点权重
UsedTimes atomic.Uint64 // 已使用次数
FailedTimes atomic.Int32 // 失败次数
UsedTimes uint64 // 已使用次数
FailedTimes int // 失败次数
Data sync.Map // 运行时自定义数据
}

9
Route.go
View File

@ -2,12 +2,7 @@ package discover
import "net/http"
// SetRoute 设置全局路由规则
// SetRoute 设置全局路由规则，可以在请求前修改 App、Method、Path 等信息
func SetRoute(route func(appClient *AppClient, request *http.Request)) {
DefaultDiscoverer.SetRoute(route)
}
// SetRoute 设置路由规则
func (d *Discoverer) SetRoute(route func(appClient *AppClient, request *http.Request)) {
d.settedRoute = route
settedRoute = route
}

4
TEST.md
View File

@ -5,7 +5,6 @@
2. **实时同步**: 验证通过 Redis PUBLISH 更新节点信息后，客户端能实时感知并更新本地节点列表。
3. **故障剔除**: 验证当节点调用持续失败时，能自动从本地列表中剔除。
4. **环境变量配置**: 验证 `EasyStart` 结合环境变量的启动流程。
5. **高效日志记录**: 验证 `DiscoverLog` 通过对象池和 `FillBase` 机制实现的高性能异步日志。
## 测试结果
- **Unit Tests**: `go test -v ./...`
@ -13,5 +12,4 @@
- `TestEasyStart`: PASS
## Benchmark
- `BenchmarkDiscover`: ~560 ns/op (Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz)
- 负载均衡选择节点耗时极低，适合高并发场景。
- 待补充（Discover 主要性能开销在负载均衡算法选择，单次选择耗时极低）。

2
go.mod
View File

@ -7,7 +7,7 @@ require (
apigo.cc/go/config v1.0.4
apigo.cc/go/http v1.0.3
apigo.cc/go/id v1.0.4
apigo.cc/go/log v1.1.0
apigo.cc/go/log v1.0.2
apigo.cc/go/redis v1.0.2
github.com/gorilla/websocket v1.5.3
)