40分钟学 Go 语言高并发：【实战】并发安全的配置管理器（功能扩展）

【实战】并发安全的配置管理器（功能扩展）

一、扩展思考

1. 分布式配置中心

• 实现配置的集中管理
• 支持多节点配置同步
• 实现配置的版本一致性

2. 配置加密

• 敏感配置的加密存储
• 配置的安全传输
• 访问权限控制

3. 配置格式支持

• 支持YAML、TOML等多种格式
• 配置格式自动识别和转换
• 支持环境变量替换

让我们实现配置格式转换的功能：

package configmanager

import (
    "encoding/json"
    "fmt"
    "gopkg.in/yaml.v2"
    "strings"
)

// FormatConverter 配置格式转换器
type FormatConverter struct {
    supportedFormats map[string]bool
}

func NewFormatConverter() *FormatConverter {
    return &FormatConverter{
        supportedFormats: map[string]bool{
            "json": true,
            "yaml": true,
            "yml":  true,
        },
    }
}

// ConvertToFormat 将配置转换为指定格式
func (fc *FormatConverter) ConvertToFormat(config Config, format string) ([]byte, error) {
    format = strings.ToLower(format)
    if !fc.supportedFormats[format] {
        return nil, fmt.Errorf("不支持的格式: %s", format)
    }

    switch format {
    case "json":
        return json.MarshalIndent(config, "", "    ")
    case "yaml", "yml":
        return yaml.Marshal(config)
    default:
        return nil, fmt.Errorf("未知格式: %s", format)
    }
}

// ParseFromFormat 从指定格式解析配置
func (fc *FormatConverter) ParseFromFormat(data []byte, format string) (*Config, error) {
    format = strings.ToLower(format)
    if !fc.supportedFormats[format] {
        return nil, fmt.Errorf("不支持的格式: %s", format)
    }

    var config Config
    var err error

    switch format {
    case "json":
        err = json.Unmarshal(data, &config)
    case "yaml", "yml":
        err = yaml.Unmarshal(data, &config)
    default:
        return nil, fmt.Errorf("未知格式: %s", format)
    }

    if err != nil {
        return nil, fmt.Errorf("解析%s格式失败: %v", format, err)
    }

    return &config, nil
}

// ConfigManager添加格式转换支持
func (cm *ConfigManager) ExportToFormat(format string) ([]byte, error) {
    cm.mu.RLock()
    defer cm.mu.RUnlock()

    converter := NewFormatConverter()
    return converter.ConvertToFormat(cm.config, format)
}

// LoadFromFormatted 从指定格式的数据加载配置
func (cm *ConfigManager) LoadFromFormatted(data []byte, format string) error {
    converter := NewFormatConverter()
    config, err := converter.ParseFromFormat(data, format)
    if err != nil {
        return err
    }

    cm.UpdateConfig(*config)
    return nil
}

二、环境变量支持

实现配置中环境变量的替换功能：

package configmanager

import (
    "fmt"
    "os"
    "regexp"
    "strings"
)

// EnvVarResolver 环境变量解析器
type EnvVarResolver struct {
    pattern *regexp.Regexp
}

func NewEnvVarResolver() *EnvVarResolver {
    return &EnvVarResolver{
        pattern: regexp.MustCompile(`\$\{([^}]+)}|\$([A-Za-z0-9_]+)`),
    }
}

// ResolveEnvVars 解析配置中的环境变量
func (r *EnvVarResolver) ResolveEnvVars(config Config) Config {
    newConfig := Config{
        Version:   config.Version,
        UpdatedAt: config.UpdatedAt,
        Data:      make(map[string]interface{}),
    }

    for key, value := range config.Data {
        newConfig.Data[key] = r.resolveValue(value)
    }

    return newConfig
}

// resolveValue 解析值中的环境变量
func (r *EnvVarResolver) resolveValue(value interface{}) interface{} {
    switch v := value.(type) {
    case string:
        return r.resolveString(v)
    case map[string]interface{}:
        newMap := make(map[string]interface{})
        for k, val := range v {
            newMap[k] = r.resolveValue(val)
        }
        return newMap
    case []interface{}:
        newSlice := make([]interface{}, len(v))
        for i, val := range v {
            newSlice[i] = r.resolveValue(val)
        }
        return newSlice
    default:
        return v
    }
}

// resolveString 解析字符串中的环境变量
func (r *EnvVarResolver) resolveString(s string) string {
    result := r.pattern.ReplaceAllStringFunc(s, func(match string) string {
        var envVar string
        if strings.HasPrefix(match, "${") {
            envVar = match[2 : len(match)-1]
        } else {
            envVar = match[1:]
        }

        if value, exists := os.LookupEnv(envVar); exists {
            return value
        }
        return match
    })
    return result
}

// ConfigManager添加环境变量支持
func (cm *ConfigManager) LoadWithEnvVars(config Config) {
    resolver := NewEnvVarResolver()
    resolvedConfig := resolver.ResolveEnvVars(config)
    cm.UpdateConfig(resolvedConfig)
}

// Example usage of environment variables in configuration:
var exampleConfigWithEnv = `
{
    "version": 1,
    "data": {
        "app_name": "my_service",
        "db_host": "${DB_HOST}",
        "db_port": "${DB_PORT}",
        "api_key": "$API_KEY",
        "environment": "${ENV:-production}",
        "paths": {
            "data": "${DATA_PATH:-/var/data}",
            "logs": "${LOG_PATH:-/var/log}"
        }
    }
}
`

三、加密配置支持

实现敏感配置的加密存储功能：

package configmanager

import (
    "crypto/aes"
    "crypto/cipher"
    "crypto/rand"
    "encoding/base64"
    "fmt"
    "io"
)

// ConfigEncryption 配置加密器
type ConfigEncryption struct {
    key []byte
}

func NewConfigEncryption(key string) (*ConfigEncryption, error) {
    if len(key) != 32 {
        return nil, fmt.Errorf("密钥长度必须为32字节")
    }
    return &ConfigEncryption{key: []byte(key)}, nil
}

// Encrypt 加密敏感配置值
func (ce *ConfigEncryption) Encrypt(value string) (string, error) {
    block, err := aes.NewCipher(ce.key)
    if err != nil {
        return "", err
    }

    plaintext := []byte(value)
    ciphertext := make([]byte, aes.BlockSize+len(plaintext))
    iv := ciphertext[:aes.BlockSize]
    if _, err := io.ReadFull(rand.Reader, iv); err != nil {
        return "", err
    }

    stream := cipher.NewCFBEncrypter(block, iv)
    stream.XORKeyStream(ciphertext[aes.BlockSize:], plaintext)

    return base64.StdEncoding.EncodeToString(ciphertext), nil
}

// Decrypt 解密敏感配置值
func (ce *ConfigEncryption) Decrypt(encrypted string) (string, error) {
    block, err := aes.NewCipher(ce.key)
    if err != nil {
        return "", err
    }

    ciphertext, err := base64.StdEncoding.DecodeString(encrypted)
    if err != nil {
        return "", err
    }

    if len(ciphertext) < aes.BlockSize {
        return "", fmt.Errorf("加密文本太短")
    }

    iv := ciphertext[:aes.BlockSize]
    ciphertext = ciphertext[aes.BlockSize:]

    stream := cipher.NewCFBDecrypter(block, iv)
    stream.XORKeyStream(ciphertext, ciphertext)

    return string(ciphertext), nil
}

// 在ConfigManager中添加加密支持
type EncryptedField struct {
    Value     string `json:"value"`
    Encrypted bool   `json:"encrypted"`
}

func (cm *ConfigManager) EncryptField(key string, value string) error {
    if cm.encryption == nil {
        return fmt.Errorf("未配置加密器")
    }

    encrypted, err := cm.encryption.Encrypt(value)
    if err != nil {
        return err
    }

    cm.mu.Lock()
    defer cm.mu.Unlock()

    cm.config.Data[key] = EncryptedField{
        Value:     encrypted,
        Encrypted: true,
    }
    return nil
}

func (cm *ConfigManager) GetDecryptedValue(key string) (string, error) {
    if cm.encryption == nil {
        return "", fmt.Errorf("未配置加密器")
    }

    cm.mu.RLock()
    defer cm.mu.RUnlock()

    field, ok := cm.config.Data[key].(EncryptedField)
    if !ok {
        return "", fmt.Errorf("字段不是加密字段")
    }

    if !field.Encrypted {
        return field.Value, nil
    }

    return cm.encryption.Decrypt(field.Value)
}

// 示例加密配置的使用
func ExampleEncryption() {
    // 创建32字节的密钥
    key := "12345678901234567890123456789012"
    encryption, _ := NewConfigEncryption(key)
    
    cm := NewConfigManager(5)
    cm.encryption = encryption

    // 加密敏感配置
    cm.EncryptField("api_key", "secret_key_123")
    cm.EncryptField("db_password", "db_password_456")

    // 读取解密后的配置
    apiKey, _ := cm.GetDecryptedValue("api_key")
    dbPassword, _ := cm.GetDecryptedValue("db_password")

    fmt.Printf("Decrypted API Key: %s\n", apiKey)
    fmt.Printf("Decrypted DB Password: %s\n", dbPassword)
}

四、分布式配置支持

为配置管理器添加分布式支持：

package configmanager

import (
    "context"
    "encoding/json"
    "fmt"
    "sync"
    "time"
)

// ConfigNode 分布式节点接口
type ConfigNode interface {
    // 获取配置
    GetConfig() (Config, error)
    // 更新配置
    UpdateConfig(Config) error
    // 监听配置变更
    WatchConfig(context.Context) (<-chan Config, error)
    // 获取节点ID
    GetNodeID() string
}

// DistributedConfigManager 分布式配置管理器
type DistributedConfigManager struct {
    *ConfigManager
    node          ConfigNode
    watchContext  context.Context
    watchCancel   context.CancelFunc
    syncInterval  time.Duration
    lastSyncTime  time.Time
    syncMu        sync.RWMutex
}

func NewDistributedConfigManager(node ConfigNode, syncInterval time.Duration) *DistributedConfigManager {
    ctx, cancel := context.WithCancel(context.Background())
    dcm := &DistributedConfigManager{
        ConfigManager: NewConfigManager(5),
        node:         node,
        watchContext: ctx,
        watchCancel:  cancel,
        syncInterval: syncInterval,
    }

    // 启动配置同步
    go dcm.startSync()
    // 启动配置监听
    go dcm.startWatch()

    return dcm
}

// startSync 开始配置同步
func (dcm *DistributedConfigManager) startSync() {
    ticker := time.NewTicker(dcm.syncInterval)
    defer ticker.Stop()

    for {
        select {
        case <-dcm.watchContext.Done():
            return
        case <-ticker.C:
            if err := dcm.syncConfig(); err != nil {
                log.Printf("配置同步失败: %v", err)
            }
        }
    }
}

// syncConfig 同步配置
func (dcm *DistributedConfigManager) syncConfig() error {
    dcm.syncMu.Lock()
    defer dcm.syncMu.Unlock()

    // 获取远程配置
    remoteConfig, err := dcm.node.GetConfig()
    if err != nil {
        return fmt.Errorf("获取远程配置失败: %v", err)
    }

    // 检查版本
    if remoteConfig.Version > dcm.config.Version {
        // 更新本地配置
        dcm.UpdateConfig(remoteConfig)
        dcm.lastSyncTime = time.Now()

package configmanager

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"
)

// 继续 DistributedConfigManager 的实现

// startWatch 开始监听配置变更
func (dcm *DistributedConfigManager) startWatch() {
    configChan, err := dcm.node.WatchConfig(dcm.watchContext)
    if err != nil {
        log.Printf("启动配置监听失败: %v", err)
        return
    }

    for {
        select {
        case <-dcm.watchContext.Done():
            return
        case newConfig := <-configChan:
            dcm.handleConfigChange(newConfig)
        }
    }
}

// handleConfigChange 处理配置变更
func (dcm *DistributedConfigManager) handleConfigChange(newConfig Config) {
    dcm.syncMu.Lock()
    defer dcm.syncMu.Unlock()

    // 检查版本
    if newConfig.Version > dcm.config.Version {
        dcm.UpdateConfig(newConfig)
        log.Printf("配置已更新到版本 %d", newConfig.Version)
    }
}

// UpdateConfig 重写更新配置方法，同步到远程
func (dcm *DistributedConfigManager) UpdateConfig(newConfig Config) error {
    dcm.syncMu.Lock()
    defer dcm.syncMu.Unlock()

    // 先更新远程配置
    if err := dcm.node.UpdateConfig(newConfig); err != nil {
        return fmt.Errorf("更新远程配置失败: %v", err)
    }

    // 更新本地配置
    dcm.ConfigManager.UpdateConfig(newConfig)
    return nil
}

// GetLastSyncTime 获取最后同步时间
func (dcm *DistributedConfigManager) GetLastSyncTime() time.Time {
    dcm.syncMu.RLock()
    defer dcm.syncMu.RUnlock()
    return dcm.lastSyncTime
}

// Stop 停止分布式配置管理器
func (dcm *DistributedConfigManager) Stop() {
    dcm.watchCancel()
}

// Redis节点实现示例
type RedisConfigNode struct {
    client     *redis.Client
    nodeID     string
    configKey  string
    versionKey string
}

func NewRedisConfigNode(addr, nodeID string) *RedisConfigNode {
    client := redis.NewClient(&redis.Options{
        Addr: addr,
    })
    
    return &RedisConfigNode{
        client:     client,
        nodeID:     nodeID,
        configKey:  "distributed_config",
        versionKey: "config_version",
    }
}

func (n *RedisConfigNode) GetConfig() (Config, error) {
    data, err := n.client.Get(n.configKey).Bytes()
    if err != nil {
        return Config{}, err
    }

    var config Config
    if err := json.Unmarshal(data, &config); err != nil {
        return Config{}, err
    }

    return config, nil
}

func (n *RedisConfigNode) UpdateConfig(config Config) error {
    data, err := json.Marshal(config)
    if err != nil {
        return err
    }

    // 使用事务确保原子性
    pipe := n.client.TxPipeline()
    pipe.Set(n.configKey, data, 0)
    pipe.Set(n.versionKey, config.Version, 0)
    
    _, err = pipe.Exec()
    return err
}

func (n *RedisConfigNode) WatchConfig(ctx context.Context) (<-chan Config, error) {
    configChan := make(chan Config)
    
    go func() {
        defer close(configChan)
        pubsub := n.client.Subscribe(n.configKey + "_changes")
        defer pubsub.Close()

        for {
            select {
            case <-ctx.Done():
                return
            case msg := <-pubsub.Channel():
                var config Config
                if err := json.Unmarshal([]byte(msg.Payload), &config); err != nil {
                    log.Printf("解析配置消息失败: %v", err)
                    continue
                }
                configChan <- config
            }
        }
    }()

    return configChan, nil
}

func (n *RedisConfigNode) GetNodeID() string {
    return n.nodeID
}

// 使用示例
func ExampleDistributed() {
    // 创建Redis节点
    node := NewRedisConfigNode("localhost:6379", "node1")
    
    // 创建分布式配置管理器
    dcm := NewDistributedConfigManager(node, 5*time.Second)
    defer dcm.Stop()

    // 更新配置
    config := Config{
        Version: 1,
        Data: map[string]interface{}{
            "app_name": "distributed_app",
            "port":     8080,
        },
    }
    
    if err := dcm.UpdateConfig(config); err != nil {
        log.Printf("更新配置失败: %v", err)
        return
    }

    // 监听配置变更
    observer := &ConfigObserver{name: "DistributedObserver"}
    dcm.Subscribe(observer)

    // 运行一段时间
    time.Sleep(30 * time.Second)
}

让我们继续完成分布式配置管理器的实现。

五、配置生命周期管理

让我们实现配置的生命周期管理功能：

package configmanager

import (
    "fmt"
    "time"
)

// ConfigLifecycle 配置生命周期管理
type ConfigLifecycle struct {
    ExpiresAt    time.Time          `json:"expires_at,omitempty"`
    ValidFrom    time.Time          `json:"valid_from,omitempty"`
    Environment  string             `json:"environment"`
    Dependencies map[string]string  `json:"dependencies,omitempty"`
    Tags         []string           `json:"tags,omitempty"`
}

// ConfigWithLifecycle 带生命周期的配置
type ConfigWithLifecycle struct {
    Config
    Lifecycle ConfigLifecycle `json:"lifecycle"`
}

// LifecycleManager 生命周期管理器
type LifecycleManager struct {
    currentEnv string
}

func NewLifecycleManager(env string) *LifecycleManager {
    return &LifecycleManager{
        currentEnv: env,
    }
}

// ValidateConfig 验证配置生命周期
func (lm *LifecycleManager) ValidateConfig(config ConfigWithLifecycle) error {
    // 验证环境
    if config.Lifecycle.Environment != "" && 
       config.Lifecycle.Environment != lm.currentEnv {
        return fmt.Errorf("配置环境不匹配: 期望 %s, 实际 %s",
            config.Lifecycle.Environment, lm.currentEnv)
    }

    // 验证时间有效性
    now := time.Now()
    if !config.Lifecycle.ValidFrom.IsZero() && now.Before(config.Lifecycle.ValidFrom) {
        return fmt.Errorf("配置尚未生效, 生效时间: %v", config.Lifecycle.ValidFrom)
    }

    if !config.Lifecycle.ExpiresAt.IsZero() && now.After(config.Lifecycle.ExpiresAt) {
        return fmt.Errorf("配置已过期, 过期时间: %v", config.Lifecycle.ExpiresAt)
    }

    return nil
}

// 扩展ConfigManager支持生命周期管理
type LifecycleConfigManager struct {
    *ConfigManager
    lifecycleManager *LifecycleManager
}

func NewLifecycleConfigManager(env string, maxVersions int) *LifecycleConfigManager {
    return &LifecycleConfigManager{
        ConfigManager:    NewConfigManager(maxVersions),
        lifecycleManager: NewLifecycleManager(env),
    }
}

// UpdateConfigWithLifecycle 更新带生命周期的配置
func (lcm *LifecycleConfigManager) UpdateConfigWithLifecycle(config ConfigWithLifecycle) error {
    // 验证生命周期
    if err := lcm.lifecycleManager.ValidateConfig(config); err != nil {
        return err
    }

    // 更新配置
    lcm.UpdateConfig(config.Config)
    return nil
}

// 示例使用
func ExampleLifecycle() {
    // 创建生命周期配置管理器
    lcm := NewLifecycleConfigManager("production", 5)

    // 创建带生命周期的配置
    config := ConfigWithLifecycle{
        Config: Config{
            Version: 1,
            Data: map[string]interface{}{
                "feature_flags": map[string]bool{
                    "new_feature": true,
                },
            },
        },
        Lifecycle: ConfigLifecycle{
            ValidFrom:    time.Now().Add(-24 * time.Hour),
            ExpiresAt:    time.Now().Add(7 * 24 * time.Hour),
            Environment:  "production",
            Dependencies: map[string]string{
                "service_a": ">=1.0.0",
                "service_b": ">=2.0.0",
            },
            Tags: []string{"feature_release", "v1.0"},
        },
    }

    // 更新配置
    if err := lcm.UpdateConfigWithLifecycle(config); err != nil {
        log.Printf("更新配置失败: %v", err)
        return
    }

    // 使用配置
    if val, exists := lcm.GetValue("feature_flags"); exists {
        log.Printf("特性开关: %v", val)
    }
}

// ConfigValidator 配置验证器
type ConfigValidator struct {
    rules map[string]ValidateFunc
}

type ValidateFunc func(interface{}) error

func NewConfigValidator() *ConfigValidator {
    return &ConfigValidator{
        rules: make(map[string]ValidateFunc),
    }
}

// AddRule 添加验证规则
func (cv *ConfigValidator) AddRule(key string, rule ValidateFunc) {
    cv.rules[key] = rule
}

// Validate 验证配置
func (cv *ConfigValidator) Validate(config Config) error {
    for key, rule := range cv.rules {
        if value, exists := config.Data[key]; exists {
            if err := rule(value); err != nil {
                return fmt.Errorf("配置项 %s 验证失败: %v", key, err)
            }
        }
    }
    return nil
}

// 示例验证规则
var (
    validatePort = func(v interface{}) error {
        port, ok := v.(float64)
        if !ok {
            return fmt.Errorf("端口必须是数字")
        }
        if port < 1 || port > 65535 {
            return fmt.Errorf("端口必须在1-65535之间")
        }
        return nil
    }

    validateString = func(v interface{}) error {
        _, ok := v.(string)
        if !ok {
            return fmt.Errorf("值必须是字符串")
        }
        return nil
    }
)

六、总结与最佳实践建议

让我们用一个流程图来总结配置管理器的完整功能：

使用建议：

1. 初始化配置

• 使用环境变量设置基础配置
• 在启动时进行配置验证
• 设置合理的默认值

2. 配置更新

• 实现优雅的热更新机制
• 保证更新操作的原子性
• 做好更新失败的回滚机制

3. 安全性

• 加密敏感配置信息
• 实现访问权限控制
• 保护配置历史记录

4. 监控与告警

• 记录配置变更日志
• 设置关键配置监控
• 配置异常告警机制

5. 性能优化

• 使用本地缓存
• 异步处理配置更新通知
• 合理设置更新检查间隔

6. 容错处理

• 配置解析异常处理
• 实现配置备份机制
• 提供服务降级策略

七、进阶功能实现

让我们实现一个完整的配置管理服务：

package configmanager

import (
    "context"
    "sync"
    "time"
)

// ConfigService 配置管理服务
type ConfigService struct {
    manager     *ConfigManager
    distributed *DistributedConfigManager
    lifecycle   *LifecycleManager
    validator   *ConfigValidator
    encryption  *ConfigEncryption
    metrics     *MetricsCollector
    watcher     *ConfigWatcher
    
    // 服务状态
    status     ServiceStatus
    statusMu   sync.RWMutex
    ctx        context.Context
    cancelFunc context.CancelFunc
}

// ServiceStatus 服务状态
type ServiceStatus struct {
    IsRunning    bool
    StartTime    time.Time
    LastError    error
    HealthStatus string
}

// ConfigServiceOptions 服务配置选项
type ConfigServiceOptions struct {
    Environment    string
    MaxVersions    int
    EncryptionKey  string
    SyncInterval   time.Duration
    WatchInterval  time.Duration
    ConfigFile     string
    DistributedURL string
}

// NewConfigService 创建配置管理服务
func NewConfigService(opts ConfigServiceOptions) (*ConfigService, error) {
    ctx, cancel := context.WithCancel(context.Background())
    
    service := &ConfigService{
        ctx:        ctx,
        cancelFunc: cancel,
    }

    // 初始化各个组件
    if err := service.initialize(opts); err != nil {
        cancel()
        return nil, err
    }

    return service, nil
}

// initialize 初始化服务组件
func (s *ConfigService) initialize(opts ConfigServiceOptions) error {
    // 初始化基础配置管理器
    s.manager = NewConfigManager(opts.MaxVersions)
    
    // 初始化生命周期管理器
    s.lifecycle = NewLifecycleManager(opts.Environment)
    
    // 初始化验证器
    s.validator = NewConfigValidator()
    s.addDefaultValidationRules()
    
    // 初始化加密组件
    if opts.EncryptionKey != "" {
        encryption, err := NewConfigEncryption(opts.EncryptionKey)
        if err != nil {
            return err
        }
        s.encryption = encryption
    }
    
    // 初始化监控指标收集器
    s.metrics = NewMetricsCollector()
    
    // 初始化配置文件监控
    if opts.ConfigFile != "" {
        watcher, err := s.manager.StartFileWatcher(opts.ConfigFile, opts.WatchInterval)
        if err != nil {
            return err
        }
        s.watcher = watcher
    }
    
    // 初始化分布式支持
    if opts.DistributedURL != "" {
        node := NewRedisConfigNode(opts.DistributedURL, opts.Environment)
        s.distributed = NewDistributedConfigManager(node, opts.SyncInterval)
    }
    
    return nil
}

// Start 启动服务
func (s *ConfigService) Start() error {
    s.statusMu.Lock()
    defer s.statusMu.Unlock()
    
    s.status = ServiceStatus{
        IsRunning:    true,
        StartTime:    time.Now(),
        HealthStatus: "running",
    }
    
    // 启动健康检查
    go s.healthCheck()
    
    return nil
}

// Stop 停止服务
func (s *ConfigService) Stop() {
    s.statusMu.Lock()
    defer s.statusMu.Unlock()
    
    s.status.IsRunning = false
    s.status.HealthStatus = "stopped"
    
    if s.watcher != nil {
        s.watcher.Stop()
    }
    
    if s.distributed != nil {
        s.distributed.Stop()
    }
    
    s.cancelFunc()
}

// healthCheck 健康检查
func (s *ConfigService) healthCheck() {
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-s.ctx.Done():
            return
        case <-ticker.C:
            s.checkHealth()
        }
    }
}

// checkHealth 执行健康检查
func (s *ConfigService) checkHealth() {
    s.statusMu.Lock()
    defer s.statusMu.Unlock()
    
    // 检查各组件状态
    if s.manager == nil {
        s.status.HealthStatus = "error"
        s.status.LastError = fmt.Errorf("配置管理器未初始化")
        return
    }
    
    // 检查分布式节点连接
    if s.distributed != nil {
        if time.Since(s.distributed.GetLastSyncTime()) > 5*time.Minute {
            s.status.HealthStatus = "warning"
            s.status.LastError = fmt.Errorf("分布式节点同步超时")
            return
        }
    }
    
    s.status.HealthStatus = "healthy"
    s.status.LastError = nil
}

// GetStatus 获取服务状态
func (s *ConfigService) GetStatus() ServiceStatus {
    s.statusMu.RLock()
    defer s.statusMu.RUnlock()
    return s.status
}

// UpdateConfig 更新配置
func (s *ConfigService) UpdateConfig(config ConfigWithLifecycle) error {
    // 验证配置生命周期
    if err := s.lifecycle.ValidateConfig(config); err != nil {
        return err
    }
    
    // 验证配置内容
    if err := s.validator.Validate(config.Config); err != nil {
        return err
    }
    
    // 更新配置
    if s.distributed != nil {
        return s.distributed.UpdateConfig(config.Config)
    }
    
    return s.manager.UpdateConfig(config.Config)
}

// addDefaultValidationRules 添加默认验证规则
func (s *ConfigService) addDefaultValidationRules() {
    s.validator.AddRule("port", validatePort)
    s.validator.AddRule("host", validateString)
    // 添加其他默认规则
}

八、使用示例

让我们创建一个完整的使用示例：

package main

import (
    "fmt"
    "log"
    "time"
)

func main() {
    // 创建服务配置选项
    opts := ConfigServiceOptions{
        Environment:    "production",
        MaxVersions:    10,
        EncryptionKey:  "12345678901234567890123456789012", // 32字节密钥
        SyncInterval:   5 * time.Second,
        WatchInterval:  1 * time.Second,
        ConfigFile:     "config.json",
        DistributedURL: "localhost:6379",
    }

    // 创建配置服务
    service, err := NewConfigService(opts)
    if err != nil {
        log.Fatalf("创建配置服务失败: %v", err)
    }

    // 启动服务
    if err := service.Start(); err != nil {
        log.Fatalf("启动服务失败: %v", err)
    }
    defer service.Stop()

    // 创建示例配置
    config := ConfigWithLifecycle{
        Config: Config{
            Version: 1,
            Data: map[string]interface{}{
                "app": map[string]interface{}{
                    "name":    "example_app",
                    "port":    8080,
                    "version": "1.0.0",
                },
                "database": map[string]interface{}{
                    "host":     "localhost",
                    "port":     5432,
                    "username": "admin",
                    "password": "${DB_PASSWORD}",
                },
                "cache": map[string]interface{}{
                    "enabled":     true,
                    "ttl":        300,
                    "max_size_mb": 1024,
                },
            },
        },
        Lifecycle: ConfigLifecycle{
            ValidFrom:   time.Now(),
            ExpiresAt:   time.Now().Add(24 * time.Hour),
            Environment: "production",
            Tags:        []string{"v1.0", "stable"},
        },
    }

    // 更新配置
    if err := service.UpdateConfig(config); err != nil {
        log.Printf("更新配置失败: %v", err)
    }

    // 监控服务状态
    go func() {
        ticker := time.NewTicker(1 * time.Second)
        defer ticker.Stop()

        for {
            select {
            case <-ticker.C:
                status := service.GetStatus()
                fmt.Printf("服务状态: %+v\n", status)
            }
        }
    }()

    // 运行一段时间
    time.Sleep(30 * time.Second)
}

// 输出运行结果
func printStatus(status ServiceStatus) {
    fmt.Printf("运行状态: %v\n", status.IsRunning)
    fmt.Printf("启动时间: %v\n", status.StartTime)
    fmt.Printf("健康状态: %v\n", status.HealthStatus)
    if status.LastError != nil {
        fmt.Printf("最后错误: %v\n", status.LastError)
    }
}

九、项目结构

推荐的项目目录结构：

configmanager/
├── cmd/
│   └── configservice/
│       └── main.go
├── internal/
│   ├── config/
│   │   ├── manager.go
│   │   ├── distributed.go
│   │   ├── lifecycle.go
│   │   ├── encryption.go
│   │   └── validator.go
│   ├── storage/
│   │   ├── redis.go
│   │   └── file.go
│   └── metrics/
│       └── collector.go
├── pkg/
│   └── configmanager/
│       ├── service.go
│       ├── types.go
│       └── options.go
└── examples/
    ├── basic/
    ├── distributed/
    └── encryption/

十、最终建议

1. 部署建议

• 使用容器化部署
• 实现优雅关闭
• 配置定期备份
• 监控系统集成

2. 安全建议

• 定期轮换加密密钥
• 实现访问控制
• 审计日志记录
• 敏感信息保护

3. 性能建议

• 使用本地缓存
• 批量更新操作
• 异步通知机制
• 合理的超时设置

4. 可靠性建议

• 实现熔断机制
• 配置定期验证
• 自动化测试
• 灾难恢复计划

5. 扩展性建议

• 模块化设计
• 插件化架构
• 标准接口定义
• 版本兼容性

通过以上内容，我们实现了一个功能完整的配置管理器，它具备了：

• 并发安全
• 热更新支持
• 分布式部署
• 版本控制
• 配置加密
• 生命周期管理
• 监控指标收集
• 完整的测试覆盖

---

怎么样今天的内容还满意吗？再次感谢观众老爷的观看，关注GZH：凡人的AI工具箱，回复666，送您价值199的AI大礼包。最后，祝您早日实现财务自由，还请给个赞，谢谢！

原文地址：https://blog.csdn.net/weixin_40780178/article/details/144017017

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！