Golang基于Zookeeper实现节点注册发现、内容配置变更

╰+攻爆jí腚メ 2023-01-18 12:27 25阅读 0赞

# 一、Zookeeper入门 #

## 1.1 Zookeeper简介 ##

Zookeeper是一个分布式数据库(程序协调服务)，Hadoop子项目；以树状方式维护节点方数据的增、删、改、查；通过监听可以获取相应消息事件；  
节点类型：

*  持久节点：一直存储在服务器上(0)
 *  临时节点：会话失效、节点自动清理(FlagEphemeral)
 *  顺序节点：节点创建时自动分配序列号(FlagSequence)

## 1.2 Zookeeper安装启动 ##

[root@localhost ~]$ wget https://mirrors.bfsu.edu.cn/apache/zookeeper/zookeeper-3.7.0/apache-zookeeper-3.7.0-bin.tar.gz
    [root@localhost ~]$ tar -xf apache-zookeeper-3.7.0-bin.tar.gz
    [root@localhost ~]$ echo 'tickTime=2000 dataDir=/var/lib/zookeeper clientPort=2181'> apache-zookeeper-3.7.0-bin/conf/zoo.cfg
    [root@localhost ~]$ apache-zookeeper-3.7.0-bin/bin/zkServer.sh start
    [root@localhost ~]$ netstat -anultp | grep 2181
    tcp6       0      0 :::2181                 :::*                    LISTEN      3484/java

# 二、Golang实现Zookeeper核心功能 #

## 2.1 增、查、删、改 ##

package main
    import "github.com/samuel/go-zookeeper/zk"
    
    func main() { 
    	conn, _, err := zk.Connect([]string{ "127.0.0.1:2181"}, 5*time.Second)
    	if err != nil { 
    		panic(err)
    	}
    
    	// 1.验证根是否存在
    	if hasRoot, _, _ := conn.Exists("/root1"); !hasRoot { 
    		// 2.新增根
    		_, err = conn.Create("/root1", []byte("root_content"), 0, zk.WorldACL(zk.PermAll))
    		if err != nil { 
    			fmt.Println("failed add root, info: ", err.Error())
    		}
    		fmt.Println("add /root1 success")
    	}
    
    	// 3.查询根
    	data, stat, err := conn.Get("/root1")
    	if err != nil { 
    		fmt.Println("failed get root, info: ", err.Error())
    	}
    	fmt.Println("text: ", string(data), stat.Version)
    
    	// 4.修改根
    	if _, err = conn.Set("/root1", []byte("update text"), stat.Version); err != nil { 
    		fmt.Println("failed update root")
    	}
    
    	// 5.设置子节点(必须要有根/父节点)
    	if _, err = conn.Create("/root1/subnode", []byte("node_text"), 0, zk.WorldACL(zk.PermAll)); err != nil { 
    		fmt.Println("failed add subnode, info: ", err.Error())
    	}
    	// 6.获取子节点列表
    	childNodes, _, err := conn.Children("/root1")
    	if err != nil { 
    		fmt.Println("failed get node list, info: ", err.Error())
    	} else { 
    		fmt.Println("node list: ", childNodes)
    	}
    
    	// 6.删除根(必须先查后删, 删完子才能删父节点)
    	_, stat, _ = conn.Get("/root1/subnode")
    	if err := conn.Delete("/root1/subnode", stat.Version); err != nil { 
    		fmt.Println("falied delete node, info: ", stat.Version, err.Error())
    	}
    	_, stat, _ = conn.Get("/root1")
    	if err := conn.Delete("/root1", stat.Version); err != nil { 
    		fmt.Println("falied delete root, info: ", stat.Version, err.Error())
    	}
    }

## 2.2 监听(发现) ##

### 2.2.1 节点变化 ###

func WatchHostsByPath(path string) (chan []string, chan error) { 
    	snapshots := make(chan []string) // 变动后的挂载目标列表
    	errors := make(chan error)       // 变动错误信息
    	go func() { 
    		for { 
    			snapshot, _, events, err := conn.ChildrenW(path)
    			if err != nil { 
    				errors <- err
    			}
    			snapshots <- snapshot
    			// 阻塞直到出现事件消息(可以不用select)
    			select { 
    			case evt := <-events:
    				if evt.Err != nil { 
    					errors <- evt.Err
    				}
    				fmt.Printf("ChildrenW Event Path:%v, Type:%v\n", evt.Path, evt.Type)
    			}
    		}
    	}()
    	return snapshots, errors
    }

### 2.2.2 节点内容变化 ###

func WatchDataByPath(nodePath string) (chan []byte, chan error) { 
    	conn := z.conn
    	snapshots := make(chan []byte)
    	errors := make(chan error)
    	go func() { 
    		for { 
    			data, _, events, err := conn.GetW(nodePath)
    			if err != nil { 
    				errors <- err
    			}
    			snapshots <- data
    			select { 
    			case evt := <-events:
    				if evt.Err != nil { 
    					errors <- evt.Err
    					return
    				}
    				fmt.Printf("GetW Event Path:%v, Type:%v\n", evt.Path, evt.Type)
    			}
    		}
    	}()
    	return snapshots, errors
    }

## 2.3 注册 ##

### 2.3.1 节点注册 ###

// RegistHostOnPath 将主机挂载在路径上(节点上)
    func RegistHostOnPath(nodePath, host string) (err error) { 
    	// 1. 若路径不存在则新建
    	ex, _, err := z.conn.Exists(nodePath)
    	if err != nil { 
    		return
    	}
    	if !ex { 
    		_, err = z.conn.Create(nodePath, nil, 0, zk.WorldACL(zk.PermAll))
    		if err != nil { 
    			return
    		}
    	}
    	// 2. 将主机进行挂载(路径为永久、主机为临时); 临时会检测主机的存活并清理
    	subNodePath := fmt.Sprintf("%s/%s", nodePath, host)
    	// 2.1 主机是否已挂载, 没挂则挂
    	if ex, _, _ := z.conn.Exists(subNodePath); !ex { 
    		_, err = z.conn.Create(subNodePath, nil, zk.FlagEphemeral, zk.WorldACL(zk.PermAll))
    	}
    	return
    }

### 2.3.2 节点内容修改 ###

// UpdatePathData 有则更新、无则新建节点并配置内容
    func UpdatePathData(nodePath string, config []byte, version int32) (err error) { 
    	ex, _, _ := z.conn.Exists(nodePath)
    	if !ex { 
    		z.conn.Create(nodePath, config, 0, zk.WorldACL(zk.PermAll))
    		return nil
    	}
    	// 需要版本才能更新
    	_, stat, err := z.GetPathData(nodePath)
    	if err != nil { 
    		return
    	}
    	_, err = z.conn.Set(nodePath, config, stat.Version)
    	return
    }

# 三、整合测试 #

## 3.1 封装以上核心方法 ##

package zookeeper
    import "github.com/samuel/go-zookeeper/zk"
    
    type ZKDriver struct { 
    	hosts      []string // zookeeper节点列表:2181
    	conn       *zk.Conn // 操作句柄
    	pathPrefix string   // 前缀标识
    }
    func NewZKDriver(hosts []string) (*ZKDriver, error) { 
    	conn, _, err := zk.Connect(hosts, 5*time.Second)
    	if err != nil { 
    		return nil, err
    	}
    	return &ZKDriver{ hosts: hosts, conn: conn, pathPrefix: "/zk_service_"}, nil
    }
    // Close 关闭服务
    func (z *ZKDriver) Close() { 
    	z.conn.Close()
    }
    // GetPathData 获取配置
    func (z *ZKDriver) GetPathData(nodePath string) ([]byte, *zk.Stat, error) { 
    	return z.conn.Get(nodePath)
    }
    // UpdatePathData 有则更新、无则新建配置
    func (z *ZKDriver) UpdatePathData(nodePath string, config []byte, version int32) (err error) { 
    	// 代码不变
    }
    // RegistHostOnPath 将主机挂载在路径上(节点上)
    func (z *ZKDriver) RegistHostOnPath(nodePath, host string) (err error) { 
    	// 代码不变
    }
    // GetListByPath 获取节点下挂载的列表
    func (z *ZKDriver) GetListByPath(nodePath string) (list []string, err error) { 
    	// 代码不变
    }
    // WatchHostsByPath 监听节点下挂载目标的变动(离线|上线)
    func (z *ZKDriver) WatchHostsByPath(path string) (chan []string, chan error) { 
    	conn := z.conn
    	// 代码不变
    }
    // WatchDataByPath 监控主机的值变化
    func (z *ZKDriver) WatchDataByPath(nodePath string) (chan []byte, chan error) { 
    	conn := z.conn
    	// 代码不变
    }

## 3.2 测试 ##

### 3.2.1 节点发现测试 ###

注册端

package main
    import "demo/zookeeper"
    func main() { 
    	driver, err := zookeeper.NewZKDriver([]string{ "127.0.0.1:2181"})
    	if err != nil { 
    		log.Fatal(err)
    	}
    	defer driver.Close()
    	// 60秒10次变动
    	for i := 0; i < 10; i++ { 
    		if err = driver.RegistHostOnPath("/server", fmt.Sprintf("192.168.1.1%d", i)); err != nil { 
    			fmt.Println("failed regist host at path")
    		} else { 
    			fmt.Println("regist host successfully")
    		}
    		time.Sleep(10 * time.Second)
    	}
    	fmt.Println("节点变动已完成")
    	quit := make(chan os.Signal)
    	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    	<-quit
    }

监听端

package main
    import "demo/zookeeper"
    func main() { 
    	driver, err := zookeeper.NewZKDriver([]string{ "127.0.0.1:2181"})
    	if err != nil { 
    		log.Fatal(err)
    	}
    	defer driver.Close()
    
    	// 获取节点列表, 如果节点不存在则新建
    	nodeList, err := driver.GetListByPath("/server")
    	if err != nil && strings.Contains(err.Error(), "not exist") { 
    		log.Fatal("首次启动请先启动注册端先把路径给创建了(持久节点)")
    	} else if err != nil { 
    		log.Fatal(err)
    	}
    	fmt.Println("server node:", nodeList)
    
    	// 监听节点变化
    	chanNodes, chanNodesErr := driver.WatchHostsByPath("/server")
    	go func() { 
    		for { 
    			select { 
    			case err := <-chanNodesErr:
    				log.Println("chanNodes occur error, info: ", err.Error())
    			case changeNodes := <-chanNodes:
    				log.Println("node list changed, new list: ", changeNodes)
    			default:
    				time.Sleep(time.Millisecond * 200)
    			}
    		}
    	}()
    	quit := make(chan os.Signal)
    	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    	<-quit
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4OTAwNTY1_size_16_color_FFFFFF_t_70]

### 3.2.2 内容变更测试 ###

变更端

package main
    import "demo/zookeeper"
    func main() { 
    	driver, err := zookeeper.NewZKDriver([]string{ "127.0.0.1:2181"})
    	if err != nil { 
    		log.Fatal(err)
    	}
    	defer driver.Close()
    	// 60秒10次变动
    	for i := 0; i < 10; i++ { 
    		conf := fmt.Sprintf("server-id=" + fmt.Sprint(i))
    		if err := driver.UpdatePathData("/my.cnf", []byte(conf), int32(i)); err != nil { 
    			fmt.Println("update content occur err, info: ", err.Error())
    		} else { 
    			fmt.Println("update content successfully, data: ", conf)
    		}
    		time.Sleep(10 * time.Second)
    	}
    
    	fmt.Println("内容变动已完成")
    	quit := make(chan os.Signal)
    	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    	<-quit
    }

监听端

package main
    import "demo/zookeeper"
    func main() { 
    	driver, err := zookeeper.NewZKDriver([]string{ "127.0.0.1:2181"})
    	if err != nil { 
    		log.Fatal(err)
    	}
    	defer driver.Close()
    
    	//获取节点内容
    	zc, _, err := driver.GetPathData("/my.cnf")
    	if err != nil && strings.Contains(err.Error(), "not exist") { 
    		log.Fatal("首次运行请先启动内容变更程序端")
    	} else if err != nil { 
    		log.Fatal(err)
    	}
    	fmt.Println("get node data:")
    	fmt.Println(string(zc))
    
    	//动态监听节点内容
    	dataChan, dataErrChan := driver.WatchDataByPath("/my.cnf")
    	go func() { 
    		for { 
    			select { 
    			case changeErr := <-dataErrChan:
    				fmt.Println("content change occur err, info: ", changeErr)
    			case changedData := <-dataChan:
    				fmt.Println("WatchGetData changed, info: ", string(changedData))
    			}
    		}
    	}()
    	quit := make(chan os.Signal)
    	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    	<-quit
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4OTAwNTY1_size_16_color_FFFFFF_t_70_pic_center]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4OTAwNTY1_size_16_color_FFFFFF_t_70]: /images/20221021/716d94f6fde1437ab52b6eca54b7ead5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4OTAwNTY1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221021/9eb7ae5ece6249beb81c9d9840f5a34f.png