Go语言连接 zookeeper

介绍

Apache ZooKeeper 是 Apache 软件基金会的一个软件项目，他为大型分布式计算提供开源的分布式配置服务、同步服务和命名注册。ZooKeeper 曾经是 Hadoop 的一个子项目，但现在是一个独立的顶级项目。

ZooKeeper 的架构通过冗余服务实现高可用性。因此，如果第一次无应答，客户端就可以询问另一台 ZooKeeper 主机。ZooKeeper 节点将它们的数据存储于一个分层的命名空间，非常类似于一个文件系统或一个前缀树结构。客户端可以在节点读写，从而以这种方式拥有一个共享的配置服务。更新是全序的。

安装

Mac 系统，安装 zookeeper 只需要在终端运行如下命令：

brew install zookeeper

启动 zookeeper：

zkserver start

关闭 zookeeper：

zkserver stop

Go 操作 zookeeper

Go 操作 zookeeper 需要依赖第三方库，这里介绍 samuel/go-zookeeper，GitHub 地址如下：
https://github.com/samuel/go-zookeeper

下面写个简单的例子来测试下基本的操作：

/*** 获取一个zk连接* @return {[type]}*/
func getConnect(zkList []string) (conn *zk.Conn) {conn, _, err := zk.Connect(zkList, 10*time.Second)if err != nil {fmt.Println(err)}return
}/*** 测试连接* @return*/
func test1() {zkList := []string{"localhost:2183"}conn := getConnect(zkList)defer conn.Close()conn.Create("/go_servers", nil, 0, zk.WorldACL(zk.PermAll))time.Sleep(20 * time.Second)
}/*** 测试临时节点* @return {[type]}*/
func test2() {zkList := []string{"localhost:2183"}conn := getConnect(zkList)defer conn.Close()conn.Create("/testadaadsasdsaw", nil, zk.FlagEphemeral, zk.WorldACL(zk.PermAll))time.Sleep(20 * time.Second)
}/*** 获取所有节点*/
func test3() {zkList := []string{"localhost:2183"}conn := getConnect(zkList)defer conn.Close()children, _, err := conn.Children("/go_servers")if err != nil {fmt.Println(err)}fmt.Printf("%v \n", children)
}func main() {test1()
}

分布式 server

目前分布式系统已经很流行了，一些开源框架也被广泛应用，如 dubbo、Motan 等。对于一个分布式服务，最基本的一项功能就是服务的注册和发现，而利用 zk 的 EPHEMERAL 节点则可以很方便的实现该功能。EPHEMERAL节点正如其名，是临时性的，其生命周期是和客户端会话绑定的，当会话连接断开时，节点也会被删除。下边我们就来实现一个简单的分布式server。

server：

服务启动时，创建zk连接，并在go_servers节点下创建一个新节点，节点名为"ip:port"，完成服务注册
服务结束时，由于连接断开，创建的节点会被删除，这样client就不会连到该节点

client：

先从zk获取go_servers节点下所有子节点，这样就拿到了所有注册的server
从server列表中选中一个节点（这里只是随机选取，实际服务一般会提供多种策略），创建连接进行通信

为了演示，每次client连接server，获取server发送的时间后就断开。主要代码如下：

server.go

func main() {go starServer("127.0.0.1:8897")go starServer("127.0.0.1:8898")go starServer("127.0.0.1:8899")a := make(chan bool, 1)<-a
}func checkError(err error) {if err != nil {fmt.Println(err)}
}func starServer(port string) {tcpAddr, err := net.ResolveTCPAddr("tcp4", port)fmt.Println(tcpAddr)checkError(err)listener, err := net.ListenTCP("tcp", tcpAddr)checkError(err)//注册zk节点q// 链接zkconn, err := GetConnect()if err != nil {fmt.Printf(" connect zk error: %s ", err)}defer conn.Close()// zk节点注册err = RegistServer(conn, port)if err != nil {fmt.Printf(" regist node error: %s ", err)}for {conn, err := listener.Accept()if err != nil {fmt.Fprintf(os.Stderr, "Error: %s", err)continue}go handleCient(conn, port)}fmt.Println("aaaaaa")
}func handleCient(conn net.Conn, port string) {defer conn.Close()daytime := time.Now().String()conn.Write([]byte(port + ": " + daytime))
}
func GetConnect() (conn *zk.Conn, err error) {zkList := []string{"localhost:2181"}conn, _, err = zk.Connect(zkList, 10*time.Second)if err != nil {fmt.Println(err)}return
}func RegistServer(conn *zk.Conn, host string) (err error) {_, err = conn.Create("/go_servers/"+host, nil, zk.FlagEphemeral, zk.WorldACL(zk.PermAll))return
}func GetServerList(conn *zk.Conn) (list []string, err error) {list, _, err = conn.Children("/go_servers")return
}

client.go

func checkError(err error) {if err != nil {fmt.Println(err)}
}
func main() {for i := 0; i < 100; i++ {startClient()time.Sleep(1 * time.Second)}
}func startClient() {// service := "127.0.0.1:8899"//获取地址serverHost, err := getServerHost()if err != nil {fmt.Printf("get server host fail: %s \n", err)return}fmt.Println("connect host: " + serverHost)tcpAddr, err := net.ResolveTCPAddr("tcp4", serverHost)checkError(err)conn, err := net.DialTCP("tcp", nil, tcpAddr)checkError(err)defer conn.Close()_, err = conn.Write([]byte("timestamp"))checkError(err)result, err := ioutil.ReadAll(conn)checkError(err)fmt.Println(string(result))return
}func getServerHost() (host string, err error) {conn, err := GetConnect()if err != nil {fmt.Printf(" connect zk error: %s \n ", err)return}defer conn.Close()serverList, err := GetServerList(conn)if err != nil {fmt.Printf(" get server list error: %s \n", err)return}count := len(serverList)if count == 0 {err = errors.New("server list is empty \n")return}//随机选中一个返回r := rand.New(rand.NewSource(time.Now().UnixNano()))host = serverList[r.Intn(3)]return
}
func GetConnect() (conn *zk.Conn, err error) {zkList := []string{"localhost:2181"}conn, _, err = zk.Connect(zkList, 10*time.Second)if err != nil {fmt.Println(err)}return
}
func GetServerList(conn *zk.Conn) (list []string, err error) {list, _, err = conn.Children("/go_servers")return

先启动 server.go，再启动 client.go。

参考

https://www.kancloud.cn/liupengjie/go/745743