经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Zookeeper » 查看文章
Rpc-实现Zookeeper注册中心
来源：cnblogs　　作者：zko0　　时间：2023/2/17 16:17:50　　对本文有异议

1.前言

本文章是笔主在声哥的手写RPC框架的学习下，对注册中心的一个拓展。因为声哥某些部分没有保留拓展性，所以本文章的项目与声哥的工程有部分区别，核心内容在Curator的注册发现与注销，思想看准即可。

本文章Git仓库：zko0/zko0-rpc

声哥的RPC项目写的确实很详细，跟学一遍受益匪浅：

何人听我楚狂声的博客

在声哥的项目里使用Nacos作为了服务注册中心。本人拓展添加了ZooKeeper实现服务注册。

Nacos的服务注册和发现，设计的不是非常好，每次服务的发现都需要去注册中心拉取。本人实现ZooKeeper注册中心时，参考了Dubbo的设计原理，结合本人自身想法，添加了本地缓存：

  • Client发现服务后缓存在本地，维护一个服务——实例列表
  • 当监听到注册中心的服务列表发生了变化，Client更新本地列表
  • 当注册中心宕机，Client能够依靠本地的服务列表继续提供服务

问题：

  1. 实现服务注册的本地缓存，还需要实现注册中心的监听，当注册中心的服务发生更改时能够实现动态更新。或者用轮训的方式，定时更新，不过这种方式的服务实时性较差
  2. 当Server宕机，非临时节点注册容易出现服务残留无法清除的问题。所以我建议全部使用临时节点去注册。

2.内容

zookeeper需要简单学一下，知识内容非常简单，搭建也很简单，在此跳过。

如果你感兴趣，可以参考我的ZooKeeper的文章：Zookeeper学习笔记 - zko0

①添加依赖

Curator：(简化ZooKeeper客户端使用)(Netfix研发,捐给Apache,是Apache顶级项目)

这里排除slf4j依赖，因为笔主使用的slf4j存在冲突

  1. <!-- https://mvnrepository.com/artifact/org.apache.curator/curator-recipes -->
  2. <dependency>
  3.     <groupId>org.apache.curator</groupId>
  4.     <artifactId>curator-recipes</artifactId>
  5.     <version>5.2.0</version>
  6.     <exclusions>
  7.         <exclusion>
  8.             <groupId>org.slf4j</groupId>
  9.             <artifactId>slf4j-api</artifactId>
  10.         </exclusion>
  11.     </exclusions>
  12. </dependency>

②代码编写

1.首先创建一个连接类：

  1. @Slf4j
  2. public class ZookeeperUtil {
  4.     //内部化构造方法
  5.     private ZookeeperUtil(){
  6.     }
  8.     private static final String SERVER_HOSTNAME= RegisterCenterConfig.getHostName();
  10.     private static final Integer SERVER_PORT=RegisterCenterConfig.getServerPort();
  12.     private static CuratorFramework zookeeperClient;
  14.     public static CuratorFramework getZookeeperClient(){
  15.         if (zookeeperClient==null){
  16.             synchronized (ZookeeperUtil.class){
  17.                 if (zookeeperClient==null){
  18.                     RetryPolicy retryPolic=new ExponentialBackoffRetry(3000,10);
  19.                     zookeeperClient = CuratorFrameworkFactory.builder()
  20.                             .connectString(SERVER_HOSTNAME+":"+SERVER_PORT)
  21.                             .retryPolicy(retryPolic)
  22.                             //  zookeeper根目录为/serviceRegister，不为/
  23.                             .namespace("serviceRegister")
  24.                             .build();
  25.                     zookeeperClient.start();
  26.                 }
  27.             }
  28.         }
  29.         return zookeeperClient;
  30.     }
  32.     public static String getServerHostname(){
  33.         return SERVER_HOSTNAME;
  34.     }
  36.     public static Integer getServerPort(){
  37.         return SERVER_PORT;
  38.     }
  40. }

其中HOST，PORT信息我保存在regiserCenter.properties配置文件夹中，使用类读取：

  1. public class RpcConfig {
  4.     //注册中心类型
  5.     private static String registerCenterType;
  7.     //序列化类型
  8.     private static String serializerType;
  10.     //负载均衡类型
  11.     private static String loadBalanceType;
  13.     //配置Nacos地址
  14.     private static String registerCenterHost;
  16.     private static Integer registerCenterPort;
  19.     private static boolean zookeeperDestoryIsEphemeral;
  21.     private static String serverHostName;
  23.     private static Integer serverPort;
  25.     static {
  26.         ResourceBundle bundle = ResourceBundle.getBundle("rpc");
  27.         registerCenterType=bundle.getString("registerCenter.type");
  28.         loadBalanceType=bundle.getString("loadBalance.type");
  29.         registerCenterHost=bundle.getString("registerCenter.host");
  30.         registerCenterPort = Integer.parseInt(bundle.getString("registerCenter.port"));
  31.         try {
  32.             zookeeperDestoryIsEphemeral="true".equals(bundle.getString("registerCenter.destory.isEphemeral"));
  33.         } catch (Exception e) {
  34.             zookeeperDestoryIsEphemeral=false;
  35.         }
  36.         serializerType=bundle.getString("serializer.type");
  37.         serverHostName=bundle.getString("server.hostName");
  38.         serverPort=Integer.parseInt(bundle.getString("server.port"));
  39.     }
  41.     public static String getRegisterCenterType() {
  42.         return registerCenterType;
  43.     }
  45.     public static String getSerializerType() {
  46.         return serializerType;
  47.     }
  49.     public static String getLoadBalanceType() {
  50.         return loadBalanceType;
  51.     }
  53.     public static String getRegisterCenterHost() {
  54.         return registerCenterHost;
  55.     }
  57.     public static Integer getRegisterCenterPort() {
  58.         return registerCenterPort;
  59.     }
  62.     public static String getServerHostName() {
  63.         return serverHostName;
  64.     }
  66.     public static Integer getServerPort() {
  67.         return serverPort;
  68.     }
  70.     public static boolean isZookeeperDestoryIsEphemeral() {
  71.         return zookeeperDestoryIsEphemeral;
  72.     }
  74. }

下面的代码我和声哥有些不同，我将服务注册，注销方法放在ServerUtils中，服务发现方法放在ClientUtils中：

服务的高一致性存在两种做法：

  • 因为ZooKeeper存在临时节点，注册中心可以实现Client（RPC的Server）断开，注册服务信息的自动丢失
  • 不设置为临时节点，手动的服务注册清除

我这里两种都实现了，虽然做两种方式不同但是功能相同的代码放在一起看起来很奇怪，这里只是做演示。选择其中一种即可。（我建议使用临时节点，当Server宕机，残留的服务信息也能及时清除）

注册实现原理图：

626723792516afae5530e634e691794

接口：

  1. public interface ServiceDiscovery {
  2.     InetSocketAddress searchService(String serviceName);
  4.     void cleanLoaclCache(String serviceName);
  5. }
  1. public interface ServiceRegistry {
  2.     //服务注册
  3.     void register(String serviceName, InetSocketAddress inetAddress);
  5.     void cleanRegistry();
  6. }

ZooKeeper接口实现：

  1. public class ZookeeperServiceDiscovery implements ServiceDiscovery{
  3.     private final LoadBalancer loadBalancer;
  5.     public ZookeeperServiceDiscovery(LoadBalancer loadBalancer) {
  6.         this.loadBalancer = loadBalancer;
  7.     }
  8.     
  9.     @Override
  10.     public InetSocketAddress searchService(String serviceName) {
  11.         return ZookeeperClientUtils.searchService(serviceName,loadBalancer);
  12.     }
  14.     @Override
  15.     public void cleanLoaclCache(String serviceName) {
  16.         ZookeeperClientUtils.cleanLocalCache(serviceName);
  17.     }
  18. }
  1. public class ZookeeperServiceRegistry implements ServiceRegistry{
  2.     @Override
  3.     public void register(String serviceName, InetSocketAddress inetAddress) {
  4.         ZookeeperServerUitls.register(serviceName,inetAddress);
  5.     }
  7.     @Override
  8.     public void cleanRegistry() {
  9.         ZookeeperServerUitls.cleanRegistry();
  10.     }
  11. }

Factory工厂：

  1. public class ServiceFactory {
  3.     private static String center = RpcConfig.getRegisterCenterType();
  4.     private static String lb= RpcConfig.getLoadBalanceType();
  6.     private static  ServiceRegistry registry;
  8.     private static  ServiceDiscovery discovery;
  10.     private static Object registerLock=new Object();
  12.     private static Object discoveryLock=new Object();
  14.     public static ServiceDiscovery getServiceDiscovery(){
  15.         if (discovery==null){
  16.             synchronized (discoveryLock){
  17.                 if (discovery==null){
  18.                     if ("nacos".equalsIgnoreCase(center)){
  19.                         discovery= new NacosServiceDiscovery(LoadBalancerFactory.getLoadBalancer(lb));
  20.                     }else if ("zookeeper".equalsIgnoreCase(center)){
  21.                         discovery= new ZookeeperServiceDiscovery(LoadBalancerFactory.getLoadBalancer(lb));
  22.                     }
  23.                 }
  24.             }
  25.         }
  26.         return discovery;
  27.     }
  29.     public static ServiceRegistry getServiceRegistry(){
  30.         if (registry==null){
  31.             synchronized (registerLock){
  32.                 if (registry==null){
  33.                     if ("nacos".equalsIgnoreCase(center)){
  34.                         registry=  new NacosServiceRegistry();
  35.                     }else if ("zookeeper".equalsIgnoreCase(center)){
  36.                         registry= new ZookeeperServiceRegistry();
  37.                     }
  38.                 }
  39.             }
  40.         }
  41.         return registry;
  42.     }
  44. }

使用Gson序列化InetSocketAddress存在问题，编写Util：

  1. public class InetSocketAddressSerializerUtil {
  2.     public static String getJsonByInetSockerAddress(InetSocketAddress address){
  3.         HashMap<String, String> map = new HashMap<>();
  4.         map.put("host",address.getHostName());
  5.         map.put("port",address.getPort()+"");
  6.         return new Gson().toJson(map);
  7.     }
  9.     public static InetSocketAddress getInetSocketAddressByJson(String json){
  10.         HashMap<String,String> hashMap = new Gson().fromJson(json, HashMap.class);
  11.         String host = hashMap.get("host");
  12.         Integer port=Integer.parseInt(hashMap.get("port"));
  13.         return new InetSocketAddress(host,port);
  14.     }
  16. }

上面主要是注册，发现的逻辑，我把主要方法写在了Utils中：

  1. @Slf4j
  2. public class ZookeeperServerUitls {
  4.     private static CuratorFramework client = ZookeeperUtil.getZookeeperClient();
  6.     private static final Set<String> instances=new ConcurrentHashSet<>();
  8.     public static void register(String serviceName, InetSocketAddress inetSocketAddress){
  10.         serviceName=ZookeeperUtil.serviceName2Path(serviceName);;
  11.         String uuid = UUID.randomUUID().toString();
  12.         serviceName=serviceName+"/"+uuid;
  13.         String json = InetSocketAddressSerializerUtil.getJsonByInetSockerAddress(inetSocketAddress);
  14.         try {
  15.             if (RpcConfig.isZookeeperDestoryIsEphemeral()){
  16.                 //会话结束节点，创建消失
  17.                 client.create()
  18.                         .creatingParentsIfNeeded()
  19.                         .withMode(CreateMode.EPHEMERAL)
  20.                         .forPath(serviceName,json.getBytes());
  21.             } else {
  22.                 client.create()
  23.                         .creatingParentsIfNeeded()
  24.                         .forPath(serviceName,json.getBytes());
  25.             }
  26.         }
  27.             catch (Exception e) {
  28.             log.error("服务注册失败");
  29.             throw  new RpcException(RpcError.REGISTER_SERVICE_FAILED);
  30.         }
  31.         //放入map
  32.         instances.add(serviceName);
  33.     }
  35.     public static void cleanRegistry(){
  36.         log.info("注销所有注册的服务");
  37.         //如果自动销毁，不需要清除
  38.         if (RpcConfig.isZookeeperDestoryIsEphemeral()) return;
  39.         if (ZookeeperUtil.getServerHostname()!=null&&ZookeeperUtil.getServerPort()!=null&&!instances.isEmpty()){
  40.             for (String path:instances) {
  41.                 try {
  42.                     client.delete().forPath(path);
  43.                 } catch (Exception e) {
  44.                     log.error("服务注销失败");
  45.                     throw new RpcException(RpcError.DESTORY_REGISTER_FALL);
  46.                 }
  47.             }
  48.         }
  49.     }
  50. }
  1. @Slf4j
  2. public class ZookeeperClientUtils {
  4.     private static CuratorFramework client = ZookeeperUtil.getZookeeperClient();
  6.     private static final Map<String,  List<InetSocketAddress>> instances=new ConcurrentHashMap<>();
  8.     public static InetSocketAddress searchService(String serviceName, LoadBalancer loadBalancer) {
  9.         InetSocketAddress address;
  10.         //本地缓存查询
  11.         if (instances.containsKey(serviceName)){
  12.             List<InetSocketAddress> addressList = instances.get(serviceName);
  13.             if (!addressList.isEmpty()){
  14.                 //使用lb进行负载均衡
  15.                 return loadBalancer.select(addressList);
  16.             }
  17.         }
  18.         try {
  19.             String path = ZookeeperUtil.serviceName2Path(serviceName);
  20.             //获取路径下所有的实现
  21.             List<String> instancePaths = client.getChildren().forPath(path);
  22.             List<InetSocketAddress> addressList = new ArrayList<>();
  23.             for (String instancePath : instancePaths) {
  24.                 byte[] bytes = client.getData().forPath(path+"/"+instancePath);
  25.                 String json = new String(bytes);
  26.                 InetSocketAddress instance = InetSocketAddressSerializerUtil.getInetSocketAddressByJson(json);
  27.                 addressList.add(instance);
  28.             }
  29.             addLocalCache(serviceName,addressList);
  30.             return loadBalancer.select(addressList);
  31.         } catch (Exception e) {
  32.             log.error("服务获取失败====>{}",e);
  33.             throw new RpcException(RpcError.SERVICE_NONE_INSTANCE);
  34.         }
  35.     }
  37.     public static void cleanLocalCache(String serviceName){
  38.         log.info("服务调用失败，清除本地缓存，重新获取实例===>{}",serviceName);
  39.         instances.remove(serviceName);
  40.     }
  42.     public static void addLocalCache(String serviceName,List<InetSocketAddress> addressList){
  43.         //直接替换原本的缓存
  44.         instances.put(serviceName,addressList);
  45.     }
  46. }

③配置文件

rpc.properties放在resources下

  1. #nacos    zookeeper
  2. #registerCenter.type=nacos
  3. registerCenter.type=zookeeper
  5. #registerCenter.host=127.0.0.1
  6. registerCenter.host=101.43.244.40
  8. #zookeeper port default 2181
  9. #registerCenter.port=9000
  10. registerCenter.port=2181
  12. registerCenter.destory.isEphemeral=false
  14. #??random?roundRobin
  15. loadBalance.type=random
  17. #kryo json jdk
  18. serializer.type=kryo
  20. server.hostName=127.0.0.1
  21. server.port=9999

④更多

声哥的代码我做了很多修改，如果上述代码和你参考的项目代码出入比较大，可以查看本文章的工程阅读。

原文链接：https://www.cnblogs.com/zko0/p/17130553.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Zookeeper热门文章

  • java Zookeeper简述
  • zookeeper-操作与应用场景-《每日五分钟搞定大数据》
  • ZooInspector 连接不到 Zookeeper 的解决方法
  • IDEA zookeeper插件的使用
  • 解决dubbo错误ip及ip乱入问题的方法
  • zookeeper集群搭建
  • 大数据开发phoenix连接hbase流程详解
  • 快来看看 Dubbo 是如何实现服务的动态发现的
  • zookeeper特性与节点说明
  • zookeeper【1】配置管理

    • Zookeeper推荐文章

  • Zookeeper安装部署
  • 得物 ZooKeeper SLA 也可以 99.99%
  • zookeeper源码(06)ZooKeeperServer及子类
  • Dubbo 我手写几行代码，就把通信模式给你解释清楚！
  • Rpc-实现Zookeeper注册中心
  • Dubbo 中 Zookeeper 注册中心原理分析
  • Java&nbsp;zookeeper服务的使用详解
  • 【入门】ZooKeeper 相关概念总结
  • zookeeper特性与节点说明
  • 快来看看 Dubbo 是如何实现服务的动态发现的

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号