zookeeper實戰(zhàn)之服務(wù)注冊與發(fā)現(xiàn)

CoreDump 發(fā)布于2019-08-16 18:28 / 3298人閱讀

摘要：應(yīng)用場景回顧下應(yīng)用場景數(shù)據(jù)發(fā)布與訂閱服務(wù)注冊與發(fā)現(xiàn)分布式鎖分布式隊列選舉配置中心命名服務(wù)負(fù)載均衡實現(xiàn)服務(wù)注冊與發(fā)現(xiàn)服務(wù)注冊與發(fā)現(xiàn)再來看看在服務(wù)注冊與發(fā)現(xiàn)中的應(yīng)用代碼實現(xiàn)邏輯服務(wù)注冊創(chuàng)建的臨時的有序節(jié)點臨時的話斷開連接了可以監(jiān)聽到有序節(jié)點創(chuàng)

zookeeper應(yīng)用場景

回顧下zk應(yīng)用場景：

數(shù)據(jù)發(fā)布與訂閱

服務(wù)注冊與發(fā)現(xiàn)

分布式鎖

分布式隊列

master選舉

配置中心

命名服務(wù)

負(fù)載均衡

zookeeper實現(xiàn)服務(wù)注冊與發(fā)現(xiàn)

服務(wù)注冊與發(fā)現(xiàn)

再來看看zk在服務(wù)注冊與發(fā)現(xiàn)中的應(yīng)用：

代碼實現(xiàn)邏輯：

服務(wù)注冊：

public class ServiceRegister {

    private  static final String BASE_SERVICES = "/services";
    private static final String  SERVICE_NAME="/products";

    public static  void register(String address,int port) {
        try {
            ZooKeeper zooKeeper = new ZooKeeper("localhost:2181",5000,(watchedEvent)->{});
            Stat exists = zooKeeper.exists(BASE_SERVICES + SERVICE_NAME, false);
            if(exists==null) {
                zooKeeper.create(BASE_SERVICES + SERVICE_NAME,"".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            String server_path = address+":"+port;
            //創(chuàng)建的臨時的有序節(jié)點
            //臨時的話斷開連接了可以監(jiān)聽到,有序節(jié)點創(chuàng)建代表每一個節(jié)點否則相同節(jié)點名稱無法創(chuàng)建
            zooKeeper.create(BASE_SERVICES + SERVICE_NAME+"/child",server_path.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println("產(chǎn)品服務(wù)注冊成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

發(fā)現(xiàn)服務(wù)：

    public class InitListener implements ServletContextListener {

    private  static final String BASE_SERVICES = "/services";
    private static final String  SERVICE_NAME="/products";

    private ZooKeeper zooKeeper;

    @Override
    public void contextInitialized(ServletContextEvent sce) {
        try {
             zooKeeper = new ZooKeeper("localhost:2181",5000,(watchedEvent)->{
                if(watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged  && watchedEvent.getPath().equals(BASE_SERVICES+SERVICE_NAME)) {
                    updateServiceList();
                }
            });

            updateServiceList();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    private void updateServiceList() {
       try{
           List children = zooKeeper.getChildren(BASE_SERVICES  + SERVICE_NAME, true);
           List newServerList = new ArrayList();
           for(String subNode:children) {
               byte[] data = zooKeeper.getData(BASE_SERVICES  + SERVICE_NAME + "/" + subNode, false, null);
               String host = new String(data, "utf-8");
               System.out.println("host:"+host);
               newServerList.add(host);
           }
           //保存注冊服務(wù)的ip端口信息，以供遠(yuǎn)程rpc調(diào)用
           LoadBalance.SERVICE_LIST = newServerList;
       }catch (Exception e) {
           e.printStackTrace();
       }
    } 
}

public abstract class LoadBalance {
    public volatile static List SERVICE_LIST;

    public abstract String choseServiceHost();

}

/**
 * 隨機負(fù)載均衡算法
 */
public class RamdomLoadBalance extends LoadBalance {
    @Override
    public String choseServiceHost() {
        String result = "";
        if(!CollectionUtils.isEmpty(SERVICE_LIST)) {
            int index = new Random().nextInt(SERVICE_LIST.size());
            result = SERVICE_LIST.get(index);
        }
        return result ;
    }
}

再講講幾個關(guān)鍵點
1.注冊服務(wù)的時候需要創(chuàng)建臨時節(jié)點,斷開連接的時候也就是服務(wù)端掛了后節(jié)點刪除可監(jiān)聽到
2.注冊服務(wù)的時候創(chuàng)建的是有序節(jié)點,一般來說相同服務(wù)都是有幾臺機器的,創(chuàng)建順序節(jié)點可區(qū)分多臺機器的服務(wù)
3.發(fā)現(xiàn)服務(wù)的一端監(jiān)聽服務(wù)節(jié)點的子節(jié)點，有子節(jié)點被刪除了或者有新子節(jié)點創(chuàng)建即重新發(fā)現(xiàn)服務(wù)可用的機器

dubbo中基于zk的服務(wù)注冊與發(fā)現(xiàn)也是這個原理