透过Kratos源码，强化Go语言学习（四）：Config分层抽象与并发安全设计

type Config interface {
    Load() error                    // 加载所有配置源
    Scan(v any) error              // 扫描配置到结构体
    Value(key string) Value        // 获取指定键的配置值
    Watch(key string, o Observer) error  // 监听配置变化
    Close() error                  // 关闭配置管理器
}

type Source interface {
    Load() ([]*KeyValue, error)    // 加载配置数据
    Watch() (Watcher, error)       // 创建监听器
}

// Reader is config reader.
type Reader interface {
	Merge(...*KeyValue) error   // 将多个配置数据源合并到内部存储中
	Value(string) (Value, bool) // 根据路径获取配置值
	Source() ([]byte, error)    // 将当前配置数据序列化为字节数组
	Resolve() error             // 解析配置中的占位符和变量引用
}

// Value 接口定义了配置值的统一访问方式
type Value interface {
    Bool() (bool, error)      // 类型转换方法
    Int() (int64, error)
    Float() (float64, error)
    String() (string, error)
    Duration() (time.Duration, error)
    Slice() ([]Value, error)  // 复杂类型支持
    Map() (map[string]Value, error)
    Scan(any) error          // 结构体映射
    Load() any               // 原始值访问
    Store(any)               // 值存储
}

// Watcher watches a source for changes.
type Watcher interface {
	Next() ([]*KeyValue, error) // 阻塞等待并返回下一批配置变更
	Stop() error                // 停止监听并清理资源
}

type config struct {
    opts      options
    reader    Reader
    cached    sync.Map     // 并发安全的配置缓存
    observers sync.Map     // 并发安全的观察者存储
    watchers  []Watcher
}

func (c *config) watch(w Watcher) {
    for {
        kvs, err := w.Next()
        // ... 错误处理 ...
        
        // 原子更新配置缓存
        c.cached.Range(func(key, value any) bool {
            k := key.(string)
            v := value.(Value)
            if n, ok := c.reader.Value(k); ok && reflect.TypeOf(n.Load()) == reflect.TypeOf(v.Load()) && !reflect.DeepEqual(n.Load(), v.Load()) {
                v.Store(n.Load())  // 原子存储新值
                if o, ok := c.observers.Load(k); ok {
                    o.(Observer)(k, v)  // 通知观察者
                }
            }
            return true
        })
    }
}

func (c *config) Value(key string) Value {
    if v, ok := c.cached.Load(key); ok {  // 并发安全的读取
        return v.(Value)
    }
    if v, ok := c.reader.Value(key); ok {
        c.cached.Store(key, v)  // 并发安全的存储
        return v
    }
    return &errValue{err: ErrNotFound}
}

func (c *config) Watch(key string, o Observer) error {
    if v := c.Value(key); v.Load() == nil {
        return ErrNotFound
    }
    c.observers.Store(key, o)  // 并发安全的观察者注册
    return nil
}

if o, ok := c.observers.Load(k); ok {  // 并发安全的观察者获取
    o.(Observer)(k, v)  // 异步通知观察者
}

func (c *config) Load() error {
    for _, src := range c.opts.sources {
        // ... 加载配置 ...
        
        w, err := src.Watch()
        if err != nil {
            log.Errorf("failed to watch config source: %v", err)
            return err
        }
        c.watchers = append(c.watchers, w)
        go c.watch(w)  // 每个 Watcher 在独立协程中运行
    }
    return nil
}

func (c *config) Close() error {
    for _, w := range c.watchers {
        if err := w.Stop(); err != nil {  // 依次停止所有 Watcher
            return err
        }
    }
    return nil
}

func (c *config) watch(w Watcher) {
    for {
        kvs, err := w.Next()
        if err != nil {
            if errors.Is(err, context.Canceled) {
                log.Infof("watcher's ctx cancel : %v", err)
                return  // 优雅退出，不影响其他协程
            }
            time.Sleep(time.Second)
            log.Errorf("failed to watch next config: %v", err)
            continue  // 错误恢复，继续监听
        }
        // 处理配置变更的并发安全逻辑
    }
}

type config struct {
    reader Reader  // Reader 内部实现并发安全
}

func (c *config) Scan(v any) error {
    data, err := c.reader.Source()  // Reader 内部使用锁保护
    if err != nil {
        return err
    }
    return unmarshalJSON(data, v)
}