摘要：
=nil｛panic（err）｝//watchkeyifer:=c.Value（key）；=w、 Next（）iferors.Is（err，=nil｛time.Sleep（time.Second）c.log.Errorf（“failedwatchnextconfig:err）continue｝ifer:=c.reader.Resolve（）；
从入口文件看配置（初始化/加载/绑定/热加载）
main.go
//main.go 初始化配置
c := config.New(
config.WithSource(
file.NewSource(flagconf), //文件配置源
//也可以自己实现远程配置中心数据源
),
)
//加载配置数据
if err := c.Load(); err != nil {
panic(err)
}
var bc conf.Bootstrap
//将配置绑定到数据结构
if err := c.Scan(&bc); err != nil {
panic(err)
}
// watch key
if err := c.Watch("service.name", func(key string, value config.Value) {
log.Printf("config changed: %s = %v
", key, value)
}); err != nil {
panic(err)
}
由入口文件可以看到，首先声明了数据源，再使用Kratos的配置接口加载/绑定/监控等操作，大胆猜测，c.Load() - c.Scan(&bc) - c.Watch() 一定调用了配置源的相关方法
先来看看实现一个配置源需要实现哪些接口
//config/source.go
//通过config.WithSource 知道了 配置来源需要实现以下两个方法
type Source interface {
Load() ([]*KeyValue, error)
Watch() (Watcher, error)
}
暂时不去看配置源的具体实现，回到入口文件，看看 c.Load() 的具体实现
func (c *config) Load() error {
for _, src := range c.opts.sources {
kvs, err := src.Load() //这里验证了我们的猜测，调用了具体配置源的 Load() 方法
if err != nil {
return err
}
if err := c.reader.Merge(kvs...); err != nil {
c.log.Errorf("failed to merge config source: %v", err)
return err
}
w, err := src.Watch() //调用了具体配置源的 Watch() 方法
if err != nil {
c.log.Errorf("failed to watch config source: %v", err)
return err
}
c.watchers = append(c.watchers, w)
go c.watch(w) //此处开启了一个协程，处理热加载
}
if err := c.reader.Resolve(); err != nil {
c.log.Errorf("failed to resolve config source: %v", err)
return err
}
return nil
}
c.Load() 中开启了一个协程监控配置源变更，稍后我们看看这个具体实现，先看 c.Watch() 中发生了什么
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
}
好，现在回过头去看上面提到的协程
func (c *config) watch(w Watcher) {
for {
kvs, err := w.Next()
if errors.Is(err, context.Canceled) {
c.log.Infof("watcher's ctx cancel : %v", err)
return
}
if err != nil {
time.Sleep(time.Second)
c.log.Errorf("failed to watch next config: %v", err)
continue
}
if err := c.reader.Merge(kvs...); err != nil {
c.log.Errorf("failed to merge next config: %v", err)
continue
}
if err := c.reader.Resolve(); err != nil {
c.log.Errorf("failed to resolve next config: %v", err)
continue
}
c.cached.Range(func(key, value interface{}) bool {
k := key.(string)
v := value.(Value)
if n, ok := c.reader.Value(k); ok && !reflect.DeepEqual(n.Load(), v.Load()) {
v.Store(n.Load())
if o, ok := c.observers.Load(k); ok {
o.(Observer)(k, v)
}
}
return true
})
}
}
以上主要能看到 Kratos 的配置是怎样加载和监控的了，具体的细节还需要去看每一个方法的实现。这里我们主要讨论，怎样实现配置源
以文件配置源举例：
实现 Source 接口即可。
type Source interface {
Load() ([]*KeyValue, error)
Watch() (Watcher, error)
}
具体实现：
Load() ([]*KeyValue, error)
config/file/file.go
//本地文件或远程配置中心只要实现以上两个方法就可以，以本地文件配置为例
func (f *file) Load() (kvs []*config.KeyValue, err error) {
fi, err := os.Stat(f.path)
if err != nil {
return nil, err
}
if fi.IsDir() {
return f.loadDir(f.path)
}
kv, err := f.loadFile(f.path)
if err != nil {
return nil, err
}
return []*config.KeyValue{kv}, nil
}
//先来看loadFile
func (f *file) loadFile(path string) (*config.KeyValue, error) {
file, err := os.Open(path)
if err != nil {
return nil, err
}
defer file.Close()
data, err := ioutil.ReadAll(file)
if err != nil {
return nil, err
}
info, err := file.Stat()
if err != nil {
return nil, err
}
return &config.KeyValue{
Key: info.Name(),
Format: format(info.Name()),
Value: data,
}, nil
}
//loadDir 读取里循环调用了loadFile, 不支持子目录和隐藏文件
func (f *file) loadDir(path string) (kvs []*config.KeyValue, err error) {
files, err := ioutil.ReadDir(f.path)
if err != nil {
return nil, err
}
for _, file := range files {
// ignore hidden files
if file.IsDir() || strings.HasPrefix(file.Name(), ".") {
continue
}
kv, err := f.loadFile(filepath.Join(f.path, file.Name()))
if err != nil {
return nil, err
}
kvs = append(kvs, kv)
}
return
}
Watch() (Watcher, error)
config/file/file.go
func (f *file) Watch() (config.Watcher, error) {
return newWatcher(f)
}
config/source.go Watcher的定义如下
// Watcher watches a source for changes.
type Watcher interface {
Next() ([]*KeyValue, error)
Stop() error
}
config/file/watcher.go
//Next() ([]*KeyValue, error) 实现
func (w *watcher) Next() ([]*config.KeyValue, error) {
select {
case <-w.ctx.Done():
return nil, w.ctx.Err()
case event := <-w.fw.Events:
if event.Op == fsnotify.Rename {
if _, err := os.Stat(event.Name); err == nil || os.IsExist(err) {
if err := w.fw.Add(event.Name); err != nil {
return nil, err
}
}
}
fi, err := os.Stat(w.f.path)
if err != nil {
return nil, err
}
path := w.f.path
if fi.IsDir() {
path = filepath.Join(w.f.path, filepath.Base(event.Name))
}
kv, err := w.f.loadFile(path)
if err != nil {
return nil, err
}
return []*config.KeyValue{kv}, nil
case err := <-w.fw.Errors:
return nil, err
}
}
//Stop() error 实现
func (w *watcher) Stop() error {
w.cancel()
return w.fw.Close()
}
func newWatcher(f *file) (config.Watcher, error) {
fw, err := fsnotify.NewWatcher()
if err != nil {
return nil, err
}
if err := fw.Add(f.path); err != nil {
return nil, err
}
ctx, cancel := context.WithCancel(context.Background())
return &watcher{f: f, fw: fw, ctx: ctx, cancel: cancel}, nil
}