package consul import ( "context" "errors" "fmt" "net" "net/http" "sync" "time" "github.com/gogf/gf/contrib/registry/consul/v2" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/net/gsel" "github.com/gogf/gf/v2/net/gsvc" "github.com/gogf/gf/v2/os/gcfg" "github.com/gogf/gf/v2/os/genv" "github.com/gogf/gf/v2/util/grand" "github.com/hashicorp/consul/api" "github.com/r3labs/diff/v2" ) var ( registry gsvc.Registry consulAddr string reconnectMutex sync.RWMutex reconnectDone chan struct{} connected bool httpClient *http.Client ) // connectConsul 连接 Consul func connectConsul(ctx context.Context) error { reconnectMutex.Lock() defer reconnectMutex.Unlock() // 如果已经连接,不再重复连接 if connected && registry != nil { return nil } var err error registry, err = consul.New(consul.WithAddress(consulAddr)) if err != nil { g.Log().Errorf(ctx, "❌ Consul 连接失败: %v", err) return err } gsvc.SetRegistry(registry) gsel.SetBuilder(gsel.NewBuilderRoundRobin()) connected = true g.Log().Infof(ctx, "✅ Consul 初始化成功: %s", consulAddr) return nil } // startHealthCheckAndReconnect 启动健康检查和自动重连 func startHealthCheckAndReconnect() { if reconnectDone != nil { close(reconnectDone) } reconnectDone = make(chan struct{}) ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() ctx := context.Background() // 初始化HTTP客户端用于健康检查 httpClient = &http.Client{ Timeout: 5 * time.Second, } for { select { case <-ticker.C: // 检查服务发现是否正常工作 if checkConsulHealth(ctx) { continue } g.Log().Warning(ctx, "⚠️ Consul 连接异常,尝试重新连接...") // 重置连接状态并重连 reconnectMutex.Lock() connected = false registry = nil reconnectMutex.Unlock() if err := connectConsul(ctx); err != nil { g.Log().Errorf(ctx, "❌ Consul 重连失败: %v,30秒后重试...", err) } case <-reconnectDone: g.Log().Info(ctx, "🛑 Consul 健康检查已停止") return } } } // checkConsulHealth 检查 Consul 健康状态 func checkConsulHealth(ctx context.Context) bool { reconnectMutex.RLock() defer reconnectMutex.RUnlock() if registry == nil || !connected { return false } // 使用consul原生API进行健康检查 // 调用 /v1/agent/self 接口检测连接状态 url := fmt.Sprintf("http://%s/v1/agent/self", consulAddr) resp, err := httpClient.Get(url) if err != nil { g.Log().Debugf(ctx, "Consul 健康检查失败: %v", err) return false } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { g.Log().Debugf(ctx, "Consul 健康检查失败,状态码: %d", resp.StatusCode) return false } //g.Log().Debugf(ctx, "✅ Consul 健康检查通过") return true } func init() { consulAddr = g.Cfg().MustGet(context.Background(), "consul.address").String() if consulAddr == "" { g.Log().Warning(context.Background(), "⚠️ Consul 配置未找到,跳过初始化") return } if err := connectConsul(context.Background()); err != nil { g.Log().Errorf(context.Background(), "❌ Consul 初始化失败: %v", err) } else { // 连接成功后启动健康检查和自动重连 go startHealthCheckAndReconnect() } loadConfigFromConsul() } func loadConfigFromConsul() { ctx := context.Background() serviceName := g.Cfg().MustGet(ctx, "server.name", "admin-go").String() env := genv.Get("APP_ENV", "").String() fmt.Printf("服务名称: %s, 环境变量 APP_ENV: %s\n", serviceName, env) if env == "" { panic("未设置 APP_ENV 环境变量,无法从 Consul 加载配置") } consulData, lastIndex, err := loadEnvFromConsul(env, serviceName) if err == nil && len(consulData) > 0 { adapter, err := gcfg.NewAdapterContent() if err != nil { fmt.Printf("创建配置适配器失败: %v\n", err) return } adapter.SetContent(string(consulData)) g.Cfg().SetAdapter(adapter) fmt.Printf("已从 Consul 成功加载初始配置\n") go watchConsulConfig(env, serviceName, lastIndex) } else { fmt.Printf("从 Consul 获取配置失败,使用本地配置文件\n") } } func loadEnvFromConsul(env string, serviceName string) ([]byte, uint64, error) { ctx := context.Background() consulAddress := g.Cfg().MustGet(ctx, "consul.address", "127.0.0.1:8500").String() fmt.Printf("Consul 地址: %s\n", consulAddress) config := api.DefaultConfig() config.Address = consulAddress client, err := api.NewClient(config) if err != nil { fmt.Printf("创建 Consul 客户端失败: %v\n", err) return nil, 0, err } consulKey := fmt.Sprintf("config/%s/%s-%s", serviceName, serviceName, env) fmt.Printf("从 Consul 读取配置键: %s\n", consulKey) kv := client.KV() opts := &api.QueryOptions{ WaitIndex: 0, WaitTime: 60 * time.Second, } pair, meta, err := kv.Get(consulKey, opts) if err != nil { fmt.Printf("从 Consul 读取配置失败: %v\n", err) return nil, 0, err } if pair == nil { fmt.Printf("Consul 中未找到配置键: %s\n", consulKey) return nil, 0, nil } fmt.Printf("已从 Consul 加载配置, Index: %d\n", meta.LastIndex) return pair.Value, meta.LastIndex, nil } func watchConsulConfig(env string, serviceName string, lastIndex uint64) { ctx := context.Background() consulAddress := g.Cfg().MustGet(ctx, "consul.address", "127.0.0.1:8500").String() config := api.DefaultConfig() config.Address = consulAddress client, err := api.NewClient(config) if err != nil { fmt.Printf("创建 Consul 监听客户端失败: %v\n", err) return } consulKey := fmt.Sprintf("config/%s/%s-%s", serviceName, serviceName, env) for { opts := &api.QueryOptions{ WaitIndex: lastIndex, WaitTime: 60 * time.Second, } pair, meta, err := client.KV().Get(consulKey, opts) if err != nil { fmt.Printf("Consul 监听出错: %v, 5秒后重试...\n", err) time.Sleep(5 * time.Second) continue } if meta.LastIndex == lastIndex { continue } if pair == nil { fmt.Printf("Consul 配置被删除: %s\n", consulKey) lastIndex = meta.LastIndex continue } fmt.Printf("检测到 Consul 配置变更: %s, New Index: %d\n", consulKey, meta.LastIndex) updateLocalConfig(pair.Value) lastIndex = meta.LastIndex } } func updateLocalConfig(content []byte) { ctx := context.Background() oldConfig, _ := g.Cfg().Data(ctx) adapter, err := gcfg.NewAdapterContent() if err != nil { fmt.Printf("创建新配置适配器失败: %v\n", err) return } adapter.SetContent(string(content)) g.Cfg().SetAdapter(adapter) newConfig, _ := g.Cfg().Data(ctx) changelog, err := diff.Diff(oldConfig, newConfig) if err != nil { fmt.Printf("配置对比失败: %v\n", err) return } if len(changelog) == 0 { fmt.Printf("配置已热更新成功(内容无实质变化)\n") return } fmt.Printf("=== 检测到配置变更 (%d 项) ===\n", len(changelog)) for _, change := range changelog { switch change.Type { case "create": fmt.Printf("[+] 新增: %s = %v\n", change.Path, change.To) case "update": fmt.Printf("[~] 修改: %s: %v -> %v\n", change.Path, change.From, change.To) case "delete": fmt.Printf("[-] 删除: %s (原值: %v)\n", change.Path, change.From) } } fmt.Printf("=================================\n") } func getLocalIP() (string, error) { // 获取本机所有网络接口 addrs, err := net.InterfaceAddrs() if err != nil { return "", err } for _, addr := range addrs { // 检查是否是IP地址 if ipNet, ok := addr.(*net.IPNet); ok && !ipNet.IP.IsLoopback() { if ipNet.IP.To4() != nil { // 返回第一个非回环的IPv4地址 return ipNet.IP.String(), nil } } } return "", fmt.Errorf("无法找到本地IP地址") } func getInstanceAddrByIp(ctx context.Context, ip string, services []gsvc.Service) (addr string) { for _, s := range services { if s.GetEndpoints()[0].Host() == ip { addr = s.GetEndpoints()[0].String() return } } return } func GetInstanceAddr(ctx context.Context, name string) (addr string, err error) { watch, err := gsvc.GetRegistry().Watch(ctx, name) if err != nil { err = errors.New("获取服务监听器失败") return } service, err := watch.Proceed() if err != nil || service == nil { err = errors.New("获取服务实例失败") return } //优先使用客户端IP获取实例(前后端在同一台机器调试) addr = getInstanceAddrByIp(ctx, g.RequestFromCtx(ctx).GetClientIp(), service) if !g.IsEmpty(addr) { return } //优先使用gateway同IP的服务实例(前后端不同机器调试) addr, err = getLocalIP() if err != nil { return } addr = getInstanceAddrByIp(ctx, addr, service) if !g.IsEmpty(addr) { return } //随机获取一个服务实例 maxService := grand.N(0, len(service)-1) addr = service[maxService].GetEndpoints()[0].String() return }