1.前言

本文章是筆主在聲哥的手寫RPC框架的學習下，對注冊中心的一個拓展，因為聲哥某些部分沒有保留拓展性，所以本文章的專案與聲哥的工程有部磁區別，核心內容在Curator的注冊發現與注銷，思想看準即可，

本文章Git倉庫：zko0/zko0-rpc

聲哥的RPC專案寫的確實很詳細，跟學一遍受益匪淺：

何人聽我楚狂聲的博客

在聲哥的專案里使用Nacos作為了服務注冊中心，本人拓展添加了ZooKeeper實作服務注冊，

Nacos的服務注冊和發現，設計的不是非常好，每次服務的發現都需要去注冊中心拉取，本人實作ZooKeeper注冊中心時，參考了Dubbo的設計原理，結合本人自身想法，添加了本地快取：

Client發現服務后快取在本地，維護一個服務——實體串列
當監聽到注冊中心的服務串列發生了變化，Client更新本地串列
當注冊中心宕機，Client能夠依靠本地的服務串列繼續提供服務

問題：

實作服務注冊的本地快取，還需要實作注冊中心的監聽，當注冊中心的服務發生更改時能夠實作動態更新，或者用輪訓的方式，定時更新，不過這種方式的服務實時性較差
當Server宕機，非臨時節點注冊容易出現服務殘留無法清除的問題，所以我建議全部使用臨時節點去注冊，

2.內容

zookeeper需要簡單學一下，知識內容非常簡單，搭建也很簡單，在此跳過，

如果你感興趣，可以參考我的ZooKeeper的文章：Zookeeper學習筆記 - zko0

①添加依賴

Curator：(簡化ZooKeeper客戶端使用)(Netfix研發,捐給Apache,是Apache頂級專案)

這里排除slf4j依賴，因為筆主使用的slf4j存在沖突

<!-- https://mvnrepository.com/artifact/org.apache.curator/curator-recipes -->
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>5.2.0</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
        </exclusion>
    </exclusions>
</dependency>

②代碼撰寫

1.首先創建一個連接類：

@Slf4j
public class ZookeeperUtil {

    //內部化構造方法
    private ZookeeperUtil(){
    }

    private static final String SERVER_HOSTNAME= RegisterCenterConfig.getHostName();

    private static final Integer SERVER_PORT=RegisterCenterConfig.getServerPort();

    private static CuratorFramework zookeeperClient;

    public static CuratorFramework getZookeeperClient(){
        if (zookeeperClient==null){
            synchronized (ZookeeperUtil.class){
                if (zookeeperClient==null){
                    RetryPolicy retryPolic=new ExponentialBackoffRetry(3000,10);
                    zookeeperClient = CuratorFrameworkFactory.builder()
                            .connectString(SERVER_HOSTNAME+":"+SERVER_PORT)
                            .retryPolicy(retryPolic)
                            //  zookeeper根目錄為/serviceRegister，不為/
                            .namespace("serviceRegister")
                            .build();
                    zookeeperClient.start();
                }
            }
        }
        return zookeeperClient;
    }

    public static String getServerHostname(){
        return SERVER_HOSTNAME;
    }

    public static Integer getServerPort(){
        return SERVER_PORT;
    }

}

其中HOST，PORT資訊我保存在regiserCenter.properties組態檔夾中，使用類讀取：

public class RpcConfig {


    //注冊中心型別
    private static String registerCenterType;

    //序列化型別
    private static String serializerType;

    //負載均衡型別
    private static String loadBalanceType;

    //配置Nacos地址
    private static String registerCenterHost;

    private static Integer registerCenterPort;


    private static boolean zookeeperDestoryIsEphemeral;

    private static String serverHostName;

    private static Integer serverPort;

    static {
        ResourceBundle bundle = ResourceBundle.getBundle("rpc");
        registerCenterType=bundle.getString("registerCenter.type");
        loadBalanceType=bundle.getString("loadBalance.type");
        registerCenterHost=bundle.getString("registerCenter.host");
        registerCenterPort = Integer.parseInt(bundle.getString("registerCenter.port"));
        try {
            zookeeperDestoryIsEphemeral="true".equals(bundle.getString("registerCenter.destory.isEphemeral"));
        } catch (Exception e) {
            zookeeperDestoryIsEphemeral=false;
        }
        serializerType=bundle.getString("serializer.type");
        serverHostName=bundle.getString("server.hostName");
        serverPort=Integer.parseInt(bundle.getString("server.port"));
    }

    public static String getRegisterCenterType() {
        return registerCenterType;
    }

    public static String getSerializerType() {
        return serializerType;
    }

    public static String getLoadBalanceType() {
        return loadBalanceType;
    }

    public static String getRegisterCenterHost() {
        return registerCenterHost;
    }

    public static Integer getRegisterCenterPort() {
        return registerCenterPort;
    }


    public static String getServerHostName() {
        return serverHostName;
    }

    public static Integer getServerPort() {
        return serverPort;
    }

    public static boolean isZookeeperDestoryIsEphemeral() {
        return zookeeperDestoryIsEphemeral;
    }

}

下面的代碼我和聲哥有些不同，我將服務注冊，注銷方法放在ServerUtils中，服務發現方法放在ClientUtils中：

服務的高一致性存在兩種做法：

因為ZooKeeper存在臨時節點，注冊中心可以實作Client（RPC的Server）斷開，注冊服務資訊的自動丟失
不設定為臨時節點，手動的服務注冊清除

我這里兩種都實作了，雖然做兩種方式不同但是功能相同的代碼放在一起看起來很奇怪，這里只是做演示，選擇其中一種即可，（我建議使用臨時節點，當Server宕機，殘留的服務資訊也能及時清除）

注冊實作原理圖：

626723792516afae5530e634e691794

介面：

public interface ServiceDiscovery {
    InetSocketAddress searchService(String serviceName);

    void cleanLoaclCache(String serviceName);
}

public interface ServiceRegistry {
    //服務注冊
    void register(String serviceName, InetSocketAddress inetAddress);

    void cleanRegistry();
}

ZooKeeper介面實作：

public class ZookeeperServiceDiscovery implements ServiceDiscovery{

    private final LoadBalancer loadBalancer;

    public ZookeeperServiceDiscovery(LoadBalancer loadBalancer) {
        this.loadBalancer = loadBalancer;
    }
    
    @Override
    public InetSocketAddress searchService(String serviceName) {
        return ZookeeperClientUtils.searchService(serviceName,loadBalancer);
    }

    @Override
    public void cleanLoaclCache(String serviceName) {
        ZookeeperClientUtils.cleanLocalCache(serviceName);
    }
}

public class ZookeeperServiceRegistry implements ServiceRegistry{
    @Override
    public void register(String serviceName, InetSocketAddress inetAddress) {
        ZookeeperServerUitls.register(serviceName,inetAddress);
    }

    @Override
    public void cleanRegistry() {
        ZookeeperServerUitls.cleanRegistry();
    }
}

Factory工廠：

public class ServiceFactory {

    private static String center = RpcConfig.getRegisterCenterType();
    private static String lb= RpcConfig.getLoadBalanceType();

    private static  ServiceRegistry registry;

    private static  ServiceDiscovery discovery;

    private static Object registerLock=new Object();

    private static Object discoveryLock=new Object();

    public static ServiceDiscovery getServiceDiscovery(){
        if (discovery==null){
            synchronized (discoveryLock){
                if (discovery==null){
                    if ("nacos".equalsIgnoreCase(center)){
                        discovery= new NacosServiceDiscovery(LoadBalancerFactory.getLoadBalancer(lb));
                    }else if ("zookeeper".equalsIgnoreCase(center)){
                        discovery= new ZookeeperServiceDiscovery(LoadBalancerFactory.getLoadBalancer(lb));
                    }
                }
            }
        }
        return discovery;
    }

    public static ServiceRegistry getServiceRegistry(){
        if (registry==null){
            synchronized (registerLock){
                if (registry==null){
                    if ("nacos".equalsIgnoreCase(center)){
                        registry=  new NacosServiceRegistry();
                    }else if ("zookeeper".equalsIgnoreCase(center)){
                        registry= new ZookeeperServiceRegistry();
                    }
                }
            }
        }
        return registry;
    }

}

使用Gson序列化InetSocketAddress存在問題，撰寫Util：

public class InetSocketAddressSerializerUtil {
    public static String getJsonByInetSockerAddress(InetSocketAddress address){
        HashMap<String, String> map = new HashMap<>();
        map.put("host",address.getHostName());
        map.put("port",address.getPort()+"");
        return new Gson().toJson(map);
    }

    public static InetSocketAddress getInetSocketAddressByJson(String json){
        HashMap<String,String> hashMap = new Gson().fromJson(json, HashMap.class);
        String host = hashMap.get("host");
        Integer port=Integer.parseInt(hashMap.get("port"));
        return new InetSocketAddress(host,port);
    }

}

上面主要是注冊，發現的邏輯，我把主要方法寫在了Utils中：

@Slf4j
public class ZookeeperServerUitls {

    private static CuratorFramework client = ZookeeperUtil.getZookeeperClient();

    private static final Set<String> instances=new ConcurrentHashSet<>();

    public static void register(String serviceName, InetSocketAddress inetSocketAddress){

        serviceName=ZookeeperUtil.serviceName2Path(serviceName);;
        String uuid = UUID.randomUUID().toString();
        serviceName=serviceName+"/"+uuid;
        String json = InetSocketAddressSerializerUtil.getJsonByInetSockerAddress(inetSocketAddress);
        try {
            if (RpcConfig.isZookeeperDestoryIsEphemeral()){
                //會話結束節點，創建消失
                client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL)
                        .forPath(serviceName,json.getBytes());
            } else {
                client.create()
                        .creatingParentsIfNeeded()
                        .forPath(serviceName,json.getBytes());
            }
        }
            catch (Exception e) {
            log.error("服務注冊失敗");
            throw  new RpcException(RpcError.REGISTER_SERVICE_FAILED);
        }
        //放入map
        instances.add(serviceName);
    }

    public static void cleanRegistry(){
        log.info("注銷所有注冊的服務");
        //如果自動銷毀，不需要清除
        if (RpcConfig.isZookeeperDestoryIsEphemeral()) return;
        if (ZookeeperUtil.getServerHostname()!=null&&ZookeeperUtil.getServerPort()!=null&&!instances.isEmpty()){
            for (String path:instances) {
                try {
                    client.delete().forPath(path);
                } catch (Exception e) {
                    log.error("服務注銷失敗");
                    throw new RpcException(RpcError.DESTORY_REGISTER_FALL);
                }
            }
        }
    }
}

@Slf4j
public class ZookeeperClientUtils {

    private static CuratorFramework client = ZookeeperUtil.getZookeeperClient();

    private static final Map<String,  List<InetSocketAddress>> instances=new ConcurrentHashMap<>();

    public static InetSocketAddress searchService(String serviceName, LoadBalancer loadBalancer) {
        InetSocketAddress address;
        //本地快取查詢
        if (instances.containsKey(serviceName)){
            List<InetSocketAddress> addressList = instances.get(serviceName);
            if (!addressList.isEmpty()){
                //使用lb進行負載均衡
                return loadBalancer.select(addressList);
            }
        }
        try {
            String path = ZookeeperUtil.serviceName2Path(serviceName);
            //獲取路徑下所有的實作
            List<String> instancePaths = client.getChildren().forPath(path);
            List<InetSocketAddress> addressList = new ArrayList<>();
            for (String instancePath : instancePaths) {
                byte[] bytes = client.getData().forPath(path+"/"+instancePath);
                String json = new String(bytes);
                InetSocketAddress instance = InetSocketAddressSerializerUtil.getInetSocketAddressByJson(json);
                addressList.add(instance);
            }
            addLocalCache(serviceName,addressList);
            return loadBalancer.select(addressList);
        } catch (Exception e) {
            log.error("服務獲取失敗====>{}",e);
            throw new RpcException(RpcError.SERVICE_NONE_INSTANCE);
        }
    }

    public static void cleanLocalCache(String serviceName){
        log.info("服務呼叫失敗，清除本地快取，重新獲取實體===>{}",serviceName);
        instances.remove(serviceName);
    }

    public static void addLocalCache(String serviceName,List<InetSocketAddress> addressList){
        //直接替換原本的快取
        instances.put(serviceName,addressList);
    }
}

③組態檔

rpc.properties放在resources下

#nacos    zookeeper
#registerCenter.type=nacos
registerCenter.type=zookeeper

#registerCenter.host=127.0.0.1
registerCenter.host=101.43.244.40

#zookeeper port default 2181
#registerCenter.port=9000
registerCenter.port=2181

registerCenter.destory.isEphemeral=false

#??random?roundRobin
loadBalance.type=random

#kryo json jdk
serializer.type=kryo

server.hostName=127.0.0.1
server.port=9999

④更多

聲哥的代碼我做了很多修改，如果上述代碼和你參考的專案代碼出入比較大，可以查看本文章的工程閱讀，

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/544169.html

標籤：Java

上一篇：正式拋棄 Feign！Spring 6 推出新特性：HTTP Interface，這波太秀了！

下一篇：2、HashMap原始碼分析

Rpc-實作Zookeeper注冊中心

1.前言

2.內容

①添加依賴

②代碼撰寫

③組態檔

④更多