golang使用zookeeper進(jìn)行CURD

更新時(shí)間：2025年11月21日 09:49:27 作者：蔡蔡開始內(nèi)卷

本文介紹Zookeeper的基本概念及其在Golang中的實(shí)現(xiàn)方法,包括連接建立、節(jié)點(diǎn)創(chuàng)建與查詢、節(jié)點(diǎn)存在性的檢查、節(jié)點(diǎn)刪除及內(nèi)容修改等,具有一定的參考價(jià)值,感興趣的可以了解一下

一、Zookeeper入門

1.1. Zookeeper簡介

Zookeeper是一個(gè)分布式數(shù)據(jù)庫(程序協(xié)調(diào)服務(wù))，Hadoop子項(xiàng)目；以樹狀方式維護(hù)節(jié)點(diǎn)方數(shù)據(jù)的增、刪、改、查；通過監(jiān)聽可以獲取相應(yīng)消息事件；
節(jié)點(diǎn)類型：

持久節(jié)點(diǎn)：一直存儲在服務(wù)器上(0)
臨時(shí)節(jié)點(diǎn)：會(huì)話失效、節(jié)點(diǎn)自動(dòng)清理(FlagEphemeral)
順序節(jié)點(diǎn)：節(jié)點(diǎn)創(chuàng)建時(shí)自動(dòng)分配序列號(FlagSequence)

二.啟動(dòng)zookeeper

三.核心包

go-zookeeper

go get github.com/samuel/go-zookeeper/zk
文檔地址: http://godoc.org/github.com/samuel/go-zookeeper/zk

四.Golang實(shí)現(xiàn)Zookeeper核心功能

4.1 建立連接

func GetConnect(zkList []string) (conn *zk.Conn) {
	// 創(chuàng)建監(jiān)聽的option，用于初始化zk
	conn, _, err := zk.Connect(zkList, 10*time.Second)
	if err != nil {
		fmt.Println(err)
	}
	return
}

4.2創(chuàng)建節(jié)點(diǎn)

func Create(conn *zk.Conn, path string, data []byte, flags int32, acl int32) (val string, err error) {
	//flags有4種取值：
	//0:永久，除非手動(dòng)刪除
	//zk.FlagEphemeral = 1:短暫，session斷開則改節(jié)點(diǎn)也被刪除
	//zk.FlagSequence  = 2:會(huì)自動(dòng)在節(jié)點(diǎn)后面添加序號
	//3:Ephemeral和Sequence，即，短暫且自動(dòng)添加序號
	// 獲取訪問控制權(quán)限
	var acls []zk.ACL
	if acl == 0 {
		/**
		PermRead = 1 << iota   1
		PermWrite              2
		PermCreate             4
		PermDelete             8
		PermAdmin             16
		PermAll = 0x1f        31
		*/
		acls = zk.WorldACL(zk.PermAll)
	} else {
		acls = zk.WorldACL(acl)
	}

	val, err = conn.Create(path, data, flags, acls)
	return
}

var (
	zkList = []string{"127.0.0.1:2181"}
	path   = "/root"
)

func TestCreate(t *testing.T) {
	conn := GetConnect(zkList)

	defer conn.Close()
	//創(chuàng)建節(jié)點(diǎn)
	val, err := Create(conn, path, []byte("root value"), 0, 31)
	if err != nil {
		fmt.Printf("創(chuàng)建失敗: %v\n", err)
		return
	}
	fmt.Printf("創(chuàng)建: %s 成功", val)
}

4.3查詢節(jié)點(diǎn)

// 查
func Get(conn *zk.Conn, path string) (dataStr string, stat *zk.Stat, err error) {
	data, stat, err := conn.Get(path)
	if err != nil {
		return "", nil, err
	}
	return string(data), stat, err
}

func TestGet(t *testing.T) {
	conn := GetConnect(zkList)

	defer conn.Close()
	//查詢節(jié)點(diǎn)
	val, _, err := Get(conn, path)
	if err != nil {
		fmt.Printf("查詢%s失敗, err: %v\n", path, err)
		return
	}
	fmt.Printf("%s 的值為 %s\n", path, val)
}

4.4 節(jié)點(diǎn)是否存在

func Exists(conn *zk.Conn, path string) (exist bool, err error) {
	exist, _, err = conn.Exists(path)
	return
}

func TestExist(t *testing.T) {
	conn := GetConnect(zkList)

	defer conn.Close()
	//是否存在
	val, err := Exists(conn, path)
	if err != nil {
		fmt.Printf("查詢%s失敗, err: %v\n", path, err)
		return
	}
	if val {
		fmt.Printf("%s 存在\n", path)
	} else {
		fmt.Printf("%s 不存在\n", path)
	}
}

4.5刪除節(jié)點(diǎn)

//刪除 cas支持
func Del(conn *zk.Conn, path string) (err error) {
	_, sate, _ := Get(conn, path)
	fmt.Println(sate)
	err = conn.Delete(path, sate.Version)
	return err
}

func TestDel(t *testing.T) {
	conn := GetConnect(zkList)

	defer conn.Close()
	// 刪除
	err := Del(conn, path)
	if err != nil {
		fmt.Printf("數(shù)據(jù)刪除失敗: %v\n", err)
		return
	}
	fmt.Println("數(shù)據(jù)刪除成功")
}

4.6 修改節(jié)點(diǎn)內(nèi)容

// 改 CAS支持
// 可以通過此種方式保證原子性
func Modify(conn *zk.Conn, path string, newData []byte) (sate *zk.Stat, err error) {
	_, sate, _ = conn.Get(path)
	fmt.Println(sate)
	sate, err = conn.Set(path, newData, sate.Version)
	return
}


func TestModify(t *testing.T) {
	conn := GetConnect(zkList)

	defer conn.Close()
	newData := []byte("hello zookeeper")
	stat, err := Modify(conn, path, newData)
	if err != nil {
		fmt.Printf("數(shù)據(jù)修改失敗: %v\n", err)
		return
	}
	fmt.Printf("數(shù)據(jù)修改成功,stat %v\n", stat)
}

4.7獲取目錄信息

func Children(conn *zk.Conn, path string) (data []string, err error) {
	data, _, err = conn.Children(path)
	return
}

func TestChildren(t *testing.T) {
	conn := GetConnect(zkList)

	defer conn.Close()
	data, err := Children(conn, "/")
	if err != nil {
		fmt.Printf("獲取數(shù)據(jù)失敗: %v\n", err)
		return
	}
	fmt.Printf("獲取數(shù)據(jù)成功,data %v\n", data)
}

五.watch

func callback(event zk.Event) {
    fmt.Println(">>>>>>>>>>>>>>>>>>>")
    fmt.Println("path:", event.Path)
    fmt.Println("type:", event.Type.String())
    fmt.Println("state:", event.State.String())
    fmt.Println("<<<<<<<<<<<<<<<<<<<")
}
 
func ZKOperateWatchTest() {
    fmt.Printf("ZKOperateWatchTest\n")
 
    option := zk.WithEventCallback(callback)
    var hosts = []string{"localhost:2181"}
    conn, _, err := zk.Connect(hosts, time.Second*5, option)
    if err != nil {
        fmt.Println(err)
        return
    }
    defer conn.Close()
 
    var path1 = "/zk_test_go1"
    var data1 = []byte("zk_test_go1_data1")
    exist, s, _, err := conn.ExistsW(path1)
    if err != nil {
        fmt.Println(err)
        return
    }
 
    fmt.Printf("path[%s] exist[%t]\n", path1, exist)
    fmt.Printf("state:\n")
    fmt.Printf("%s\n", ZkStateStringFormat(s))
 
    // try create
    var acls = zk.WorldACL(zk.PermAll)
    p, err_create := conn.Create(path1, data1, zk.FlagEphemeral, acls)
    if err_create != nil {
        fmt.Println(err_create)
        return
    }
    fmt.Printf("created path[%s]\n", p)
 
    time.Sleep(time.Second * 2)
 
    exist, s, _, err = conn.ExistsW(path1)
    if err != nil {
        fmt.Println(err)
        return
    }
 
    fmt.Printf("path[%s] exist[%t] after create\n", path1, exist)
    fmt.Printf("state:\n")
    fmt.Printf("%s\n", ZkStateStringFormat(s))
 
    // delete
    conn.Delete(path1, s.Version)
 
    exist, s, _, err = conn.ExistsW(path1)
    if err != nil {
        fmt.Println(err)
        return
    }
    fmt.Printf("path[%s] exist[%t] after delete\n", path1, exist)
    fmt.Printf("state:\n")
    fmt.Printf("%s\n", ZkStateStringFormat(s))
}

不止會(huì)打印watch監(jiān)聽的節(jié)點(diǎn)信息，還有打印session會(huì)話的狀態(tài)

到此這篇關(guān)于golang使用zookeeper進(jìn)行CURD的文章就介紹到這了,更多相關(guān)golang zookeeper進(jìn)行CURD內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: