viper配置框架的介绍支持zookeeper的读取和监听

作者: Delimma隐
来源: 51数据库
2021-08-22

viper作为配置框架，其功能非常的强大，我们没有理由不去了解一下。我们先看官网对它的功能简介：

viper是完整配置解决方案，他可以处理所有类型和格式的配置文件，他有如下功能：

设置默认配置
支持读取 json toml yaml hcl 和 java 属性配置文件
监听配置文件变化，实时读取读取配置文件内容
读取环境变量值
读取远程配置系统 (etcd consul) 和监控配置变化
读取命令 flag 值
读取 buffer 值
读取确切值

乍一看，未免有相见恨晚之感，可仔细一想，不免脑袋里有另外一种声音：不会不支持读取 zookeeper 吧？好吧，至少我是这样的。

基于这种想法，当然要去立马尝试，如下：

viper.addremoteprovider("zookeeper", "xx.xx.xx.xx:2181", "/viper/test")

返回结果是：

unsupported remote provider type zookeeper

果不其然，于是追踪 viper.addremoteprovider 的源码，发现viper只支持如下几种

var supportedremoteproviders = []string{"etcd", "consul", "firestore"}

如果就此打住，未免有点太可惜，作为偏执狂，总想着能否来改造下viper，让其支持 zookeeper ，于是在issue上找是否有人遇到同样的问题，还整让我找到了，传送。但是不完整，且稍微有点bug。所以根据他的基础上，我做了些调整。进入正题，我们开始修改viper源码。说明下，我的viper版本是最新的 1.7.0

修改源码

1、添加zookeeper.go

添加的位置： github.com/bketelsen/crypt/zookeeper , zookeeper 目录需要自己创建， github.com/bketelsen/crypt 是viper的依赖包，会自动下载

文件内容：

package zookeeper

import (
	"errors"
	"fmt"
	zk "github.com/samuel/go-zookeeper/zk"
	//"github.com/xordataexchange/crypt/backend"
	"github.com/bketelsen/crypt/backend"
	"strings"
	"time"
)

type client struct {
	client *zk.conn
	waitindex uint64
}

func new(machines []string) (*client, error) {
	zkclient, _, err := zk.connect(machines, time.second)
	if err != nil {
		return nil, err
	}
	return &client{zkclient, 0}, nil
}

func (c *client) get(key string) ([]byte, error) {
	resp, _, err := c.client.get(key)
	if err != nil {
		return nil, err
	}
	return []byte(resp), nil
}

func nodewalk(prefix string, c *client, vars map[string]string) error {
	l, stat, err := c.client.children(prefix)
	if err != nil {
		return err
	}

	if stat.numchildren == 0 {
		b, _, err := c.client.get(prefix)
		if err != nil {
			return err
		}
		vars[prefix] = string(b)

	} else {
		for _, key := range l {
			s := prefix + "/" + key
			_, stat, err := c.client.exists(s)
			if err != nil {
				return err
			}
			if stat.numchildren == 0 {
				b, _, err := c.client.get(s)
				if err != nil {
					return err
				}
				vars[s] = string(b)
			} else {
				nodewalk(s, c, vars)
			}
		}
	}
	return nil
}

func (c *client) getvalues(key string, keys []string) (map[string]string, error) {
	vars := make(map[string]string)
	for _, v := range keys {
		v = fmt.sprintf("%s/%s", key, v)
		v = strings.replace(v, "/*", "", -1)
		_, _, err := c.client.exists(v)
		if err != nil {
			return vars, err
		}
		if v == "/" {
			v = ""
		}
		err = nodewalk(v, c, vars)
		if err != nil {
			return vars, err
		}
	}
	return vars, nil
}

func (c *client) list(key string) (backend.kvpairs, error) {
	var list backend.kvpairs
	resp, stat, err := c.client.children(key)
	if err != nil {
		return nil, err
	}

	if stat.numchildren == 0 {
		return list, nil
	}

	entries, err := c.getvalues(key, resp)
	if err != nil {
		return nil, err
	}

	for k, v := range entries {
		list = append(list, &backend.kvpair{key: k, value: []byte(v)})
	}
	return list, nil
}

func (c *client) createparents(key string) error {
	flags := int32(0)
	acl := zk.worldacl(zk.permall)

	if key[0] != '/' {
		return errors.new("invalid path")
	}

	payload := []byte("")
	pathstring := ""
	pathnodes := strings.split(key, "/")
	for i := 1; i < len(pathnodes); i++ {
		pathstring += "/" + pathnodes[i]
		_, err := c.client.create(pathstring, payload, flags, acl)
		// not being able to create the node because it exists or not having
		// sufficient rights is not an issue. it is ok for the node to already
		// exist and/or us to only have read rights
		if err != nil && err != zk.errnodeexists && err != zk.errnoauth {
			return err
		}
	}
	return nil
}

func (c *client) set(key string, value []byte) error {
	err := c.createparents(key)
	if err != nil {
		return err
	}
	_, err = c.client.set(key, []byte(value), -1)
	return err
}

func (c *client) watch(key string, stop chan bool) <-chan *backend.response {
	respchan := make(chan *backend.response, 0)
	go func() {
		for {
			resp, _, watch, err := c.client.getw(key)
			if err != nil {
				respchan <- &backend.response{nil, err}
				time.sleep(time.second * 5)
			}

			select {
			case e := <-watch:
				if e.type == zk.eventnodedatachanged {
					resp, _, err = c.client.get(key)
					if err != nil {
						respchan <- &backend.response{nil, err}
					}
					c.waitindex = 0
					respchan <- &backend.response{[]byte(resp), nil}
				}
			}
		}
	}()
	return respchan
}

这个文件是实现 configmanager 接口，我们在上图中看到 etcd ， consul ， filestore ，均有实现该接口，接口的定义很简单

type configmanager interface {
	get(key string) ([]byte, error)
	list(key string) (kvpairs, error)
	set(key string, value []byte) error
	watch(key string, stop chan bool) <-chan *response
}

2、修改config.go

文件的位置： github.com/bketelsen/crypt/config/config.go ，如下图

在 func newstandardetcdconfigmanager(machines []string) (configmanager, error) 方法下面添加如下方法：

// newstandardzookeeperconfigmanager returns a new configmanager backed by zookeeper.
// data will be encrypted.
func newstandardzookeeperconfigmanager(machines []string) (configmanager, error) {
	store, err := zookeeper.new(machines)
	if err != nil {
		return nil, err
	}
	return newstandardconfigmanager(store)
}

在 func newetcdconfigmanager(machines []string, keystore io.reader) (configmanager, error) 方法下面添加如下方法：

// newzookeeperconfigmanager returns a new configmanager backed by zookeeper.
// data will be encrypted.
func newzookeeperconfigmanager(machines []string, keystore io.reader) (configmanager, error) {
	store, err := zookeeper.new(machines)
	if err != nil {
		return nil, err
	}
	return newconfigmanager(store, keystore)
}

这两个方法是初始化 configmanager 对象，也就是我们刚才添加的 zookeeper.go 文件的对象

3、修改remote.go

文件的位置： github.com/spf13/viper/remote/remote.go ,如下图

找到74行，用下面的代码替换 func getconfigmanager(rp viper.remoteprovider) (crypt.configmanager, error) 方法

func getconfigmanager(rp viper.remoteprovider) (crypt.configmanager, error) {
	var cm crypt.configmanager
	var err error

	if rp.secretkeyring() != "" {
		var kr *os.file
		kr, err = os.open(rp.secretkeyring())
		if err != nil {
			return nil, err
		}
		defer kr.close()
		switch rp.provider() {
		case "etcd":
			cm, err = crypt.newetcdconfigmanager([]string{rp.endpoint()}, kr)
		case "zookeeper":
			cm, err = crypt.newzookeeperconfigmanager([]string{rp.endpoint()}, kr)
		case "firestore":
			cm, err = crypt.newfirestoreconfigmanager([]string{rp.endpoint()}, kr)
		default:
			cm, err = crypt.newconsulconfigmanager([]string{rp.endpoint()}, kr)
		}
	} else {
		switch rp.provider() {
		case "etcd":
			cm, err = crypt.newstandardetcdconfigmanager([]string{rp.endpoint()})
		case "zookeeper":
			cm, err = crypt.newstandardzookeeperconfigmanager([]string{rp.endpoint()})
		case "firestore":
			cm, err = crypt.newstandardfirestoreconfigmanager([]string{rp.endpoint()})
		default:
			cm, err = crypt.newstandardconsulconfigmanager([]string{rp.endpoint()})
		}
	}
	if err != nil {
		return nil, err
	}
	return cm, nil
}

细心的读者可能已经发现，其实就添加了两个case选项：

4、修改viper.go

文件的位置： github.com/spf13/viper/viper.go ,如下图

取+监听zookeeper(1)\image-20200521222843002.png)

找到两个 supportedremoteproviders 定义的定法，1.7.0版本的行号分别是：290，331。只要添加 zookeeper ,即可

supportedremoteproviders = []string{"etcd", "consul", "firestore", "zookeeper"}

好了，修改代码的工作已经完了，接下来我们来测试：

测试

注意：zookeeper中已经设置了内容

set /viper/test {"appname":"test","nodes":["127.0.0.1","127.0.0.2","127.0.0.3"]}

package main

import (
	"fmt"
	"github.com/fsnotify/fsnotify"
	"github.com/spf13/viper"
	_ "github.com/spf13/viper/remote"
	"time"
)


type config struct {
	appname string
	nodes []string
}

func main() {
	var waitgroup=sync.waitgroup{}
	waitgroup.add(1)
	readremotezookeeper()
	go watchremotezookeeper()
	waitgroup.wait()
}

func readremotezookeeper() {
	viper.addremoteprovider("zookeeper", "62.234.15.24:2181", "/viper/test")
	viper.setconfigtype("json") 
	err := viper.readremoteconfig()
	if err != nil {
		panic(fmt.sprintf("read remote zookeeper error:+%v", err))
	}

	var c config
	viper.unmarshal(&c)
	fmt.printf("从zookeeper读取配置内容:%+v\n", c)
}

func watchremotezookeeper() {
	go func() {
		for {
  //delay after each request
			time.sleep(time.second * 5)
			err := viper.watchremoteconfig()
			if err != nil {
				fmt.errorf("unable to read remote config: %v", err)
				continue
			}
			fmt.printf("从zookeeper读取更新内容:appname=%s,nodes=%+v\n", viper.get("appname"), viper.get("nodes"))
		}
	}()
}

输出内容：

从zookeeper读取配置内容:{appname:test nodes:[127.0.0.1 127.0.0.2 127.0.0.3]}
从zookeeper读取更新内容:appname=test,nodes=[127.0.0.1 127.0.0.2 127.0.0.3]

如果我们修改zookeeper的内容，则viper会读取到更新后的内容：

set /viper/test {"appname":"test","nodes":["127.0.0.1","127.0.0.2","127.0.0.3","127.0.0.4"]}

从zookeeper读取更新内容:appname=test,nodes=[127.0.0.1 127.0.0.2 127.0.0.3 127.0.0.4]

结语

让viper支持 zookeeper 并不复杂的，并且基本上不需要修改原有的方法，这要归结于viper用到一个非常重要的设计原则： 开闭原则 ，读者可以自行体会。

关于viper的基本使用，已经有非常详细的例子，这里就不再赘述，如有疑问，可以私信我

到此这篇关于viper配置框架的介绍支持zookeeper的读取和监听的文章就介绍到这了,更多相关viper配置框架支持zookeeper的读取和监听内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！