Golang 操作zookeeper

使用人比較多的zookeeper go client有

• gozk
  • 文檔地址: https://wiki.ubuntu.com/gozk
• go-zookeeper
  • 文檔地址: http://godoc.org/github.com/samuel/go-zookeeper/zk

后者文檔比較全一點，本文中使用go-zookeeper 編寫使用代碼

連接zk server

package main

import (
	"fmt"
	"github.com/samuel/go-zookeeper/zk"
	"time"
)

func main() {
	// 創建zk連接地址
	hosts := []string{"127.0.0.1:2181"}
	// 連接zk
	conn, _, err := zk.Connect(hosts, time.Second*5)
	defer conn.Close()
	if err != nil {
		fmt.Println(err)
		return
	}
	println(conn.Server())
}

增刪改查

package main

import (
	"fmt"
	"github.com/samuel/go-zookeeper/zk"
	"time"
)

var (
	path = "/test"
)

// 增
func add(conn *zk.Conn) {
	var data = []byte("test value")
	// flags有4種取值：
	// 0:永久，除非手動刪除
	// zk.FlagEphemeral = 1:短暫，session斷開則該節點也被刪除
	// zk.FlagSequence  = 2:會自動在節點后面添加序號
	// 3:Ephemeral和Sequence，即，短暫且自動添加序號
	var flags int32 = 0
	// 獲取訪問控制權限
	acls := zk.WorldACL(zk.PermAll)
	s, err := conn.Create(path, data, flags, acls)
	if err != nil {
		fmt.Printf("創建失敗: %v\n", err)
		return
	}
	fmt.Printf("創建: %s 成功", s)
}

// 查
func get(conn *zk.Conn) {
	data, _, err := conn.Get(path)
	if err != nil {
		fmt.Printf("查詢%s失敗, err: %v\n", path, err)
		return
	}
	fmt.Printf("%s 的值為 %s\n", path, string(data))
}

// 刪改與增不同在於其函數中的version參數,其中version是用於 CAS支持
// 可以通過此種方式保證原子性
// 改
func modify(conn *zk.Conn) {
	new_data := []byte("hello zookeeper")
	_, sate, _ := conn.Get(path)
	_, err := conn.Set(path, new_data, sate.Version)
	if err != nil {
		fmt.Printf("數據修改失敗: %v\n", err)
		return
	}
	fmt.Println("數據修改成功")
}

// 刪
func del(conn *zk.Conn) {
	_, sate, _ := conn.Get(path)
	err := conn.Delete(path, sate.Version)
	if err != nil {
		fmt.Printf("數據刪除失敗: %v\n", err)
		return
	}
	fmt.Println("數據刪除成功")
}

func main() {
	// 創建zk連接地址
	hosts := []string{"127.0.0.1:2181"}
	// 連接zk
	conn, _, err := zk.Connect(hosts, time.Second*5)
	defer conn.Close()
	if err != nil {
		fmt.Println(err)
		return
	}

	/* 增刪改查 */
	//add(conn)
	//get(conn)
	//modify(conn)
	del(conn)
	get(conn)
}

watch機制

Java API中是通過Watcher實現的，在go-zookeeper中則是通過Event。道理都是一樣的
全局監聽：

1. 調用zk.WithEventCallback(callback)設置回調

package main

import (
	"fmt"
	"github.com/samuel/go-zookeeper/zk"
	"time"
)

var (
	hosts       = []string{"127.0.0.1:2181"}
	path        = "/wtzk"
	flags int32 = zk.FlagEphemeral
	data        = []byte("zk data 001")
	acls        = zk.WorldACL(zk.PermAll)
)

func main() {
	// 創建監聽的option，用於初始化zk
	eventCallbackOption := zk.WithEventCallback(callback)
	// 連接zk
	conn, _, err := zk.Connect(hosts, time.Second*5, eventCallbackOption)
	defer conn.Close()
	if err != nil {
		fmt.Println(err)
		return
	}

	// 開始監聽path
	_, _, _, err = conn.ExistsW(path)
	if err != nil {
		fmt.Println(err)
		return
	}

	// 觸發創建數據操作
	create(conn, path, data)

	//再次監聽path
	_, _, _, err = conn.ExistsW(path)
	if err != nil {
		fmt.Println(err)
		return
	}
	// 觸發刪除數據操作
	del(conn, path)

}

// zk watch 回調函數
func callback(event zk.Event) {
	// zk.EventNodeCreated
	// zk.EventNodeDeleted
	fmt.Println("###########################")
	fmt.Println("path: ", event.Path)
	fmt.Println("type: ", event.Type.String())
	fmt.Println("state: ", event.State.String())
	fmt.Println("---------------------------")
}

// 創建數據
func create(conn *zk.Conn, path string, data []byte) {
	_, err := conn.Create(path, data, flags, acls)
	if err != nil {
		fmt.Printf("創建數據失敗: %v\n", err)
		return
	}
	fmt.Println("創建數據成功")
}

// 刪除數據
func del(conn *zk.Conn, path string) {
	_, stat, _ := conn.Get(path)
	err := conn.Delete(path, stat.Version)
	if err != nil {
		fmt.Printf("刪除數據失敗: %v\n", err)
		return
	}
	fmt.Println("刪除數據成功")
}

2. 部分監聽：
   1.調用conn.ExistsW(path) 或GetW(path)為對應節點設置監聽，該監聽只生效一次
   2.開啟一個協程處理chanel中傳來的event事件
   （注意：watchCreataNode一定要放在一個協程中，不能直接在main中調用，不然會阻塞main）

package main

import (
	"fmt"
	"github.com/samuel/go-zookeeper/zk"
	"time"
)

var (
	hosts       = []string{"127.0.0.1:2181"}
	path        = "/wtzk"
	flags int32 = zk.FlagEphemeral
	data        = []byte("zk data 001")
	acls        = zk.WorldACL(zk.PermAll)
)

func main() {
	// 連接zk
	conn, _, err := zk.Connect(hosts, time.Second*5)
	defer conn.Close()
	if err != nil {
		fmt.Println(err)
		return
	}

	// 開始監聽path
	_, _, event, err := conn.ExistsW(path)
	if err != nil {
		fmt.Println(err)
		return
	}

	// 協程調用監聽事件
	go watchZkEvent(event)

	// 觸發創建數據操作
	create(conn, path, data)

}

// zk 回調函數
func watchZkEvent(e <-chan zk.Event) {
	event := <-e
	fmt.Println("###########################")
	fmt.Println("path: ", event.Path)
	fmt.Println("type: ", event.Type.String())
	fmt.Println("state: ", event.State.String())
	fmt.Println("---------------------------")
}

// 創建數據
func create(conn *zk.Conn, path string, data []byte) {
	_, err := conn.Create(path, data, flags, acls)
	if err != nil {
		fmt.Printf("創建數據失敗: %v\n", err)
		return
	}
	fmt.Println("創建數據成功")
}

1.如果即設置了全局監聽又設置了部分監聽，那么最終是都會觸發的，並且全局監聽在先執行
2.如果設置了監聽子節點，那么事件的觸發是先子節點后父節點

客戶端隨機hostname支持

最終就是Round Robin策略

var hosts = []string{"host01:2181", "host02:2181", "host03:2181",}
hostPro := new(zk.DNSHostProvider)
//先初始化
err := hostPro.Init(hosts)

if err != nil {
	fmt.Println(err)
	return
}
//獲得host
server, retryStart := hostPro.Next()
fmt.Println(server, retryStart)
//連接成功后會調用
hostPro.Connected()