Compare commits
12 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5de04b4a63 | ||
|
|
18fdafe09f | ||
|
|
ab4f5b6885 | ||
|
|
7d5e6b00e3 | ||
|
|
5d8dcc65c0 | ||
|
|
b1fcba1a42 | ||
|
|
fb0f9167b5 | ||
|
|
0ae27f613a | ||
|
|
ae0cd90904 | ||
|
|
5eefc06bba | ||
|
|
4b47f31c80 | ||
|
|
20c1f3da78 |
1
.gitignore
vendored
Normal file
1
.gitignore
vendored
Normal file
@ -0,0 +1 @@
|
|||||||
|
.log.meta.json
|
||||||
48
AppClient.go
48
AppClient.go
@ -8,58 +8,68 @@ import (
|
|||||||
|
|
||||||
// AppClient 用于管理单个请求的重试和负载均衡状态
|
// AppClient 用于管理单个请求的重试和负载均衡状态
|
||||||
type AppClient struct {
|
type AppClient struct {
|
||||||
excludes map[string]bool
|
discoverer *Discoverer
|
||||||
tryTimes int
|
excludes map[string]bool // 本次请求已排除的节点
|
||||||
Logger *log.Logger
|
attempts int // 本次请求的重试次数
|
||||||
App string
|
Logger *log.Logger // 用于日志记录的 Logger
|
||||||
Method string
|
App string // 目标应用名称
|
||||||
Path string
|
Method string // 请求方法
|
||||||
Data *map[string]any
|
Path string // 请求路径
|
||||||
Headers *map[string]string
|
Data map[string]any // 请求数据
|
||||||
|
Headers map[string]string // 请求头
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ac *AppClient) logError(error string, extra ...any) {
|
// logError 记录 Discover 客户端错误
|
||||||
|
func (ac *AppClient) logError(msg string, extra ...any) {
|
||||||
if ac.Logger == nil {
|
if ac.Logger == nil {
|
||||||
ac.Logger = log.DefaultLogger
|
ac.Logger = log.DefaultLogger
|
||||||
}
|
}
|
||||||
ac.Logger.Error("Discover Client: "+error, extra...)
|
ac.Logger.Error("Discover Client: "+msg, extra...)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Next 获取下一个可用节点
|
||||||
func (ac *AppClient) Next(app string, request *http.Request) *NodeInfo {
|
func (ac *AppClient) Next(app string, request *http.Request) *NodeInfo {
|
||||||
|
if ac.discoverer == nil {
|
||||||
|
ac.discoverer = DefaultDiscoverer
|
||||||
|
}
|
||||||
return ac.NextWithNode(app, "", request)
|
return ac.NextWithNode(app, "", request)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// CheckApp 检查并尝试添加应用
|
||||||
func (ac *AppClient) CheckApp(app string) bool {
|
func (ac *AppClient) CheckApp(app string) bool {
|
||||||
nodes := getAppNodes(app)
|
nodes := ac.discoverer.GetAppNodes(app)
|
||||||
if nodes == nil {
|
if nodes == nil {
|
||||||
if !addApp(app, "", true) {
|
conf := ac.discoverer.GetConfig()
|
||||||
ac.logError("app not found", "app", app, "calls", Config.Calls)
|
if !ac.discoverer.AddExternalApp(app, "") {
|
||||||
|
ac.logError("app not found", "app", app, "calls", conf.Calls)
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// NextWithNode 获取下一个可用节点,支持指定节点
|
||||||
func (ac *AppClient) NextWithNode(app, withNode string, request *http.Request) *NodeInfo {
|
func (ac *AppClient) NextWithNode(app, withNode string, request *http.Request) *NodeInfo {
|
||||||
if ac.excludes == nil {
|
if ac.excludes == nil {
|
||||||
ac.excludes = make(map[string]bool)
|
ac.excludes = make(map[string]bool)
|
||||||
}
|
}
|
||||||
|
|
||||||
allNodes := getAppNodes(app)
|
allNodes := ac.discoverer.GetAppNodes(app)
|
||||||
if len(allNodes) == 0 {
|
if len(allNodes) == 0 {
|
||||||
ac.logError("node not found", "app", app)
|
ac.logError("node not found", "app", app)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
ac.tryTimes++
|
ac.attempts++
|
||||||
if withNode != "" {
|
if withNode != "" {
|
||||||
ac.excludes[withNode] = true
|
ac.excludes[withNode] = true
|
||||||
return allNodes[withNode]
|
return allNodes[withNode]
|
||||||
}
|
}
|
||||||
|
|
||||||
readyNodes := make([]*NodeInfo, 0)
|
conf := ac.discoverer.GetConfig()
|
||||||
|
readyNodes := make([]*NodeInfo, 0, len(allNodes))
|
||||||
for _, node := range allNodes {
|
for _, node := range allNodes {
|
||||||
if ac.excludes[node.Addr] || node.FailedTimes >= Config.CallRetryTimes {
|
if ac.excludes[node.Addr] || node.FailedTimes.Load() >= int32(conf.CallRetryTimes) {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
readyNodes = append(readyNodes, node)
|
readyNodes = append(readyNodes, node)
|
||||||
@ -76,14 +86,14 @@ func (ac *AppClient) NextWithNode(app, withNode string, request *http.Request) *
|
|||||||
|
|
||||||
var node *NodeInfo
|
var node *NodeInfo
|
||||||
if len(readyNodes) > 0 {
|
if len(readyNodes) > 0 {
|
||||||
node = settedLoadBalancer.Next(ac, readyNodes, request)
|
node = ac.discoverer.settedLoadBalancer.Next(ac, readyNodes, request)
|
||||||
if node != nil {
|
if node != nil {
|
||||||
ac.excludes[node.Addr] = true
|
ac.excludes[node.Addr] = true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if node == nil {
|
if node == nil {
|
||||||
ac.logError("no available node", "app", app, "tryTimes", ac.tryTimes)
|
ac.logError("no available node", "app", app, "attempts", ac.attempts)
|
||||||
}
|
}
|
||||||
|
|
||||||
return node
|
return node
|
||||||
|
|||||||
35
CHANGELOG.md
35
CHANGELOG.md
@ -1,9 +1,32 @@
|
|||||||
# CHANGELOG
|
# 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` 变量,保证业务代码通过不同路径读取配置的一致性。
|
||||||
|
|
||||||
|
## v1.0.3 (2026-05-05)
|
||||||
|
- 架构深度优化:将 HTTP 客户端连接池(`appClientPools`)移入 `Discoverer` 实例,实现完全的资源隔离。
|
||||||
|
- 并发安全增强:引入读写锁保护 `Config` 结构,防止高并发下的配置读写冲突。
|
||||||
|
- 生命周期管理优化:使用 `atomic.Bool` 管理 `daemonRunning` 状态,确保线程安全。
|
||||||
|
- 资源回收机制:在 `Stop()` 方法中新增 HTTP 连接池清理逻辑(调用 `Destroy` 释放闲置连接),防止内存与句柄泄漏。
|
||||||
|
- 接口严谨性:将 `Discoverer` 内部字段(如 `config`)设为私有,通过 `GetConfig`/`SetConfig` 统一访问。
|
||||||
|
|
||||||
|
## v1.0.2
|
||||||
|
- 架构重构:支持多 Discoverer 实例,消灭包级全局状态。
|
||||||
|
- 兼容性:保留包级 API 转发至 `DefaultDiscoverer`。
|
||||||
|
|
||||||
|
## v1.0.1
|
||||||
|
- 优化代码规范:修复变量名冲突,改进命名语义。
|
||||||
|
- 性能优化:优化 `AppClient` 类型,减少寻址开销。
|
||||||
|
- 故障隔离:实现本地隔离机制,不再篡跨全局 Redis 状态。
|
||||||
|
- 压力缓解:心跳间隔优化至 5 秒。
|
||||||
|
|
||||||
## v1.0.0
|
## v1.0.0
|
||||||
- 从 `ssgo/discover` 迁移至 `apigo.cc/go/discover`。
|
- 初始版本:从 `ssgo/discover` 迁移并重构。
|
||||||
- 采用全新的 `apigo.cc/go` 基础设施(log, redis, http, cast, u)。
|
|
||||||
- 优化了注册中心同步机制,使用 `redis.Subscribe` 简化 PubSub 处理。
|
|
||||||
- 增强了负载均衡算法,引入更精确的得分计算。
|
|
||||||
- 统一了 Header 定义,对齐 `go/http` 标准。
|
|
||||||
- 移除所有 `panic`,通过 `error` 返回和日志记录确保系统稳定性。
|
|
||||||
|
|||||||
120
Caller.go
120
Caller.go
@ -5,7 +5,6 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
"reflect"
|
"reflect"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/gorilla/websocket"
|
"github.com/gorilla/websocket"
|
||||||
@ -14,20 +13,17 @@ import (
|
|||||||
"apigo.cc/go/log"
|
"apigo.cc/go/log"
|
||||||
)
|
)
|
||||||
|
|
||||||
var appClientPools = make(map[string]*gohttp.Client)
|
func (d *Discoverer) getHttpClient(app string, timeout time.Duration, h2c bool) *gohttp.Client {
|
||||||
var appClientPoolsLock sync.RWMutex
|
d.appClientPoolsLock.RLock()
|
||||||
|
c := d.appClientPools[app]
|
||||||
func getHttpClient(app string, timeout time.Duration, h2c bool) *gohttp.Client {
|
d.appClientPoolsLock.RUnlock()
|
||||||
appClientPoolsLock.RLock()
|
|
||||||
c := appClientPools[app]
|
|
||||||
appClientPoolsLock.RUnlock()
|
|
||||||
if c != nil {
|
if c != nil {
|
||||||
return c
|
return c
|
||||||
}
|
}
|
||||||
|
|
||||||
appClientPoolsLock.Lock()
|
d.appClientPoolsLock.Lock()
|
||||||
defer appClientPoolsLock.Unlock()
|
defer d.appClientPoolsLock.Unlock()
|
||||||
c = appClientPools[app]
|
c = d.appClientPools[app]
|
||||||
if c != nil {
|
if c != nil {
|
||||||
return c
|
return c
|
||||||
}
|
}
|
||||||
@ -37,52 +33,84 @@ func getHttpClient(app string, timeout time.Duration, h2c bool) *gohttp.Client {
|
|||||||
} else {
|
} else {
|
||||||
c = gohttp.NewClient(timeout)
|
c = gohttp.NewClient(timeout)
|
||||||
}
|
}
|
||||||
appClientPools[app] = c
|
d.appClientPools[app] = c
|
||||||
return c
|
return c
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Caller 用于发起服务间调用
|
||||||
type Caller struct {
|
type Caller struct {
|
||||||
Request *http.Request
|
discoverer *Discoverer
|
||||||
NoBody bool
|
Request *http.Request // 原始请求,用于透传 Header
|
||||||
logger *log.Logger
|
NoBody bool // 是否不发送请求体
|
||||||
|
logger *log.Logger // 用于日志记录的 Logger
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// NewCaller 创建一个新的调用器
|
||||||
func NewCaller(request *http.Request, logger *log.Logger) *Caller {
|
func NewCaller(request *http.Request, logger *log.Logger) *Caller {
|
||||||
return &Caller{Request: request, logger: logger}
|
return DefaultDiscoverer.NewCaller(request, logger)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Caller) logError(error string, extra ...any) {
|
// 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) {
|
||||||
if c.logger == nil {
|
if c.logger == nil {
|
||||||
c.logger = log.DefaultLogger
|
c.logger = log.DefaultLogger
|
||||||
}
|
}
|
||||||
c.logger.Error("Discover Caller: "+error, extra...)
|
c.logger.Error("Discover Caller: "+msg, extra...)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Get 发起 GET 请求
|
||||||
func (c *Caller) Get(app, path string, headers ...string) *gohttp.Result {
|
func (c *Caller) Get(app, path string, headers ...string) *gohttp.Result {
|
||||||
return c.Do("GET", app, path, nil, headers...)
|
return c.Do("GET", app, path, nil, headers...)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Post 发起 POST 请求
|
||||||
func (c *Caller) Post(app, path string, data any, headers ...string) *gohttp.Result {
|
func (c *Caller) Post(app, path string, data any, headers ...string) *gohttp.Result {
|
||||||
return c.Do("POST", app, path, data, headers...)
|
return c.Do("POST", app, path, data, headers...)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Put 发起 PUT 请求
|
||||||
func (c *Caller) Put(app, path string, data any, headers ...string) *gohttp.Result {
|
func (c *Caller) Put(app, path string, data any, headers ...string) *gohttp.Result {
|
||||||
return c.Do("PUT", app, path, data, headers...)
|
return c.Do("PUT", app, path, data, headers...)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Delete 发起 DELETE 请求
|
||||||
func (c *Caller) Delete(app, path string, data any, headers ...string) *gohttp.Result {
|
func (c *Caller) Delete(app, path string, data any, headers ...string) *gohttp.Result {
|
||||||
return c.Do("DELETE", app, path, data, headers...)
|
return c.Do("DELETE", app, path, data, headers...)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Head 发起 HEAD 请求
|
||||||
func (c *Caller) Head(app, path string, headers ...string) *gohttp.Result {
|
func (c *Caller) Head(app, path string, headers ...string) *gohttp.Result {
|
||||||
return c.Do("HEAD", app, path, nil, headers...)
|
return c.Do("HEAD", app, path, nil, headers...)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Call 发起通用的泛型请求并自动解析响应
|
||||||
|
func Call[T any](method, app, path string, data any, headers ...string) (T, error) {
|
||||||
|
return CallT[T](DefaultDiscoverer.NewCaller(nil, nil), method, app, path, data, headers...)
|
||||||
|
}
|
||||||
|
|
||||||
|
// CallT 发起泛型请求并自动解析响应 (由于 Go 方法不支持泛型,故使用函数)
|
||||||
|
func CallT[T any](c *Caller, method, app, path string, data any, headers ...string) (T, error) {
|
||||||
|
var result T
|
||||||
|
res := c.Do(method, app, path, data, headers...)
|
||||||
|
if res.Error != nil {
|
||||||
|
return result, res.Error
|
||||||
|
}
|
||||||
|
err := res.To(&result)
|
||||||
|
return result, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Do 发起通用请求
|
||||||
func (c *Caller) Do(method, app, path string, data any, headers ...string) *gohttp.Result {
|
func (c *Caller) Do(method, app, path string, data any, headers ...string) *gohttp.Result {
|
||||||
r, _ := c.DoWithNode(method, app, "", path, data, headers...)
|
r, _ := c.DoWithNode(method, app, "", path, data, headers...)
|
||||||
return r
|
return r
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Open 发起 WebSocket 连接
|
||||||
func (c *Caller) Open(app, path string, headers ...string) *websocket.Conn {
|
func (c *Caller) Open(app, path string, headers ...string) *websocket.Conn {
|
||||||
r, _ := c.doWithNode(false, "WS", app, "", path, nil, headers...)
|
r, _ := c.doWithNode(false, "WS", app, "", path, nil, headers...)
|
||||||
if v, ok := r.(*websocket.Conn); ok {
|
if v, ok := r.(*websocket.Conn); ok {
|
||||||
@ -91,6 +119,7 @@ func (c *Caller) Open(app, path string, headers ...string) *websocket.Conn {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DoWithNode 发起请求并返回结果及节点地址
|
||||||
func (c *Caller) DoWithNode(method, app, withNode, path string, data any, headers ...string) (*gohttp.Result, string) {
|
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...)
|
r, nodeAddr := c.doWithNode(false, method, app, withNode, path, data, headers...)
|
||||||
if v, ok := r.(*gohttp.Result); ok {
|
if v, ok := r.(*gohttp.Result); ok {
|
||||||
@ -99,6 +128,7 @@ func (c *Caller) DoWithNode(method, app, withNode, path string, data any, header
|
|||||||
return nil, nodeAddr
|
return nil, nodeAddr
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ManualDoWithNode 发起请求(手动处理响应)并返回结果及节点地址
|
||||||
func (c *Caller) ManualDoWithNode(method, app, withNode, path string, data any, headers ...string) (*gohttp.Result, string) {
|
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...)
|
r, nodeAddr := c.doWithNode(true, method, app, withNode, path, data, headers...)
|
||||||
if v, ok := r.(*gohttp.Result); ok {
|
if v, ok := r.(*gohttp.Result); ok {
|
||||||
@ -107,15 +137,16 @@ func (c *Caller) ManualDoWithNode(method, app, withNode, path string, data any,
|
|||||||
return nil, nodeAddr
|
return nil, nodeAddr
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Caller) doWithNode(manualDo bool, method, app, withNode, path string, data any, headers ...string) (any, string) {
|
func (c *Caller) doWithNode(manual bool, method, app, withNode, path string, data any, headers ...string) (any, string) {
|
||||||
callerHeaders := make(map[string]string)
|
callerHeaders := make(map[string]string)
|
||||||
for i := 1; i < len(headers); i += 2 {
|
for i := 1; i < len(headers); i += 2 {
|
||||||
callerHeaders[headers[i-1]] = headers[i]
|
callerHeaders[headers[i-1]] = headers[i]
|
||||||
}
|
}
|
||||||
|
|
||||||
if isServer {
|
conf := c.discoverer.GetConfig()
|
||||||
callerHeaders[HeaderFromApp] = Config.App
|
if c.discoverer.isServer {
|
||||||
callerHeaders[HeaderFromNode] = myAddr
|
callerHeaders[HeaderFromApp] = conf.App
|
||||||
|
callerHeaders[HeaderFromNode] = c.discoverer.myAddr
|
||||||
}
|
}
|
||||||
|
|
||||||
callData := make(map[string]any)
|
callData := make(map[string]any)
|
||||||
@ -127,16 +158,17 @@ func (c *Caller) doWithNode(manualDo bool, method, app, withNode, path string, d
|
|||||||
}
|
}
|
||||||
|
|
||||||
appClient := AppClient{
|
appClient := AppClient{
|
||||||
Logger: c.logger,
|
discoverer: c.discoverer,
|
||||||
App: app,
|
Logger: c.logger,
|
||||||
Method: method,
|
App: app,
|
||||||
Path: path,
|
Method: method,
|
||||||
Data: &callData,
|
Path: path,
|
||||||
Headers: &callerHeaders,
|
Data: callData,
|
||||||
|
Headers: callerHeaders,
|
||||||
}
|
}
|
||||||
|
|
||||||
if settedRoute != nil {
|
if c.discoverer.settedRoute != nil {
|
||||||
settedRoute(&appClient, c.Request)
|
c.discoverer.settedRoute(&appClient, c.Request)
|
||||||
app = appClient.App
|
app = appClient.App
|
||||||
method = appClient.Method
|
method = appClient.Method
|
||||||
path = appClient.Path
|
path = appClient.Path
|
||||||
@ -146,7 +178,7 @@ func (c *Caller) doWithNode(manualDo bool, method, app, withNode, path string, d
|
|||||||
return &gohttp.Result{Error: fmt.Errorf("app %s not found", app)}, ""
|
return &gohttp.Result{Error: fmt.Errorf("app %s not found", app)}, ""
|
||||||
}
|
}
|
||||||
|
|
||||||
callInfo := getCallInfo(app)
|
callInfo := c.discoverer.getCallInfo(app)
|
||||||
if callInfo != nil && callInfo.Token != "" {
|
if callInfo != nil && callInfo.Token != "" {
|
||||||
callerHeaders["Access-Token"] = callInfo.Token
|
callerHeaders["Access-Token"] = callInfo.Token
|
||||||
}
|
}
|
||||||
@ -162,14 +194,14 @@ func (c *Caller) doWithNode(manualDo bool, method, app, withNode, path string, d
|
|||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
node.UsedTimes++
|
node.UsedTimes.Add(1)
|
||||||
startTime := time.Now()
|
startTime := time.Now()
|
||||||
scheme := "http"
|
scheme := "http"
|
||||||
if callInfo != nil && callInfo.SSL {
|
if callInfo != nil && callInfo.SSL {
|
||||||
scheme = "https"
|
scheme = "https"
|
||||||
}
|
}
|
||||||
|
|
||||||
hc := getHttpClient(app, callInfo.Timeout, callInfo.HttpVersion == 2 && !callInfo.SSL)
|
hc := c.discoverer.getHttpClient(app, callInfo.Timeout, callInfo.HttpVersion == 2 && !callInfo.SSL)
|
||||||
hc.NoBody = c.NoBody
|
hc.NoBody = c.NoBody
|
||||||
|
|
||||||
var res *gohttp.Result
|
var res *gohttp.Result
|
||||||
@ -194,13 +226,13 @@ func (c *Caller) doWithNode(manualDo bool, method, app, withNode, path string, d
|
|||||||
res = &gohttp.Result{Error: err, Response: resp}
|
res = &gohttp.Result{Error: err, Response: resp}
|
||||||
} else {
|
} else {
|
||||||
if c.Request != nil {
|
if c.Request != nil {
|
||||||
if manualDo {
|
if manual {
|
||||||
res = hc.ManualDoByRequest(c.Request, method, url, data, settedHeaders...)
|
res = hc.ManualDoByRequest(c.Request, method, url, data, settedHeaders...)
|
||||||
} else {
|
} else {
|
||||||
res = hc.DoByRequest(c.Request, method, url, data, settedHeaders...)
|
res = hc.DoByRequest(c.Request, method, url, data, settedHeaders...)
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
if manualDo {
|
if manual {
|
||||||
res = hc.ManualDo(method, url, data, settedHeaders...)
|
res = hc.ManualDo(method, url, data, settedHeaders...)
|
||||||
} else {
|
} else {
|
||||||
res = hc.Do(method, url, data, settedHeaders...)
|
res = hc.Do(method, url, data, settedHeaders...)
|
||||||
@ -209,10 +241,11 @@ func (c *Caller) doWithNode(manualDo bool, method, app, withNode, path string, d
|
|||||||
}
|
}
|
||||||
|
|
||||||
responseTime := time.Since(startTime)
|
responseTime := time.Since(startTime)
|
||||||
settedLoadBalancer.Response(&appClient, node, res.Error, res.Response, responseTime)
|
usedTimeMs := float32(responseTime.Nanoseconds()) / 1e6
|
||||||
|
c.discoverer.settedLoadBalancer.Response(&appClient, node, res.Error, res.Response, responseTime)
|
||||||
|
|
||||||
if res.Error != nil || (res.Response != nil && res.Response.StatusCode >= 502 && res.Response.StatusCode <= 504) {
|
if res.Error != nil || (res.Response != nil && res.Response.StatusCode >= 502 && res.Response.StatusCode <= 504) {
|
||||||
node.FailedTimes++
|
node.FailedTimes.Add(1)
|
||||||
errStr := ""
|
errStr := ""
|
||||||
if res.Error != nil {
|
if res.Error != nil {
|
||||||
errStr = res.Error.Error()
|
errStr = res.Error.Error()
|
||||||
@ -220,18 +253,17 @@ func (c *Caller) doWithNode(manualDo bool, method, app, withNode, path string, d
|
|||||||
errStr = res.Response.Status
|
errStr = res.Response.Status
|
||||||
}
|
}
|
||||||
|
|
||||||
c.logError(errStr, "app", app, "node", node.Addr, "path", path, "tryTimes", appClient.tryTimes)
|
c.logError(errStr, "app", app, "node", node.Addr, "path", path, "attempts", appClient.attempts)
|
||||||
|
appClient.Log(node.Addr, usedTimeMs, fmt.Errorf("%s", errStr))
|
||||||
|
|
||||||
if node.FailedTimes >= Config.CallRetryTimes {
|
if node.FailedTimes.Load() >= int32(conf.CallRetryTimes) {
|
||||||
logError("node removed due to high failures", "app", app, "node", node.Addr)
|
c.discoverer.logError("node isolated locally 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
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
|
node.FailedTimes.Store(0)
|
||||||
|
appClient.Log(node.Addr, usedTimeMs, nil)
|
||||||
if strings.ToUpper(method) == "WS" {
|
if strings.ToUpper(method) == "WS" {
|
||||||
return wsConn, node.Addr
|
return wsConn, node.Addr
|
||||||
}
|
}
|
||||||
|
|||||||
29
Config.go
29
Config.go
@ -1,14 +1,37 @@
|
|||||||
package discover
|
package discover
|
||||||
|
|
||||||
// Config 存储发现服务的全局配置
|
import (
|
||||||
var Config = struct {
|
"sync"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ConfigStruct 存储发现服务的配置
|
||||||
|
type ConfigStruct struct {
|
||||||
Registry string // 注册中心地址,如 redis://:@127.0.0.1:6379/15
|
Registry string // 注册中心地址,如 redis://:@127.0.0.1:6379/15
|
||||||
App string // 当前应用名称
|
App string // 当前应用名称
|
||||||
Weight int // 权重,默认为 100
|
Weight int // 权重,默认为 100
|
||||||
Calls map[string]string // 调用的应用列表及其配置
|
Calls map[string]string // 调用的应用列表及其配置
|
||||||
CallRetryTimes int // 调用重试次数
|
CallRetryTimes int // 调用重试次数
|
||||||
IpPrefix string // 指定使用的 IP 网段
|
IpPrefix string // 指定使用的 IP 网段
|
||||||
}{
|
}
|
||||||
|
|
||||||
|
// Config 存储发现服务的全局配置(兼容旧代码)
|
||||||
|
var Config = ConfigStruct{
|
||||||
Weight: 100,
|
Weight: 100,
|
||||||
CallRetryTimes: 10,
|
CallRetryTimes: 10,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var configLock sync.RWMutex
|
||||||
|
|
||||||
|
// SetConfig 安全地设置全局配置
|
||||||
|
func SetConfig(conf ConfigStruct) {
|
||||||
|
configLock.Lock()
|
||||||
|
defer configLock.Unlock()
|
||||||
|
Config = conf
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetConfig 安全地获取全局配置
|
||||||
|
func GetConfig() ConfigStruct {
|
||||||
|
configLock.RLock()
|
||||||
|
defer configLock.RUnlock()
|
||||||
|
return Config
|
||||||
|
}
|
||||||
|
|||||||
48
Constants.go
48
Constants.go
@ -1,35 +1,27 @@
|
|||||||
package discover
|
package discover
|
||||||
|
|
||||||
const (
|
import (
|
||||||
HeaderFromApp = "X-Discover-From-App"
|
gohttp "apigo.cc/go/http"
|
||||||
HeaderFromNode = "X-Discover-From-Node"
|
|
||||||
|
|
||||||
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 = []string{
|
const (
|
||||||
HeaderClientIp,
|
HeaderFromApp = gohttp.HeaderFromApp
|
||||||
HeaderForwardedFor,
|
HeaderFromNode = gohttp.HeaderFromNode
|
||||||
HeaderUserId,
|
|
||||||
HeaderDeviceId,
|
HeaderClientIP = gohttp.HeaderClientIP
|
||||||
HeaderClientAppName,
|
HeaderForwardedFor = gohttp.HeaderForwardedFor
|
||||||
HeaderClientAppVersion,
|
HeaderUserID = gohttp.HeaderUserID
|
||||||
HeaderSessionId,
|
HeaderDeviceID = gohttp.HeaderDeviceID
|
||||||
HeaderRequestId,
|
HeaderClientAppName = gohttp.HeaderClientAppName
|
||||||
HeaderHost,
|
HeaderClientAppVersion = gohttp.HeaderClientAppVersion
|
||||||
HeaderScheme,
|
HeaderSessionID = gohttp.HeaderSessionID
|
||||||
HeaderUserAgent,
|
HeaderRequestID = gohttp.HeaderRequestID
|
||||||
}
|
HeaderHost = gohttp.HeaderHost
|
||||||
|
HeaderScheme = gohttp.HeaderScheme
|
||||||
|
HeaderUserAgent = gohttp.HeaderUserAgent
|
||||||
|
)
|
||||||
|
|
||||||
|
var RelayHeaders = gohttp.RelayHeaders
|
||||||
|
|
||||||
const DefaultRegistry = "127.0.0.1:6379::15"
|
const DefaultRegistry = "127.0.0.1:6379::15"
|
||||||
const EnvRegistry = "DISCOVER_REGISTRY"
|
const EnvRegistry = "DISCOVER_REGISTRY"
|
||||||
|
|||||||
536
Discover.go
536
Discover.go
@ -9,36 +9,45 @@ import (
|
|||||||
"regexp"
|
"regexp"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"syscall"
|
"syscall"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"apigo.cc/go/cast"
|
"apigo.cc/go/cast"
|
||||||
"apigo.cc/go/config"
|
"apigo.cc/go/config"
|
||||||
|
gohttp "apigo.cc/go/http"
|
||||||
"apigo.cc/go/id"
|
"apigo.cc/go/id"
|
||||||
"apigo.cc/go/log"
|
"apigo.cc/go/log"
|
||||||
"apigo.cc/go/redis"
|
"apigo.cc/go/redis"
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
// Discoverer 发现服务实例
|
||||||
|
type Discoverer struct {
|
||||||
|
config ConfigStruct
|
||||||
|
configLock sync.RWMutex
|
||||||
|
|
||||||
serverRedisPool *redis.Redis
|
serverRedisPool *redis.Redis
|
||||||
clientRedisPool *redis.Redis
|
clientRedisPool *redis.Redis
|
||||||
pubsubRedisPool *redis.Redis
|
pubsubRedisPool *redis.Redis
|
||||||
isServer = false
|
isServer bool
|
||||||
isClient = false
|
isClient bool
|
||||||
daemonRunning = false
|
daemonRunning atomic.Bool
|
||||||
myAddr = ""
|
myAddr string
|
||||||
_logger = log.DefaultLogger
|
logger *log.Logger
|
||||||
_inited = false
|
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
|
||||||
|
|
||||||
daemonStopChan chan bool
|
appClientPools map[string]*gohttp.Client
|
||||||
appLock sync.RWMutex
|
appClientPoolsLock sync.RWMutex
|
||||||
_calls = map[string]*callInfoType{}
|
|
||||||
_appNodes = map[string]map[string]*NodeInfo{}
|
|
||||||
appSubscribed = map[string]bool{}
|
|
||||||
|
|
||||||
settedRoute func(*AppClient, *http.Request) = nil
|
settedRoute func(*AppClient, *http.Request)
|
||||||
settedLoadBalancer LoadBalancer = &DefaultLoadBalancer{}
|
settedLoadBalancer LoadBalancer
|
||||||
)
|
}
|
||||||
|
|
||||||
type callInfoType struct {
|
type callInfoType struct {
|
||||||
Timeout time.Duration
|
Timeout time.Duration
|
||||||
@ -47,138 +56,202 @@ type callInfoType struct {
|
|||||||
SSL bool
|
SSL bool
|
||||||
}
|
}
|
||||||
|
|
||||||
func IsServer() bool { return isServer }
|
// DefaultDiscoverer 默认的全局发现服务实例
|
||||||
func IsClient() bool { return isClient }
|
var DefaultDiscoverer = NewDiscoverer()
|
||||||
|
|
||||||
func logError(error string, extra ...any) {
|
// NewDiscoverer 创建一个新的发现服务实例
|
||||||
_logger.Error("Discover: "+error, append(extra, "app", Config.App, "addr", myAddr)...)
|
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),
|
||||||
|
appClientPools: make(map[string]*gohttp.Client),
|
||||||
|
settedLoadBalancer: &DefaultLoadBalancer{},
|
||||||
|
daemonStopSignal: make(chan struct{}),
|
||||||
|
daemonDoneSignal: make(chan struct{}),
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func logInfo(info string, extra ...any) {
|
// GetConfig 安全地获取配置
|
||||||
_logger.Info("Discover: "+info, append(extra, "app", Config.App, "addr", myAddr)...)
|
func (d *Discoverer) GetConfig() ConfigStruct {
|
||||||
|
d.configLock.RLock()
|
||||||
|
defer d.configLock.RUnlock()
|
||||||
|
return d.config
|
||||||
}
|
}
|
||||||
|
|
||||||
func SetLogger(logger *log.Logger) {
|
// SetConfig 安全地设置配置
|
||||||
_logger = logger
|
func (d *Discoverer) SetConfig(conf ConfigStruct) {
|
||||||
|
d.configLock.Lock()
|
||||||
|
defer d.configLock.Unlock()
|
||||||
|
d.config = conf
|
||||||
}
|
}
|
||||||
|
|
||||||
func Init() {
|
// IsServer 返回当前节点是否作为服务端运行
|
||||||
appLock.Lock()
|
func (d *Discoverer) IsServer() bool { return d.isServer }
|
||||||
defer appLock.Unlock()
|
|
||||||
if _inited {
|
// IsClient 返回当前节点是否作为客户端运行
|
||||||
|
func (d *Discoverer) IsClient() bool { return d.isClient }
|
||||||
|
|
||||||
|
func (d *Discoverer) logError(msg string, extra ...any) {
|
||||||
|
conf := d.GetConfig()
|
||||||
|
d.logger.Error("Discover: "+msg, append(extra, "app", conf.App, "addr", d.myAddr)...)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d *Discoverer) logInfo(msg string, extra ...any) {
|
||||||
|
conf := d.GetConfig()
|
||||||
|
d.logger.Info("Discover: "+msg, append(extra, "app", conf.App, "addr", d.myAddr)...)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
_inited = true
|
d.inited = true
|
||||||
_ = config.Load(&Config, "discover")
|
|
||||||
|
|
||||||
if Config.CallRetryTimes <= 0 {
|
conf := d.GetConfig()
|
||||||
Config.CallRetryTimes = 10
|
// 如果是默认实例,尝试加载配置
|
||||||
}
|
if d == DefaultDiscoverer {
|
||||||
if Config.Weight <= 0 {
|
_ = config.Load(&conf, "discover")
|
||||||
Config.Weight = 100
|
d.SetConfig(conf)
|
||||||
}
|
SetConfig(conf) // 保持全局 Config 变量同步
|
||||||
if Config.Registry == "" {
|
|
||||||
Config.Registry = DefaultRegistry
|
|
||||||
}
|
}
|
||||||
|
|
||||||
_logger = log.New(id.MakeID(12))
|
if conf.App == "" {
|
||||||
|
conf.App = os.Getenv("DISCOVER_APP")
|
||||||
|
}
|
||||||
|
|
||||||
|
if conf.CallRetryTimes <= 0 {
|
||||||
|
conf.CallRetryTimes = 10
|
||||||
|
}
|
||||||
|
if conf.Weight <= 0 {
|
||||||
|
conf.Weight = 100
|
||||||
|
}
|
||||||
|
if conf.Registry == "" {
|
||||||
|
conf.Registry = DefaultRegistry
|
||||||
|
}
|
||||||
|
d.SetConfig(conf)
|
||||||
|
|
||||||
|
if d.logger == log.DefaultLogger || d.logger == nil {
|
||||||
|
d.logger = log.New(id.MakeID(12))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func Start(addr string) bool {
|
// Start 启动服务发现,指定当前节点的外部访问地址
|
||||||
Init()
|
func (d *Discoverer) Start(addr string) bool {
|
||||||
myAddr = addr
|
d.Init()
|
||||||
|
d.myAddr = addr
|
||||||
|
|
||||||
isServer = Config.App != "" && Config.Weight > 0
|
conf := d.GetConfig()
|
||||||
if isServer && Config.Registry != "" {
|
d.isServer = conf.App != "" && conf.Weight > 0
|
||||||
serverRedisPool = redis.GetRedis(Config.Registry, _logger)
|
if d.isServer && conf.Registry != "" {
|
||||||
if serverRedisPool.Error != nil {
|
d.serverRedisPool = redis.GetRedis(conf.Registry, d.logger)
|
||||||
logError(serverRedisPool.Error.Error())
|
if d.serverRedisPool.Error != nil {
|
||||||
|
d.logError(d.serverRedisPool.Error.Error())
|
||||||
}
|
}
|
||||||
|
|
||||||
// 注册节点
|
// 注册节点
|
||||||
if serverRedisPool.Do("HSET", Config.App, addr, Config.Weight).Error == nil {
|
if d.serverRedisPool.Do("HSET", conf.App, addr, conf.Weight).Error == nil {
|
||||||
serverRedisPool.Do("SETEX", Config.App+"_"+addr, 10, "1")
|
d.serverRedisPool.Do("SETEX", conf.App+"_"+addr, 10, "1")
|
||||||
logInfo("registered")
|
d.logInfo("registered")
|
||||||
serverRedisPool.PUBLISH("CH_"+Config.App, fmt.Sprintf("%s %d", addr, Config.Weight))
|
d.serverRedisPool.PUBLISH("CH_"+conf.App, fmt.Sprintf("%s %d", addr, conf.Weight))
|
||||||
daemonRunning = true
|
d.daemonRunning.Store(true)
|
||||||
daemonStopChan = make(chan bool)
|
d.daemonStopSignal = make(chan struct{})
|
||||||
go daemon()
|
d.daemonDoneSignal = make(chan struct{})
|
||||||
|
go d.daemon()
|
||||||
} else {
|
} else {
|
||||||
logError("register failed")
|
d.logError("register failed")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
calls := getCalls()
|
calls := d.getCalls()
|
||||||
if len(calls) > 0 {
|
if len(calls) > 0 {
|
||||||
for app, conf := range calls {
|
for app, c := range calls {
|
||||||
addApp(app, conf, false)
|
d.addApp(app, c, false)
|
||||||
}
|
}
|
||||||
if !startSub() {
|
if !d.startSub() {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
func daemon() {
|
func (d *Discoverer) daemon() {
|
||||||
logInfo("daemon thread started")
|
d.logInfo("daemon thread started")
|
||||||
ticker := time.NewTicker(time.Second)
|
ticker := time.NewTicker(5 * time.Second)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|
||||||
for daemonRunning {
|
for d.daemonRunning.Load() {
|
||||||
<-ticker.C
|
select {
|
||||||
if !daemonRunning {
|
case <-ticker.C:
|
||||||
break
|
if !d.daemonRunning.Load() {
|
||||||
}
|
break
|
||||||
|
|
||||||
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 {
|
|
||||||
serverRedisPool.Do("SETEX", Config.App+"_"+myAddr, 10, "1")
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
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
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
logInfo("daemon thread stopped")
|
done:
|
||||||
if daemonStopChan != nil {
|
d.logInfo("daemon thread stopped")
|
||||||
daemonStopChan <- true
|
close(d.daemonDoneSignal)
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func startSub() bool {
|
func (d *Discoverer) startSub() bool {
|
||||||
if Config.Registry == "" {
|
conf := d.GetConfig()
|
||||||
|
if conf.Registry == "" {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
appLock.Lock()
|
d.appLock.Lock()
|
||||||
if clientRedisPool == nil {
|
if d.clientRedisPool == nil {
|
||||||
clientRedisPool = redis.GetRedis(Config.Registry, _logger)
|
d.clientRedisPool = redis.GetRedis(conf.Registry, d.logger)
|
||||||
}
|
}
|
||||||
|
|
||||||
if pubsubRedisPool == nil {
|
if d.pubsubRedisPool == nil {
|
||||||
pubsubRedisPool = redis.GetRedis(Config.Registry, _logger.New(id.MakeID(12)))
|
d.pubsubRedisPool = redis.GetRedis(conf.Registry, d.logger.New(id.MakeID(12)))
|
||||||
// 订阅所有已注册的应用
|
// 订阅所有已注册的应用
|
||||||
for app := range appSubscribed {
|
for app := range d.appSubscribed {
|
||||||
subscribeAppUnderLock(app)
|
d.subscribeApp(app)
|
||||||
}
|
}
|
||||||
// 必须在释放锁之前完成配置,但在释放锁之后启动,避免死锁
|
// 必须在释放锁之前完成配置,但在释放锁之后启动,避免死锁
|
||||||
appLock.Unlock()
|
d.appLock.Unlock()
|
||||||
pubsubRedisPool.Start()
|
d.pubsubRedisPool.Start()
|
||||||
appLock.Lock()
|
d.appLock.Lock()
|
||||||
}
|
}
|
||||||
|
|
||||||
isClient = true
|
d.isClient = true
|
||||||
appLock.Unlock()
|
d.appLock.Unlock()
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
func subscribeAppUnderLock(app string) {
|
func (d *Discoverer) subscribeApp(app string) {
|
||||||
pubsubRedisPool.Subscribe("CH_"+app, func() {
|
d.pubsubRedisPool.Subscribe("CH_"+app, func() {
|
||||||
fetchApp(app)
|
d.fetchApp(app)
|
||||||
}, func(data []byte) {
|
}, func(data []byte) {
|
||||||
a := strings.Split(string(data), " ")
|
a := strings.Split(string(data), " ")
|
||||||
addr := a[0]
|
addr := a[0]
|
||||||
@ -186,39 +259,66 @@ func subscribeAppUnderLock(app string) {
|
|||||||
if len(a) == 2 {
|
if len(a) == 2 {
|
||||||
weight = cast.Int(a[1])
|
weight = cast.Int(a[1])
|
||||||
}
|
}
|
||||||
logInfo("received node update", "app", app, "addr", addr, "weight", weight)
|
d.logInfo("received node update", "app", app, "addr", addr, "weight", weight)
|
||||||
pushNode(app, addr, weight)
|
d.pushNode(app, addr, weight)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
func Stop() {
|
// Stop 停止 Discover 并从注册中心注销当前节点
|
||||||
appLock.Lock()
|
func (d *Discoverer) Stop() {
|
||||||
if isClient && pubsubRedisPool != nil {
|
d.appLock.Lock()
|
||||||
pubsubRedisPool.Stop()
|
|
||||||
isClient = false
|
// 1. 提取需要的状态,提前修改标志位
|
||||||
}
|
isClient := d.isClient
|
||||||
|
pubsub := d.pubsubRedisPool
|
||||||
|
d.isClient = false
|
||||||
|
|
||||||
|
isServer := d.isServer
|
||||||
|
serverPool := d.serverRedisPool
|
||||||
|
myAddr := d.myAddr
|
||||||
|
|
||||||
if isServer {
|
if isServer {
|
||||||
daemonRunning = false
|
d.daemonRunning.Store(false)
|
||||||
if serverRedisPool != nil {
|
if d.daemonStopSignal != nil {
|
||||||
serverRedisPool.Do("HDEL", Config.App, myAddr)
|
close(d.daemonStopSignal)
|
||||||
serverRedisPool.Do("DEL", Config.App+"_"+myAddr)
|
|
||||||
serverRedisPool.PUBLISH("CH_"+Config.App, fmt.Sprintf("%s %d", myAddr, 0))
|
|
||||||
}
|
}
|
||||||
isServer = false
|
d.isServer = false
|
||||||
}
|
}
|
||||||
appLock.Unlock()
|
|
||||||
|
// 核心修复:在这里尽早释放锁!避免与 pushNode 回调发生死锁
|
||||||
|
d.appLock.Unlock()
|
||||||
|
|
||||||
|
// 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()
|
||||||
|
}
|
||||||
|
d.appClientPools = make(map[string]*gohttp.Client)
|
||||||
|
d.appClientPoolsLock.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
func Wait() {
|
// Wait 等待守护进程退出
|
||||||
if daemonStopChan != nil {
|
func (d *Discoverer) Wait() {
|
||||||
<-daemonStopChan
|
if d.daemonDoneSignal != nil {
|
||||||
daemonStopChan = nil
|
<-d.daemonDoneSignal
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func EasyStart() (string, int) {
|
// EasyStart 自动根据环境变量和本地网卡信息启动 Discover
|
||||||
Init()
|
func (d *Discoverer) EasyStart() (string, int) {
|
||||||
|
d.Init()
|
||||||
port := 0
|
port := 0
|
||||||
if listen := os.Getenv("DISCOVER_LISTEN"); listen != "" {
|
if listen := os.Getenv("DISCOVER_LISTEN"); listen != "" {
|
||||||
if _, p, err := net.SplitHostPort(listen); err == nil {
|
if _, p, err := net.SplitHostPort(listen); err == nil {
|
||||||
@ -230,13 +330,14 @@ func EasyStart() (string, int) {
|
|||||||
|
|
||||||
ln, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
|
ln, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logError("failed to listen", "err", err)
|
d.logError("failed to listen", "err", err)
|
||||||
return "", 0
|
return "", 0
|
||||||
}
|
}
|
||||||
addrInfo := ln.Addr().(*net.TCPAddr)
|
addrInfo := ln.Addr().(*net.TCPAddr)
|
||||||
_ = ln.Close()
|
_ = ln.Close()
|
||||||
port = addrInfo.Port
|
port = addrInfo.Port
|
||||||
|
|
||||||
|
conf := d.GetConfig()
|
||||||
ip := addrInfo.IP
|
ip := addrInfo.IP
|
||||||
if !ip.IsGlobalUnicast() {
|
if !ip.IsGlobalUnicast() {
|
||||||
addrs, _ := net.InterfaceAddrs()
|
addrs, _ := net.InterfaceAddrs()
|
||||||
@ -246,7 +347,7 @@ func EasyStart() (string, int) {
|
|||||||
if ip4 == nil || !ip4.IsGlobalUnicast() {
|
if ip4 == nil || !ip4.IsGlobalUnicast() {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if Config.IpPrefix != "" && strings.HasPrefix(ip4.String(), Config.IpPrefix) {
|
if conf.IpPrefix != "" && strings.HasPrefix(ip4.String(), conf.IpPrefix) {
|
||||||
ip = ip4
|
ip = ip4
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
@ -258,7 +359,7 @@ func EasyStart() (string, int) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
addr := fmt.Sprintf("%s:%d", ip.String(), port)
|
addr := fmt.Sprintf("%s:%d", ip.String(), port)
|
||||||
if !Start(addr) {
|
if !d.Start(addr) {
|
||||||
return "", 0
|
return "", 0
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -266,48 +367,64 @@ func EasyStart() (string, int) {
|
|||||||
signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
|
signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
|
||||||
go func() {
|
go func() {
|
||||||
<-sigChan
|
<-sigChan
|
||||||
Stop()
|
d.Stop()
|
||||||
}()
|
}()
|
||||||
|
|
||||||
return ip.String(), port
|
return ip.String(), port
|
||||||
}
|
}
|
||||||
|
|
||||||
func AddExternalApp(app, callConf string) bool {
|
// AddExternalApp 动态添加需要发现的外部应用
|
||||||
if addApp(app, callConf, true) {
|
func (d *Discoverer) AddExternalApp(app, callConf string) bool {
|
||||||
if !isClient {
|
if d.addApp(app, callConf, true) {
|
||||||
startSub()
|
if !d.isClient {
|
||||||
|
d.startSub()
|
||||||
} else {
|
} else {
|
||||||
appLock.Lock()
|
d.subscribeApp(app)
|
||||||
subscribeAppUnderLock(app)
|
|
||||||
appLock.Unlock()
|
|
||||||
}
|
}
|
||||||
|
d.fetchApp(app) // 同步拉取一次
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
func SetNode(app, addr string, weight int) {
|
// SetNode 手动设置某个服务的节点信息
|
||||||
pushNode(app, addr, weight)
|
func (d *Discoverer) SetNode(app, addr string, weight int) {
|
||||||
|
d.pushNode(app, addr, weight)
|
||||||
}
|
}
|
||||||
|
|
||||||
func getCallInfo(app string) *callInfoType {
|
func (d *Discoverer) getCallInfo(app string) *callInfoType {
|
||||||
appLock.RLock()
|
d.appLock.RLock()
|
||||||
defer appLock.RUnlock()
|
defer d.appLock.RUnlock()
|
||||||
return _calls[app]
|
return d.calls[app]
|
||||||
}
|
}
|
||||||
|
|
||||||
var numberMatcher = regexp.MustCompile(`^\d+(s|ms|us|µs|ns?)?$`)
|
var numberMatcher = regexp.MustCompile(`^\d+(s|ms|us|µs|ns?)?$`)
|
||||||
|
|
||||||
func addApp(app, callConf string, fetch bool) bool {
|
func (d *Discoverer) addApp(app, callConf string, fetch bool) bool {
|
||||||
appLock.Lock()
|
d.appLock.Lock()
|
||||||
if Config.Calls == nil {
|
conf := d.GetConfig()
|
||||||
Config.Calls = make(map[string]string)
|
|
||||||
|
// 1. 写时复制(Copy-on-Write):创建一个全新的 Map 避免影响读操作
|
||||||
|
newCalls := make(map[string]string)
|
||||||
|
for k, v := range conf.Calls {
|
||||||
|
newCalls[k] = v
|
||||||
}
|
}
|
||||||
if Config.Calls[app] == callConf && _appNodes[app] != nil {
|
|
||||||
appLock.Unlock()
|
if newCalls[app] == callConf && d.appNodes[app] != nil {
|
||||||
|
d.appLock.Unlock()
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
Config.Calls[app] = callConf
|
|
||||||
|
newCalls[app] = callConf
|
||||||
|
conf.Calls = newCalls // 将新的 Map 赋值给 ConfigStruct
|
||||||
|
|
||||||
|
// 2. 更新实例配置
|
||||||
|
d.SetConfig(conf)
|
||||||
|
|
||||||
|
// 3. 如果是默认的全局实例,保持包级全局配置同步
|
||||||
|
if d == DefaultDiscoverer {
|
||||||
|
SetConfig(conf)
|
||||||
|
}
|
||||||
|
|
||||||
callInfo := &callInfoType{
|
callInfo := &callInfoType{
|
||||||
Timeout: 10 * time.Second,
|
Timeout: 10 * time.Second,
|
||||||
@ -339,23 +456,23 @@ func addApp(app, callConf string, fetch bool) bool {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
_calls[app] = callInfo
|
d.calls[app] = callInfo
|
||||||
if _appNodes[app] == nil {
|
if d.appNodes[app] == nil {
|
||||||
_appNodes[app] = make(map[string]*NodeInfo)
|
d.appNodes[app] = make(map[string]*NodeInfo)
|
||||||
}
|
}
|
||||||
appSubscribed[app] = true
|
d.appSubscribed[app] = true
|
||||||
appLock.Unlock()
|
d.appLock.Unlock()
|
||||||
|
|
||||||
if fetch {
|
if fetch && d.isClient {
|
||||||
fetchApp(app)
|
d.fetchApp(app)
|
||||||
}
|
}
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
func fetchApp(app string) {
|
func (d *Discoverer) fetchApp(app string) {
|
||||||
appLock.RLock()
|
d.appLock.RLock()
|
||||||
pool := clientRedisPool
|
pool := d.clientRedisPool
|
||||||
appLock.RUnlock()
|
d.appLock.RUnlock()
|
||||||
if pool == nil {
|
if pool == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@ -370,80 +487,131 @@ func fetchApp(app string) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
currentNodes := getAppNodes(app)
|
currentNodes := d.getAppNodes(app)
|
||||||
if currentNodes != nil {
|
if currentNodes != nil {
|
||||||
for addr := range currentNodes {
|
for addr := range currentNodes {
|
||||||
if _, ok := results[addr]; !ok {
|
if _, ok := results[addr]; !ok {
|
||||||
pushNode(app, addr, 0)
|
d.pushNode(app, addr, 0)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
for addr, res := range results {
|
for addr, res := range results {
|
||||||
pushNode(app, addr, res.Int())
|
d.pushNode(app, addr, res.Int())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func getAppNodes(app string) map[string]*NodeInfo {
|
func (d *Discoverer) getAppNodes(app string) map[string]*NodeInfo {
|
||||||
appLock.RLock()
|
d.appLock.RLock()
|
||||||
defer appLock.RUnlock()
|
defer d.appLock.RUnlock()
|
||||||
if _appNodes[app] == nil {
|
if d.appNodes[app] == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
nodes := make(map[string]*NodeInfo)
|
nodes := make(map[string]*NodeInfo)
|
||||||
for k, v := range _appNodes[app] {
|
for k, v := range d.appNodes[app] {
|
||||||
nodes[k] = v
|
nodes[k] = v
|
||||||
}
|
}
|
||||||
return nodes
|
return nodes
|
||||||
}
|
}
|
||||||
|
|
||||||
func getCalls() map[string]string {
|
func (d *Discoverer) getCalls() map[string]string {
|
||||||
appLock.RLock()
|
conf := d.GetConfig()
|
||||||
defer appLock.RUnlock()
|
|
||||||
calls := make(map[string]string)
|
calls := make(map[string]string)
|
||||||
for k, v := range Config.Calls {
|
for k, v := range conf.Calls {
|
||||||
calls[k] = v
|
calls[k] = v
|
||||||
}
|
}
|
||||||
return calls
|
return calls
|
||||||
}
|
}
|
||||||
|
|
||||||
func GetAppNodes(app string) map[string]*NodeInfo {
|
// GetAppNodes 获取某个应用的所有节点列表
|
||||||
return getAppNodes(app)
|
func (d *Discoverer) GetAppNodes(app string) map[string]*NodeInfo {
|
||||||
|
return d.getAppNodes(app)
|
||||||
}
|
}
|
||||||
|
|
||||||
func pushNode(app, addr string, weight int) {
|
func (d *Discoverer) pushNode(app, addr string, weight int) {
|
||||||
appLock.Lock()
|
d.appLock.Lock()
|
||||||
defer appLock.Unlock()
|
defer d.appLock.Unlock()
|
||||||
|
|
||||||
if weight <= 0 {
|
if weight <= 0 {
|
||||||
if _appNodes[app] != nil {
|
if d.appNodes[app] != nil {
|
||||||
delete(_appNodes[app], addr)
|
delete(d.appNodes[app], addr)
|
||||||
}
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if _appNodes[app] == nil {
|
if d.appNodes[app] == nil {
|
||||||
_appNodes[app] = make(map[string]*NodeInfo)
|
d.appNodes[app] = make(map[string]*NodeInfo)
|
||||||
}
|
}
|
||||||
|
|
||||||
if node, ok := _appNodes[app][addr]; ok {
|
if node, ok := d.appNodes[app][addr]; ok {
|
||||||
if node.Weight != weight {
|
if node.Weight != weight {
|
||||||
node.UsedTimes = uint64(float64(node.UsedTimes) / float64(node.Weight) * float64(weight))
|
used := node.UsedTimes.Load()
|
||||||
|
node.UsedTimes.Store(uint64(float64(used) / float64(node.Weight) * float64(weight)))
|
||||||
node.Weight = weight
|
node.Weight = weight
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
var avgUsed uint64 = 0
|
var avgUsed uint64 = 0
|
||||||
if len(_appNodes[app]) > 0 {
|
if len(d.appNodes[app]) > 0 {
|
||||||
var totalScore float64
|
var totalScore float64
|
||||||
for _, n := range _appNodes[app] {
|
for _, n := range d.appNodes[app] {
|
||||||
totalScore += float64(n.UsedTimes) / float64(n.Weight)
|
totalScore += float64(n.UsedTimes.Load()) / float64(n.Weight)
|
||||||
}
|
}
|
||||||
avgUsed = uint64(totalScore / float64(len(_appNodes[app])) * float64(weight))
|
avgUsed = uint64(totalScore / float64(len(d.appNodes[app])) * float64(weight))
|
||||||
}
|
}
|
||||||
_appNodes[app][addr] = &NodeInfo{
|
node := &NodeInfo{
|
||||||
Addr: addr,
|
Addr: addr,
|
||||||
Weight: weight,
|
Weight: weight,
|
||||||
UsedTimes: avgUsed,
|
|
||||||
}
|
}
|
||||||
|
node.UsedTimes.Store(avgUsed)
|
||||||
|
d.appNodes[app][addr] = node
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 以下是包级别 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)
|
||||||
|
}
|
||||||
|
|||||||
@ -15,11 +15,12 @@ import (
|
|||||||
|
|
||||||
func TestDiscover(t *testing.T) {
|
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 {
|
if err != nil {
|
||||||
t.Skip("failed to listen on :18001, skipping test")
|
t.Skip("failed to listen, skipping test")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
addr := l.Addr().String()
|
||||||
mux := http.NewServeMux()
|
mux := http.NewServeMux()
|
||||||
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
|
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
|
||||||
_, _ = w.Write([]byte("OK"))
|
_, _ = w.Write([]byte("OK"))
|
||||||
@ -46,11 +47,13 @@ func TestDiscover(t *testing.T) {
|
|||||||
defer server.Close()
|
defer server.Close()
|
||||||
|
|
||||||
// 配置 Discover
|
// 配置 Discover
|
||||||
discover.Config.App = "test-app"
|
conf := discover.DefaultDiscoverer.GetConfig()
|
||||||
discover.Config.Registry = "redis://127.0.0.1:6379/15"
|
conf.App = "test-app"
|
||||||
|
conf.Registry = "redis://127.0.0.1:6379/15"
|
||||||
|
discover.DefaultDiscoverer.SetConfig(conf)
|
||||||
|
|
||||||
// 启动 Discover
|
// 启动 Discover
|
||||||
if !discover.Start("127.0.0.1:18001") {
|
if !discover.Start(addr) {
|
||||||
t.Skip("failed to start discover (check redis), skipping test")
|
t.Skip("failed to start discover (check redis), skipping test")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@ -138,3 +141,24 @@ func TestEasyStart(t *testing.T) {
|
|||||||
fmt.Printf("EasyStart: %s:%d\n", ip, port)
|
fmt.Printf("EasyStart: %s:%d\n", ip, port)
|
||||||
discover.Stop()
|
discover.Stop()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func BenchmarkDiscover(b *testing.B) {
|
||||||
|
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++ {
|
||||||
|
// 模拟 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")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@ -7,7 +7,12 @@ import (
|
|||||||
|
|
||||||
// SetLoadBalancer 设置全局负载均衡策略
|
// SetLoadBalancer 设置全局负载均衡策略
|
||||||
func SetLoadBalancer(lb LoadBalancer) {
|
func SetLoadBalancer(lb LoadBalancer) {
|
||||||
settedLoadBalancer = lb
|
DefaultDiscoverer.SetLoadBalancer(lb)
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetLoadBalancer 设置负载均衡策略
|
||||||
|
func (d *Discoverer) SetLoadBalancer(lb LoadBalancer) {
|
||||||
|
d.settedLoadBalancer = lb
|
||||||
}
|
}
|
||||||
|
|
||||||
// LoadBalancer 负载均衡接口
|
// LoadBalancer 负载均衡接口
|
||||||
@ -22,20 +27,17 @@ type LoadBalancer interface {
|
|||||||
// DefaultLoadBalancer 默认负载均衡器(简单权重轮询/得分最小者优先)
|
// DefaultLoadBalancer 默认负载均衡器(简单权重轮询/得分最小者优先)
|
||||||
type DefaultLoadBalancer struct{}
|
type DefaultLoadBalancer struct{}
|
||||||
|
|
||||||
|
// Response 在默认负载均衡器中不再执行写操作,减少锁竞争
|
||||||
func (lb *DefaultLoadBalancer) Response(appClient *AppClient, node *NodeInfo, err error, response *http.Response, responseTime time.Duration) {
|
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 {
|
func (lb *DefaultLoadBalancer) Next(appClient *AppClient, nodes []*NodeInfo, request *http.Request) *NodeInfo {
|
||||||
var minScore float64 = -1
|
var minScore float64 = -1
|
||||||
var minNode *NodeInfo
|
var minNode *NodeInfo
|
||||||
for _, node := range nodes {
|
for _, node := range nodes {
|
||||||
scoreValue, ok := node.Data.Load("score")
|
// 动态计算得分,避免使用 sync.Map 存储,减少内存分配和锁竞争
|
||||||
if !ok {
|
score := float64(node.UsedTimes.Load()) / float64(node.Weight)
|
||||||
scoreValue = float64(node.UsedTimes) / float64(node.Weight)
|
|
||||||
node.Data.Store("score", scoreValue)
|
|
||||||
}
|
|
||||||
score := scoreValue.(float64)
|
|
||||||
if minNode == nil || score < minScore {
|
if minNode == nil || score < minScore {
|
||||||
minScore = score
|
minScore = score
|
||||||
minNode = node
|
minNode = node
|
||||||
|
|||||||
55
Log.go
Normal file
55
Log.go
Normal file
@ -0,0 +1,55 @@
|
|||||||
|
package discover
|
||||||
|
|
||||||
|
import (
|
||||||
|
"apigo.cc/go/log"
|
||||||
|
)
|
||||||
|
|
||||||
|
const LogTypeDiscover = "discover"
|
||||||
|
|
||||||
|
type DiscoverLog struct {
|
||||||
|
log.BaseLog
|
||||||
|
App string `log:"pos:6,color:cyan"`
|
||||||
|
Method string `log:"pos:7,color:magenta"`
|
||||||
|
Path string `log:"pos:8,color:blue"`
|
||||||
|
Node string `log:"pos:9,color:yellow"`
|
||||||
|
Attempts int `log:"pos:10"`
|
||||||
|
UsedTime float32 `log:"pos:11,format:%.2fms"`
|
||||||
|
Error string `log:"pos:12,color:red"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func (l *DiscoverLog) Reset() {
|
||||||
|
l.BaseLog.Reset()
|
||||||
|
l.App = ""
|
||||||
|
l.Method = ""
|
||||||
|
l.Path = ""
|
||||||
|
l.Node = ""
|
||||||
|
l.Attempts = 0
|
||||||
|
l.UsedTime = 0
|
||||||
|
l.Error = ""
|
||||||
|
}
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
log.RegisterType(LogTypeDiscover, DiscoverLog{})
|
||||||
|
}
|
||||||
|
|
||||||
|
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)
|
||||||
|
}
|
||||||
88
MultiInstance_test.go
Normal file
88
MultiInstance_test.go
Normal file
@ -0,0 +1,88 @@
|
|||||||
|
package discover_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"net"
|
||||||
|
"net/http"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"apigo.cc/go/discover"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestMultipleDiscoverer(t *testing.T) {
|
||||||
|
// 启动两个模拟服务
|
||||||
|
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, 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}
|
||||||
|
go func() { _ = server2.Serve(l2) }()
|
||||||
|
defer server2.Close()
|
||||||
|
|
||||||
|
registry := "redis://127.0.0.1:6379/15"
|
||||||
|
|
||||||
|
// 实例 1
|
||||||
|
d1 := discover.NewDiscoverer()
|
||||||
|
c1conf := d1.GetConfig()
|
||||||
|
c1conf.App = "app1"
|
||||||
|
c1conf.Registry = registry + "?id=1"
|
||||||
|
d1.SetConfig(c1conf)
|
||||||
|
if !d1.Start(addr1) {
|
||||||
|
t.Skip("redis not available")
|
||||||
|
}
|
||||||
|
defer d1.Stop()
|
||||||
|
|
||||||
|
// 实例 2
|
||||||
|
d2 := discover.NewDiscoverer()
|
||||||
|
c2conf := d2.GetConfig()
|
||||||
|
c2conf.App = "app2"
|
||||||
|
c2conf.Registry = registry + "?id=2"
|
||||||
|
d2.SetConfig(c2conf)
|
||||||
|
if !d2.Start(addr2) {
|
||||||
|
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,只要正确配置
|
||||||
|
d1.AddExternalApp("app2", "1")
|
||||||
|
time.Sleep(200 * time.Millisecond) // 等待同步
|
||||||
|
res3 := c1.Get("app2", "/")
|
||||||
|
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")
|
||||||
|
}
|
||||||
11
NodeInfo.go
11
NodeInfo.go
@ -2,13 +2,14 @@ package discover
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
)
|
)
|
||||||
|
|
||||||
// NodeInfo 存储服务节点信息
|
// NodeInfo 存储服务节点信息
|
||||||
type NodeInfo struct {
|
type NodeInfo struct {
|
||||||
Addr string // 节点地址
|
Addr string // 节点地址
|
||||||
Weight int // 节点权重
|
Weight int // 节点权重
|
||||||
UsedTimes uint64 // 已使用次数
|
UsedTimes atomic.Uint64 // 已使用次数
|
||||||
FailedTimes int // 失败次数
|
FailedTimes atomic.Int32 // 失败次数
|
||||||
Data sync.Map // 运行时自定义数据
|
Data sync.Map // 运行时自定义数据
|
||||||
}
|
}
|
||||||
|
|||||||
@ -28,6 +28,8 @@ discover:
|
|||||||
|
|
||||||
### 服务调用 (Caller)
|
### 服务调用 (Caller)
|
||||||
- `NewCaller(request *http.Request, logger *log.Logger) *Caller`: 创建调用器。传入原始请求可自动透传 Header。
|
- `NewCaller(request *http.Request, logger *log.Logger) *Caller`: 创建调用器。传入原始请求可自动透传 Header。
|
||||||
|
- `Call[T](method, app, path, data, headers...) (T, error)`: **[推荐]** 泛型快捷调用,自动解析 JSON 结果。
|
||||||
|
- `CallT[T](caller, ...) (T, error)`: 针对指定调用器的泛型调用。
|
||||||
- `Caller.Get / Post / Put / Delete / Head`: 发起同步请求。
|
- `Caller.Get / Post / Put / Delete / Head`: 发起同步请求。
|
||||||
- `Caller.Do(method, app, path, data, headers...)`: 发起通用请求,返回 `http.Result`。
|
- `Caller.Do(method, app, path, data, headers...)`: 发起通用请求,返回 `http.Result`。
|
||||||
- `Caller.Open(app, path, headers...)`: 发起 WebSocket 连接。
|
- `Caller.Open(app, path, headers...)`: 发起 WebSocket 连接。
|
||||||
|
|||||||
9
Route.go
9
Route.go
@ -2,7 +2,12 @@ package discover
|
|||||||
|
|
||||||
import "net/http"
|
import "net/http"
|
||||||
|
|
||||||
// SetRoute 设置全局路由规则,可以在请求前修改 App、Method、Path 等信息
|
// SetRoute 设置全局路由规则
|
||||||
func SetRoute(route func(appClient *AppClient, request *http.Request)) {
|
func SetRoute(route func(appClient *AppClient, request *http.Request)) {
|
||||||
settedRoute = route
|
DefaultDiscoverer.SetRoute(route)
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetRoute 设置路由规则
|
||||||
|
func (d *Discoverer) SetRoute(route func(appClient *AppClient, request *http.Request)) {
|
||||||
|
d.settedRoute = route
|
||||||
}
|
}
|
||||||
|
|||||||
28
TEST.md
28
TEST.md
@ -1,15 +1,19 @@
|
|||||||
# Test Report
|
# Discover Module Test Report
|
||||||
|
|
||||||
## 测试场景
|
## Test Coverage
|
||||||
1. **基础发现与调用**: 验证服务启动后能自动注册到 Redis,并能通过 Caller 正确发起请求。
|
- **Standard Discovery**: Basic registration and discovery via Redis. (Verified in `TestDiscover`)
|
||||||
2. **实时同步**: 验证通过 Redis PUBLISH 更新节点信息后,客户端能实时感知并更新本地节点列表。
|
- **WebSocket Support**: Caller supports transparent WebSocket proxying. (Verified in `TestDiscover`)
|
||||||
3. **故障剔除**: 验证当节点调用持续失败时,能自动从本地列表中剔除。
|
- **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`)
|
||||||
4. **环境变量配置**: 验证 `EasyStart` 结合环境变量的启动流程。
|
- **EasyStart**: Automatic IP and port detection for zero-config startup. (Verified in `TestEasyStart`)
|
||||||
|
|
||||||
## 测试结果
|
## Deadlock & Stability Fixes
|
||||||
- **Unit Tests**: `go test -v ./...`
|
- **Stop() Deadlock**: Fixed by releasing `appLock` before calling blocking Redis/PubSub operations.
|
||||||
- `TestDiscover`: PASS
|
- **AddExternalApp Deadlock**: Fixed by removing unnecessary lock acquisition during subscription, avoiding non-reentrant lock traps.
|
||||||
- `TestEasyStart`: PASS
|
- **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
|
## Benchmarks
|
||||||
- 待补充(Discover 主要性能开销在负载均衡算法选择,单次选择耗时极低)。
|
- **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
|
||||||
|
|||||||
33
go.mod
33
go.mod
@ -3,27 +3,26 @@ module apigo.cc/go/discover
|
|||||||
go 1.25.0
|
go 1.25.0
|
||||||
|
|
||||||
require (
|
require (
|
||||||
apigo.cc/go/cast v1.2.6
|
apigo.cc/go/cast v1.2.8
|
||||||
apigo.cc/go/config v1.0.4
|
apigo.cc/go/config v1.0.7
|
||||||
apigo.cc/go/http v1.0.3
|
apigo.cc/go/http v1.0.9
|
||||||
apigo.cc/go/id v1.0.4
|
apigo.cc/go/id v1.0.5
|
||||||
apigo.cc/go/log v1.0.2
|
apigo.cc/go/log v1.1.13
|
||||||
apigo.cc/go/redis v1.0.2
|
apigo.cc/go/redis v1.0.7
|
||||||
github.com/gorilla/websocket v1.5.3
|
github.com/gorilla/websocket v1.5.3
|
||||||
)
|
)
|
||||||
|
|
||||||
require (
|
require (
|
||||||
apigo.cc/go/convert v1.0.4 // indirect
|
apigo.cc/go/crypto v1.1.0 // indirect
|
||||||
apigo.cc/go/crypto v1.0.4 // indirect
|
apigo.cc/go/encoding v1.1.1 // indirect
|
||||||
apigo.cc/go/encoding v1.0.4 // indirect
|
apigo.cc/go/file v1.0.7 // indirect
|
||||||
apigo.cc/go/file v1.0.4 // indirect
|
apigo.cc/go/rand v1.0.5 // indirect
|
||||||
apigo.cc/go/rand v1.0.4 // indirect
|
apigo.cc/go/safe v1.0.5 // indirect
|
||||||
apigo.cc/go/safe v1.0.4 // indirect
|
apigo.cc/go/shell v1.0.5 // indirect
|
||||||
apigo.cc/go/shell v1.0.4 // indirect
|
|
||||||
github.com/gomodule/redigo v1.9.3 // indirect
|
github.com/gomodule/redigo v1.9.3 // indirect
|
||||||
golang.org/x/crypto v0.50.0 // indirect
|
golang.org/x/crypto v0.51.0 // indirect
|
||||||
golang.org/x/net v0.53.0 // indirect
|
golang.org/x/net v0.54.0 // indirect
|
||||||
golang.org/x/sys v0.43.0 // indirect
|
golang.org/x/sys v0.44.0 // indirect
|
||||||
golang.org/x/text v0.36.0 // indirect
|
golang.org/x/text v0.37.0 // indirect
|
||||||
gopkg.in/yaml.v3 v3.0.1 // indirect
|
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||||
)
|
)
|
||||||
|
|||||||
71
go.sum
71
go.sum
@ -1,48 +1,61 @@
|
|||||||
apigo.cc/go/cast v1.2.6 h1:xnWiaQAGsRCrnu1p8fIFQfg5HFSc7CxR+3ItiDIDMaY=
|
apigo.cc/go/cast v1.2.8 h1:plb676DH2TjYljzf8OEMGT6lIhmZ/xaxEFfs0kDOiSI=
|
||||||
apigo.cc/go/cast v1.2.6/go.mod h1:lGlwImiOvHxG7buyMWhFzcdvQzmSaoKbmr7bcDfUpHk=
|
apigo.cc/go/cast v1.2.8/go.mod h1:lGlwImiOvHxG7buyMWhFzcdvQzmSaoKbmr7bcDfUpHk=
|
||||||
apigo.cc/go/config v1.0.4 h1:WG9zrQkqfFPkrKIL7RNvvAbbkuUBt1Av11ZP/aIfldM=
|
apigo.cc/go/config v1.0.7 h1:lldkjsuUrWVHz/0y08/pF1vbsTvZC3TNQ2tFQ38jRNw=
|
||||||
apigo.cc/go/config v1.0.4/go.mod h1:obryzJiK6j7lQex/58d5eWYOGx5O5IABguqNWxyyXJo=
|
apigo.cc/go/config v1.0.7/go.mod h1:9oogTK83NvNmvAEIe/zuK2EKOnDtNz3bZoVqvyhFMW8=
|
||||||
apigo.cc/go/convert v1.0.4 h1:5+qPjC3dlPB59GnWZRlmthxcaXQtKvN+iOuiLdJ1GvQ=
|
apigo.cc/go/crypto v1.1.0 h1:dv9ZRbtJHnnLbDHUfjP//GHLniu0/5ja0w5QE5hwwOU=
|
||||||
apigo.cc/go/convert v1.0.4/go.mod h1:Hp+geeSyhqg/zwIKPOrDoceIREzcwM14t1I5q/dtbfU=
|
apigo.cc/go/crypto v1.1.0/go.mod h1:0NUsQMGiP95TWHJexb3F1MxNdW+LR8TD1VqwHPN8PR8=
|
||||||
apigo.cc/go/crypto v1.0.4 h1:VPUyHCH2N3LLEgdpwUc+DQssNHzLlxVzLNRa0Jm6O4o=
|
apigo.cc/go/encoding v1.1.1 h1:p/57IfKIeB+b3rTfZSN3KegZigfPOEEfuuuOmZuc+sM=
|
||||||
apigo.cc/go/crypto v1.0.4/go.mod h1:5sI8BLw6YHZfDReYwCO3TFD2LKm36HMdLg1S5oPv/QU=
|
apigo.cc/go/encoding v1.1.1/go.mod h1:GeAz5OnCkFybTR1+GWFqdMgfq5v6r4MsjWVPOk/mpf4=
|
||||||
apigo.cc/go/encoding v1.0.4 h1:aezB0J/qFuHs6iXkbtuJP5JIHUtmjsr5SFb0NNvbObY=
|
apigo.cc/go/file v1.0.7 h1:j1VBtmMZqNGnH++DYjHecX1XAKTlKAuqUiUW1HafRas=
|
||||||
apigo.cc/go/encoding v1.0.4/go.mod h1:V5CgT7rBbCxy+uCU20q0ptcNNRSgMtpA8cNOs6r8IeI=
|
apigo.cc/go/file v1.0.7/go.mod h1:2qC+p8p7iHx0DHAPubHXkLrEuLGO9WXTtdwyFjrSc1I=
|
||||||
apigo.cc/go/file v1.0.4 h1:qCKegV7OYh7r0qc3jZjGA/aKh0vIHgmr1OEbhfEmGX8=
|
apigo.cc/go/http v1.0.9 h1:fDJ11Rj/GhigSR0ROSHm0oWJidBCZ56+B+ntoEtw6+g=
|
||||||
apigo.cc/go/file v1.0.4/go.mod h1:C9gNo7386iA21OiBmuWh6CznKWlVBDFkhE4f0H0Susg=
|
apigo.cc/go/http v1.0.9/go.mod h1:ca9LYy3TURMUKKB5654ofpRgNqLaq5ibxOrWb5PgOiU=
|
||||||
apigo.cc/go/http v1.0.3 h1:c19ppdb7gR9aIPeY3qOjOj4X3+jZLXln76jTTj7i4vM=
|
apigo.cc/go/id v1.0.5 h1:23YkR7oklSA69gthYlu8zl/kpIkeIoEYxi1f1Sz5l3A=
|
||||||
apigo.cc/go/http v1.0.3/go.mod h1:oHQYlBLN6u53C2t1BihxT7cnUQd+zLTAYr3ALjWUkpg=
|
apigo.cc/go/id v1.0.5/go.mod h1:ZaYLIyrJvkf3j7J8a0lnKywSAHljaczWxU0x2HmQDzg=
|
||||||
apigo.cc/go/id v1.0.4 h1:w+JSdeVit52iefIUolrh1qLEZS9XqHNKr1UygFcgv+s=
|
apigo.cc/go/log v1.1.13 h1:ZABeVA9DxhdneLqHrYEc+6YijgoygG8eEsgDxYDzpDc=
|
||||||
apigo.cc/go/id v1.0.4/go.mod h1:kg7QuceAKtGNzGWt0+pIIh8Qom1eMSWGb8+0Yhi/QVY=
|
apigo.cc/go/log v1.1.13/go.mod h1:eabuI2SynGNgo5FXPbGgQtyxjp94wT643XzjYhEIP3A=
|
||||||
apigo.cc/go/log v1.0.2 h1:OY6T3SC28blDNkMpdRvDK2N4sGdriAB9DBItGl/qOos=
|
apigo.cc/go/rand v1.0.5 h1:AkUoWr0SELgeDmRjLEDjOIp29nXdzqQQvmGRIHpTN7U=
|
||||||
apigo.cc/go/log v1.0.2/go.mod h1:tvPgFpebY9Wf/DlqMHZ0ZjxDp9AaQTywOQKvtBaNqNo=
|
apigo.cc/go/rand v1.0.5/go.mod h1:mZ/4Soa3bk+XvDaqPWJuUe1bfEi4eThBj1XmEAuYxsk=
|
||||||
apigo.cc/go/rand v1.0.4 h1:we070eWSL0dB8NEMaWjXj43+EekXQTm/h0kKpZ/frqw=
|
apigo.cc/go/redis v1.0.7 h1:WFhbsjwIdUsmBcpMO45QTHlKu/nETpcRewLWqF5TnpQ=
|
||||||
apigo.cc/go/rand v1.0.4/go.mod h1:mZ/4Soa3bk+XvDaqPWJuUe1bfEi4eThBj1XmEAuYxsk=
|
apigo.cc/go/redis v1.0.7/go.mod h1:gEgnzhrrlZHL6XzsKEG+zR2y6l/eWIbwdT1dbhbG/7g=
|
||||||
apigo.cc/go/redis v1.0.2 h1:gWBrL/6eDxtouTFSZrPKQNdEg1AZr2aKTpCOhwim3dI=
|
apigo.cc/go/safe v1.0.5 h1:yZJLhpMntJrtqU/ev0UlyOoHu/cLrnnGUO4aHyIZcwE=
|
||||||
apigo.cc/go/redis v1.0.2/go.mod h1:auQ3cyORgD67HF5dNvZ1lA8bqMH1xIbnuKBuZWclNy4=
|
apigo.cc/go/safe v1.0.5/go.mod h1:i9xnh7reJIFPauLnlzuIDgvrQvhjxpFlpVh3O6ulWd0=
|
||||||
apigo.cc/go/safe v1.0.4 h1:07pRSdEHprF/2v6SsqAjICYFoeLcqjjvHGEdh6Dzrzg=
|
apigo.cc/go/shell v1.0.5 h1:bmvUTJGe1GwsHAy42v3iaoK40PoBC7Xq1aMCYxUZmtg=
|
||||||
apigo.cc/go/safe v1.0.4/go.mod h1:o568sHS5rTRSVPmhxWod0tGdc+8l1KjidsNY1/OVZr0=
|
apigo.cc/go/shell v1.0.5/go.mod h1:sx/nYw5CihHWmo5JHkaZUbmMYXNHx8swzArbQCUGHjc=
|
||||||
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/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
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/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 h1:dNPSXeXv6HCq2jdyWfjgmhBdqnR6PRO3m/G05nvpPC8=
|
||||||
github.com/gomodule/redigo v1.9.3/go.mod h1:KsU3hiK/Ay8U42qpaJk+kuNa3C+spxapWpM+ywhcgtw=
|
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 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
|
||||||
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
|
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 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||||
github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
|
github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
|
||||||
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
|
github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
|
||||||
|
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
|
||||||
|
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
|
||||||
golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI=
|
golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI=
|
||||||
golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q=
|
golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q=
|
||||||
|
golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI=
|
||||||
|
golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8=
|
||||||
golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
|
golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
|
||||||
golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
|
golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
|
||||||
|
golang.org/x/net v0.54.0 h1:2zJIZAxAHV/OHCDTCOHAYehQzLfSXuf/5SoL/Dv6w/w=
|
||||||
|
golang.org/x/net v0.54.0/go.mod h1:Sj4oj8jK6XmHpBZU/zWHw3BV3abl4Kvi+Ut7cQcY+cQ=
|
||||||
golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
|
golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
|
||||||
golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||||
|
golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ=
|
||||||
|
golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||||
golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg=
|
golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg=
|
||||||
golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164=
|
golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164=
|
||||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
|
golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
|
||||||
|
golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
|
||||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
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 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user