详解ZooKeeper在微服务注册中心的应用

雁过留声  金牌会员 | 2024-2-27 00:01:31 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 830|帖子 830|积分 2490

本文分享自华为云社区《SpringCloud ZooKeeper 详解，以及与Go、Rust等非Java服务的集成》，作者： 张俭。
ZooKeeper，是一个开源的分布式协调服务，不仅支持分布式选举、任务分配，还可以用于微服务的注册中心和配置中心。本文，我们将深入探讨ZooKeeper用做微服务注册中心的场景。
ZooKeeper中的服务注册路径

SpringCloud ZooKeeper遵循特定的路径结构进行服务注册
  1. /services/${spring.application.name}/${serviceId}
复制代码
示例：
  1. /services/provider-service/d87a3891-1173-45a0-bdfa-a1b60c71ef4e
复制代码
/services和/${spring.application.name}是ZooKeeper中的永久节点，/${serviceId}是临时节点，当服务下线时，ZooKeeper会自动删除该节点。
注：当微服务的最后一个实例下线时，SpringCloud ZooKeeper框架会删除/${spring.application.name}节点。
ZooKeeper中的服务注册数据

下面是一个典型的服务注册内容示例：
  1. {
  3. "name":"provider-service",
  5. "id":"d87a3891-1173-45a0-bdfa-a1b60c71ef4e",
  7. "address":"192.168.0.105",
  9. "port":8080,
  11. "sslPort":null,
  13. "payload":{
  15. "@class":"org.springframework.cloud.zookeeper.discovery.ZookeeperInstance",
  17. "id":"provider-service",
  19. "name":"provider-service",
  21. "metadata":{
  23. "instance_status":"UP"
  25. }
  27. },
  29. "registrationTimeUTC":1695401004882,
  31. "serviceType":"DYNAMIC",
  33. "uriSpec":{
  35. "parts":[
  37. {
  39. "value":"scheme",
  41. "variable":true
  43. },
  45. {
  47. "value":"://",
  49. "variable":false
  51. },
  53. {
  55. "value":"address",
  57. "variable":true
  59. },
  61. {
  63. "value":":",
  65. "variable":false
  67. },
  69. {
  71. "value":"port",
  73. "variable":true
  75. }
  77. ]
  79. }
  81. }
复制代码
其中，address、port和uriSpec是最核心的数据。uriSpec中的parts区分了哪些内容是可变的，哪些是固定的。
SpringCloud 服务使用OpenFeign互相调用

一旦两个微服务都注册到了ZooKeeper，那么它们就可以通过OpenFeign互相调用了。简单的示例如下
服务提供者

创建SpringBoot项目

创建SpringBoot项目，并添加spring-cloud-starter-zookeeper-discovery和spring-boot-starter-web依赖。
配置application.yaml
  1. spring:
  3. application:
  5. name: provider-service
  7. cloud:
  9. zookeeper:
  11. connect-string: localhost:2181
  13. server:
  15. port: 8082
复制代码
注册到ZooKeeper

在启动类上添加@EnableDiscoveryClient注解。
创建一个简单的REST接口
  1. @RestController
  3. public class ProviderController {
  5. @GetMapping("/hello")
  7. public String hello() {
  9. return "Hello from Provider Service!";
  11. }
  13. }
复制代码
服务消费者

创建SpringBoot项目

创建SpringBoot项目，并添加spring-cloud-starter-zookeeper-discovery、spring-cloud-starter-openfeign和spring-boot-starter-web依赖。
配置application.yaml
  1. spring:
  3. application:
  5. name: consumer-service
  7. cloud:
  9. zookeeper:
  11. connect-string: localhost:2181
  13. server:
  15. port: 8081
复制代码
注册到ZooKeeper

在启动类上添加@EnableDiscoveryClient注解。
创建一个REST接口，通过OpenFeign调用服务提供者
  1. @RestController
  3. public class ConsumerController {
  7. @Autowired
  9. private ProviderClient providerClient;
  11. @GetMapping("/getHello")
  13. public String getHello() {
  15. return providerClient.hello();
  17. }
  19. }
复制代码
运行效果
  1. curl localhost:8081/getHello -i
  3. HTTP/1.1 200
  5. Content-Type: text/plain;charset=UTF-8
  7. Content-Length: 28
  9. Date: Wed, 18 Oct 2023 02:40:57 GMT
  11. Hello from Provider Service!
复制代码
非Java服务在SpringCloud ZooKeeper中注册

可能有些读者乍一看觉得有点奇怪，为什么要在SpringCloud ZooKeeper中注册非Java服务呢？没有这个应用场景。
当然，这样的场景比较少，常见于大部分项目都是用SpringCloud开发，但有少部分项目因为种种原因，不得不使用其他语言开发，比如Go、Rust等。这时候，我们就需要在SpringCloud ZooKeeper中注册非Java服务了。
对于非JVM语言开发的服务，只需确保它们提供了Rest/HTTP接口并正确地注册到ZooKeeper，就可以被SpringCloud的Feign客户端所调用。
Go服务在SpringCloud ZooKeeper

example代码组织：
  1. ├── consumer
  3. │ └── consumer.go
  5. ├── go.mod
  7. ├── go.sum
  9. └── provider
  11. └── provider.go
复制代码
Go服务提供者在SpringCloud ZooKeeper

注：该代码的质量为demo级别，实际生产环境需要更加严谨的代码，如重连机制、超时机制、更优秀的服务ID生成算法等。
  1. package main
  3. import (
  5. "fmt"
  7. "log"
  9. "net/http"
  11. "time"
  13. "encoding/json"
  15. "github.com/gin-gonic/gin"
  17. "github.com/samuel/go-zookeeper/zk"
  19. )
  21. const (
  23. zkServers = "localhost:2181" // Zookeeper服务器地址
  25. )
  27. func main() {
  29. // 初始化gin框架
  31. r := gin.Default()
  33. // 添加一个简单的hello接口
  35. r.GET("/hello", func(c *gin.Context) {
  37. c.String(http.StatusOK, "Hello from Go service!")
  39. })
  41. // 注册服务到zookeeper
  43. registerToZookeeper()
  45. // 启动gin服务器
  47. r.Run(":8080")
  49. }
  51. func registerToZookeeper() {
  53. conn, _, err := zk.Connect([]string{zkServers}, time.Second*5)
  55. if err != nil {
  57. panic(err)
  59. }
  61. // 检查并创建父级路径
  63. ensurePathExists(conn, "/services")
  65. ensurePathExists(conn, "/services/provider-service")
  67. // 构建注册的数据
  69. data, _ := json.Marshal(map[string]interface{}{
  71. "name": "provider-service",
  73. "address": "127.0.0.1",
  75. "port": 8080,
  77. "sslPort": nil,
  79. "payload": map[string]interface{}{"@class": "org.springframework.cloud.zookeeper.discovery.ZookeeperInstance", "id": "provider-service", "name": "provider-service", "metadata": map[string]string{"instance_status": "UP"}},
  81. "serviceType": "DYNAMIC",
  83. "uriSpec": map[string]interface{}{
  85. "parts": []map[string]interface{}{
  87. {"value": "scheme", "variable": true},
  89. {"value": "://", "variable": false},
  91. {"value": "address", "variable": true},
  93. {"value": ":", "variable": false},
  95. {"value": "port", "variable": true},
  97. },
  99. },
  101. })
  103. // 在zookeeper中注册服务
  105. path := "/services/provider-service/" + generateServiceId()
  107. _, err = conn.Create(path, data, zk.FlagEphemeral, zk.WorldACL(zk.PermAll))
  109. if err != nil {
  111. log.Fatalf("register service error: %s", err)
  113. } else {
  115. log.Println(path)
  117. }
  119. }
  121. func ensurePathExists(conn *zk.Conn, path string) {
  123. exists, _, err := conn.Exists(path)
  125. if err != nil {
  127. log.Fatalf("check path error: %s", err)
  129. }
  131. if !exists {
  133. _, err := conn.Create(path, []byte{}, 0, zk.WorldACL(zk.PermAll))
  135. if err != nil {
  137. log.Fatalf("create path error: %s", err)
  139. }
  141. }
  143. }
  145. func generateServiceId() string {
  147. // 这里简化为使用当前时间生成ID，实际生产环境可能需要更复杂的算法
  149. return fmt.Sprintf("%d", time.Now().UnixNano())
  151. }
复制代码
调用效果
  1. curl localhost:8081/getHello -i
  3. HTTP/1.1 200
  5. Content-Type: text/plain;charset=UTF-8
  7. Content-Length: 28
  9. Date: Wed, 18 Oct 2023 02:43:52 GMT
  11. Hello from Go Service!
复制代码
Go服务消费者在SpringCloud ZooKeeper
  1. package main
  3. import (
  5. "encoding/json"
  7. "fmt"
  9. "io"
  11. "log"
  13. "net/http"
  15. "time"
  17. "github.com/samuel/go-zookeeper/zk"
  19. )
  21. const (
  23. zkServers = "localhost:2181" // Zookeeper服务器地址
  25. )
  27. var conn *zk.Conn
  29. func main() {
  31. // 初始化ZooKeeper连接
  33. initializeZookeeper()
  35. // 获取服务信息
  37. serviceInfo := getServiceInfo("/services/provider-service")
  39. fmt.Println("Fetched service info:", serviceInfo)
  41. port := int(serviceInfo["port"].(float64))
  43. resp, err := http.Get(fmt.Sprintf("http://%s:%d/hello", serviceInfo["address"], port))
  45. if err != nil {
  47. panic(err)
  49. }
  51. body, err := io.ReadAll(resp.Body)
  53. if err != nil {
  55. panic(err)
  57. }
  59. fmt.Println(string(body))
  61. }
  63. func initializeZookeeper() {
  65. var err error
  67. conn, _, err = zk.Connect([]string{zkServers}, time.Second*5)
  69. if err != nil {
  71. log.Fatalf("Failed to connect to ZooKeeper: %s", err)
  73. }
  75. }
  77. func getServiceInfo(path string) map[string]interface{} {
  79. children, _, err := conn.Children(path)
  81. if err != nil {
  83. log.Fatalf("Failed to get children of %s: %s", path, err)
  85. }
  87. if len(children) == 0 {
  89. log.Fatalf("No services found under %s", path)
  91. }
  93. // 这里只获取第一个服务节点的信息作为示例，实际上可以根据负载均衡策略选择一个服务节点
  95. data, _, err := conn.Get(fmt.Sprintf("%s/%s", path, children[0]))
  97. if err != nil {
  99. log.Fatalf("Failed to get data of %s: %s", children[0], err)
  101. }
  103. var serviceInfo map[string]interface{}
  105. if err := json.Unmarshal(data, &serviceInfo); err != nil {
  107. log.Fatalf("Failed to unmarshal data: %s", err)
  109. }
  111. return serviceInfo
  113. }
复制代码
Rust服务在SpringCloud ZooKeeper

example代码组织：
  1. ├── Cargo.lock
  3. ├── Cargo.toml
  5. └── src
  7. └── bin
  9. ├── consumer.rs
  11. └── provider.rs
复制代码
Rust服务提供者在SpringCloud ZooKeeper
  1. use std::collections::HashMap;
  3. use std::time::Duration;
  5. use serde_json::Value;
  7. use warp::Filter;
  9. use zookeeper::{Acl, CreateMode, WatchedEvent, Watcher, ZooKeeper};
  11. static ZK_SERVERS: &str = "localhost:2181";
  13. static mut ZK_CONN: Option<ZooKeeper> = None;
  15. struct LoggingWatcher;
  17. impl Watcher for LoggingWatcher {
  19. fn handle(&self, e: WatchedEvent) {
  21. println!("WatchedEvent: {:?}", e);
  23. }
  25. }
  27. #[tokio::main]
  29. async fn main() {
  31. let hello = warp::path!("hello").map(|| warp::reply::html("Hello from Rust service!"));
  33. register_to_zookeeper().await;
  35. warp::serve(hello).run(([127, 0, 0, 1], 8083)).await;
  37. }
  39. async fn register_to_zookeeper() {
  41. unsafe {
  43. ZK_CONN = Some(ZooKeeper::connect(ZK_SERVERS, Duration::from_secs(5), LoggingWatcher).unwrap());
  45. let zk = ZK_CONN.as_ref().unwrap();
  47. let path = "/services/provider-service";
  49. if zk.exists(path, false).unwrap().is_none() {
  51. zk.create(path, vec![], Acl::open_unsafe().clone(), CreateMode::Persistent).unwrap();
  53. }
  55. let service_data = get_service_data();
  57. let service_path = format!("{}/{}", path, generate_service_id());
  59. zk.create(&service_path, service_data, Acl::open_unsafe().clone(), CreateMode::Ephemeral).unwrap();
  61. }
  63. }
  65. fn get_service_data() -> Vec<u8> {
  67. let mut data: HashMap<&str, Value> = HashMap::new();
  69. data.insert("name", serde_json::Value::String("provider-service".to_string()));
  71. data.insert("address", serde_json::Value::String("127.0.0.1".to_string()));
  73. data.insert("port", serde_json::Value::Number(8083.into()));
  75. serde_json::to_vec(&data).unwrap()
  77. }
  79. fn generate_service_id() -> String {
  81. format!("{}", chrono::Utc::now().timestamp_nanos())
  83. }
复制代码
Rust服务消费者在SpringCloud ZooKeeper
  1. use std::collections::HashMap;
  3. use std::time::Duration;
  5. use zookeeper::{WatchedEvent, Watcher, ZooKeeper};
  7. use reqwest;
  9. use serde_json::Value;
  11. static ZK_SERVERS: &str = "localhost:2181";
  13. struct LoggingWatcher;
  15. impl Watcher for LoggingWatcher {
  17. fn handle(&self, e: WatchedEvent) {
  19. println!("WatchedEvent: {:?}", e);
  21. }
  23. }
  25. #[tokio::main]
  27. async fn main() {
  29. let provider_data = fetch_provider_data_from_zookeeper().await;
  31. let response = request_provider(&provider_data).await;
  33. println!("Response from provider: {}", response);
  35. }
  37. async fn fetch_provider_data_from_zookeeper() -> HashMap<String, Value> {
  39. let zk = ZooKeeper::connect(ZK_SERVERS, Duration::from_secs(5), LoggingWatcher).unwrap();
  41. let children = zk.get_children("/services/provider-service", false).unwrap();
  43. if children.is_empty() {
  45. panic!("No provider services found!");
  47. }
  49. // For simplicity, we just take the first child (i.e., service instance).
  51. // In a real-world scenario, load balancing strategies would determine which service instance to use.
  53. let data = zk.get_data(&format!("/services/provider-service/{}", children[0]), false).unwrap();
  55. serde_json::from_slice(&data.0).unwrap()
  57. }
  59. async fn request_provider(provider_data: &HashMap<String, Value>) -> String {
  61. let address = provider_data.get("address").unwrap().as_str().unwrap();
  63. let port = provider_data.get("port").unwrap().as_i64().unwrap();
  65. let url = format!("http://{}:{}/hello", address, port);
  67. let response = reqwest::get(&url).await.unwrap();
  69. response.text().await.unwrap()
  71. }
复制代码
 
点击关注，第一时间了解华为云新鲜技术~
 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

雁过留声

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表