手写zookeeper来模拟dubbo的注册/发现

zookeeper可能平时大家直接操作的并不多,而zookeeper的要点就在于4个节点状态(永久,有序,临时,临时有序)以及1个watcher的监控.

1.模拟注册:

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

<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.6</version>

<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.47</version>

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