手寫zookeeper來模擬dubbo的註冊/發現

zookeeper可能平時你們直接操做的並很少,而zookeeper的要點就在於4個節點狀態(永久,有序,臨時,臨時有序)以及1個watcher的監控.

1.模擬註冊:

pom引用(本人使用的zookeeper爲3.4.6)

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.6</version>
</dependency>
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.47</version>
</dependency>

application.yml的內容爲

server:
    port: 8081
    address: localhost
zookeeper:
    address: 192.168.5.129:2181
    sessionouttime: 4000
spring:
    application:
        name: zk

先寫一個註冊中心

/**
 * 註冊中心,對外提供註冊服務
 * Created by Administrator on 2018/9/12.
 */
@Component
public class ZookeeperServer {
    private ZooKeeper zk;

    public ZooKeeper getConnection(String host,Watcher watcher) throws IOException {
        zk = new ZooKeeper(host, 500, watcher);
        return zk;
    }
}

再寫一個提供者註冊類向Zookeeper註冊

@Component
public class ZookRegister implements Watcher {
    //得到配置資源
    @Autowired
    Environment env;

    @Autowired
    private ZookeeperServer zkServer ;
    private ZooKeeper zk;
    //固定的根目錄好比公司名
    final String fixedpath = "/guanjian";

    @Value("spring.application.name")
    String servername;

    //spring容器初始化ZookRegister的實例時執行
    @PostConstruct
    public void register() throws Exception {
        String servername = env.getProperty("spring.application.name");
        String port = env.getProperty("server.port");
        String ip = env.getProperty("server.address");
        String address = env.getProperty("zookeeper.address");
//        PrivderServer.zooKeeper = zook.create();
//        ZooKeeper zooKeeper = PrivderServer.zooKeeper;
        this.zk = zkServer.getConnection(address ,this);
        Stat existsFixedpath = this.zk.exists(fixedpath, false);
        if (existsFixedpath == null) {
            //參數分別是建立的節點路徑、節點的數據、權限（此處對全部用戶開放）、節點的類型（此處是持久節點）
            zk.create(fixedpath, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        String svnode = fixedpath + "/" + servername;
        Stat existsSvnode = zk.exists(svnode, false);
        //create(String path, byte[] data, List<ACL> acl, CreateMode createMode)
        if (existsSvnode == null) {
            zk.create(svnode, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        if (ip == null || "".equals(null)) {
            //若是配置文件中沒有指定服務ip獲取本機ip
            ip = InetAddress.getLocalHost().getHostAddress();
        }
        if (port == null || "".equals(null)) {
            port = "8080";
        }
        NodeStat nodeStat = new NodeStat();
        nodeStat.setIp(ip);
        nodeStat.setPort(port);
        nodeStat.setName(servername);
        nodeStat.setNum(0);
        nodeStat.setStatus(Status.wait);
        //臨時節點的前綴是服務名稱,節點數據是服務address
        String svipmlNode = fixedpath + "/" + servername + "/" + servername;
        //重點在於這裏建立的是臨時有序節點
        zk.create(svipmlNode, JSONObject.toJSONString(nodeStat).getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    }

    @Override
    public void process(WatchedEvent watchedEvent) {

    }
}

其中NodeStat,Status類以下

@Data
public class NodeStat implements Serializable {
    private String ip;
    private String name;
    private String port;
    private Integer num;
    private String status;
    private String node;
    private String client;
}

public class Status {
    //wait無消費者,run運行中,stop禁用中
    public static final String run = "run";
    public static final String wait = "wait";
    public static final String stop = "stop";
}

啓動Springboot

@SpringBootApplication
public class ZkApplication {

   public static void main(String[] args) throws InterruptedException {
      ApplicationContext applicationContext = SpringApplication.run(ZkApplication.class, args);
      Thread.sleep(Long.MAX_VALUE);
   }
}

由於有一個長時間的休眠,當咱們進入zookeeper查詢的時候,咱們會發如今/guanjian/zk下多了一個zk0000000004的節點.

WatchedEvent state:SyncConnected type:None path:null
[zk: localhost:2181(CONNECTED) 0] ls /
[zookeeper, guanjian]
[zk: localhost:2181(CONNECTED) 1] ls /guanjian
[zk]
[zk: localhost:2181(CONNECTED) 2] ls /guanjian/zk
[]
[zk: localhost:2181(CONNECTED) 3] ls /guanjian/zk
[zk0000000004]
[zk: localhost:2181(CONNECTED) 4] get /guanjian/zk/zk0000000004
{"ip":"localhost","name":"zk","num":0,"port":"8081","status":"wait"}
cZxid = 0x2b
ctime = Thu Sep 13 11:20:03 CST 2018
mZxid = 0x2b
mtime = Thu Sep 13 11:20:03 CST 2018
pZxid = 0x2b
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x165cc5492840010
dataLength = 68
numChildren = 0

當咱們手動結束main方法後,再查詢zookeeper(注:/guanjian/zk是永久節點)

[zk: localhost:2181(CONNECTED) 5] ls /guanjian/zk              
[]

咱們發現這個節點沒有了,即這個臨時節點消失.

2.模擬發現:

發現爲在/guanjian/zk下隨機獲取一個臨時排序節點做爲咱們要用的註冊進來的服務,以進行後續操做,並更新這個節點的狀態爲run.

@Component
public class ClientComsumer implements Watcher {

    //本地緩存服務列表
    private static Map<String, List<String>> servermap;
    @Autowired
    private ZookeeperServer zkServer ;
    private ZooKeeper zk;
    @Autowired
    Environment env;

    @PostConstruct
    private void init() throws IOException {
        String address = env.getProperty("zookeeper.address");
        this.zk = zkServer.getConnection(address,this);
    }

    private List<String> getNodeList(String serverName) throws KeeperException, InterruptedException, IOException {
        if (servermap == null) {
            servermap = new HashMap<>();
        }
        Stat exists = null;
        try {
            String s = "/guanjian/" + serverName;
            exists = zk.exists(s,this);
        } catch (Exception e) {
        }

        //判斷是否存在該服務
        if (exists == null) return null;
        List<String> serverList = servermap.get(serverName);
        if (serverList != null && serverList.size() > 0) {
            return serverList;
        }
        List<String> children = zk.getChildren("/guanjian/" + serverName,this);
        List<String> list = new ArrayList<>();
        for (String s : children) {
            byte[] data = zk.getData("/guanjian/" + serverName + "/" + s, this, null);
            String datas = new String(data);
            NodeStat nodeStat = JSONObject.parseObject(datas, NodeStat.class);
            if (!Status.stop.equals(nodeStat.getStatus())) {
                list.add(datas);
            }
        }
        servermap.put(serverName, list);
        return list;
    }

    public String getServerinfo(String serverName) throws KeeperException, InterruptedException, IOException {
        try {
            List<String> nodeList = getNodeList(serverName);
            if (nodeList == null|| nodeList.size()<1) {
                return null;
            }
            //這裏使用得隨機負載策略，如需須要本身能夠實現其餘得負載策略
            String snode = nodeList.get((int) (Math.random() * nodeList.size()));
            NodeStat nodeStat = JSONObject.parseObject(snode, NodeStat.class);
            List<String> children = zk.getChildren("/guanjian/" + serverName,this);
            //隨機負載後,將隨機取得節點後的狀態更新爲run
            for (String s : children) {
                byte[] data = zk.getData("/guanjian/" + serverName + "/" + s, this, null);
                String datas = new String(data);
                if (snode.equals(datas)) {
                    nodeStat.setStatus(Status.run);
                    zk.setData("/guanjian/" + serverName + "/" + s,JSONObject.toJSONString(nodeStat).getBytes(),0);
                    break;
                }
            }
            return JSONObject.toJSONString(nodeStat);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        //若是服務節點數據發生變化則清空本地緩存
        if (watchedEvent.getType().equals(Event.EventType.NodeChildrenChanged)) {
            servermap = null;
        }
    }
}

調整main方法

@SpringBootApplication
public class ZkApplication {

   public static void main(String[] args) throws InterruptedException, IOException, KeeperException {
      ApplicationContext applicationContext = SpringApplication.run(ZkApplication.class, args);
      ClientComsumer getServer = applicationContext.getBean(ClientComsumer.class);
      System.out.println(getServer.getServerinfo("zk"));
      Thread.sleep(Long.MAX_VALUE);
   }
}

查看zookeeper內容以下:

[zk: localhost:2181(CONNECTED) 6] ls /guanjian/zk [zk0000000005] [zk: localhost:2181(CONNECTED) 7] get /guanjian/zk/zk0000000005 {"ip":"localhost","name":"zk","num":0,"port":"8081","status":"run"} cZxid = 0x31 ctime = Thu Sep 13 13:41:27 CST 2018 mZxid = 0x32 mtime = Thu Sep 13 13:41:27 CST 2018 pZxid = 0x31 cversion = 0 dataVersion = 1 aclVersion = 0 ephemeralOwner = 0x165cc5492840013 dataLength = 67 numChildren = 0