zookeeper使用及模拟注册中心原理

什么是Zookeeper

Zookeeper是一个分布式开源框架，提供了协调分布式应用的基本服务，它向外部应用暴露一组通用服务——分布式同步（Distributed
Synchronization）、命名服务（Naming Service）、集群维护（Group
Maintenance）等，简化分布式应用协调及其管理的难度，提供高性能的分布式服务。ZooKeeper本身可以以单机模式安装运行，不过它的长处在于通过分布式ZooKeeper集群（一个Leader，多个Follower），基于一定的策略来保证ZooKeeper集群的稳定性和可用性，从而实现分布式应用的可靠性。

1、Zookeeper是为别的分布式程序服务的

2、Zookeeper本身就是一个分布式程序（只要有半数以上节点存活，zk就能正常服务）

3、Zookeeper所提供的服务涵盖：主从协调、服务器节点动态上下线、统一配置管理、分布式共享锁、统> 一名称服务等

4、虽然说可以提供各种服务，但是zookeeper在底层其实只提供了两个功能：
管理(存储，读取)用户程序提交的数据（类似namenode中存放的metadata）；
并为用户程序提供数据节点监听服务；

Zookeeper集群机制

Zookeeper集群的角色： Leader 和 follower
只要集群中有半数以上节点存活，集群就能提供服务

Zookeeper特性

1、Zookeeper：一个leader，多个follower组成的集群
2、全局数据一致：每个server保存一份相同的数据副本，client无论连接到哪个server，数据都是一致的
3、分布式读写，更新请求转发，由leader实施
4、更新请求顺序进行，来自同一个client的更新请求按其发送顺序依次执行
5、数据更新原子性，一次数据更新要么成功，要么失败
6、实时性，在一定时间范围内，client能读到最新数据

Zookeeper数据结构

1、层次化的目录结构，命名符合常规文件系统规范(类似文件系统）【节点可以存储数据，类似key/value】

2、每个节点在zookeeper中叫做znode,并且其有一个唯一的路径标识

3、节点Znode可以包含数据和子节点（但是EPHEMERAL类型的节点不能有子节点）
节点类型

a、Znode有两种类型：

临时（ephemeral）（create -e /app1/test1 “test1” 客户端断开连接zk删除ephemeral类型节点）
持久（persistent）（create -s /app1/test2 “test2” 客户端断开连接zk不删除persistent类型节点）

b、Znode有四种形式的目录节点（默认是persistent ）
PERSISTENT 持久节点，即使服务挂掉还是存在
PERSISTENT_SEQUENTIAL（持久序列/test0000000019 ）
EPHEMERAL 临时节点，服务关闭就消失了
EPHEMERAL_SEQUENTIAL 临时顺序节点

c、创建znode时设置顺序标识，znode名称后会附加一个值，顺序号是一个单调递增的计数器，由父节点维护

d、在分布式系统中，顺序号可以被用于为所有的事件进行全局排序，这样客户端可以通过顺序号推断事件的顺序

Zookeeper应用场景

数据发布与订阅（配置中心）
发布与订阅模型，即所谓的配置中心，顾名思义就是发布者将数据发布到ZK节点上，供订阅者动态获取数据，实现配置信息的集中式管理和动态更新。例如全局的配置信息，服务式服务框架的服务地址列表等就非常适合使用。

负载均衡

这里说的负载均衡是指软负载均衡。在分布式环境中，为了保证高可用性，通常同一个应用或同一个服务的提供方都会部署多份，达到对等服务。而消费者就须要在这些对等的服务器中选择一个来执行相关的业务逻辑，其中比较典型的是消息中间件中的生产者，消费者负载均衡。
消息中间件中发布者和订阅者的负载均衡，linkedin开源的KafkaMQ和阿里开源的 metaq都是通过zookeeper来做到生产者、消费者的负载均衡。这里以metaq为例如讲下：
生产者负载均衡：metaq发送消息的时候，生产者在发送消息的时候必须选择一台broker上的一个分区来发送消息，因此metaq在运行过程中，会把所有broker和对应的分区信息全部注册到ZK指定节点上，默认的策略是一个依次轮询的过程，生产者在通过ZK获取分区列表之后，会按照brokerId和partition的顺序排列组织成一个有序的分区列表，发送的时候按照从头到尾循环往复的方式选择一个分区来发送消息。

消费负载均衡：在消费过程中，一个消费者会消费一个或多个分区中的消息，但是一个分区只会由一个消费者来消费。MetaQ的消费策略是：

每个分区针对同一个group只挂载一个消费者。
如果同一个group的消费者数目大于分区数目，则多出来的消费者将不参与消费。
如果同一个group的消费者数目小于分区数目，则有部分消费者需要额外承担消费任务。
在某个消费者故障或者重启等情况下，其他消费者会感知到这一变化（通过 zookeeper watch消费者列表），然后重新进行负载均衡，保证所有的分区都有消费者进行消费。

命名服务(Naming Service)

命名服务也是分布式系统中比较常见的一类场景。在分布式系统中，通过使用命名服务，客户端应用能够根据指定名字来获取资源或服务的地址，提供者等信息。被命名的实体通常可以是集群中的机器，提供的服务地址，远程对象等等——这些我们都可以统称他们为名字（Name）。其中较为常见的就是一些分布式服务框架中的服务地址列表。通过调用ZK提供的创建节点的API，能够很容易创建一个全局唯一的path，这个path就可以作为一个名称。

阿里开源的分布式服务框架Dubbo中使用ZooKeeper来作为其命名服务，维护全局的服务地址列表， 在Dubbo实现中：
服务提供者在启动的时候，向ZK上的指定节点/dubbo/${serviceName}/providers目录下写入自己的URL地址，这个操作就完成了服务的发布。

服务消费者启动的时候，订阅/dubbo/serviceName/providers目录下的提供者URL地址，并向/dubbo/{serviceName}/providers目录下的提供者URL地址，并向/dubbo/serviceName/providers目录下的提供者URL地址，并向/dubbo/{serviceName} /consumers目录下写入自己的URL地址。

注意，所有向ZK上注册的地址都是临时节点，这样就能够保证服务提供者和消费者能够自动感应资源的变化。另外，Dubbo还有针对服务粒度的监控，方法是订阅/dubbo/${serviceName}目录下所有提供者和消费者的信息。

分布式通知/协调

ZooKeeper中特有watcher注册与异步通知机制，能够很好的实现分布式环境下不同系统之间的通知与协调，实现对数据变更的实时处理。使用方法通常是不同系统都对ZK上同一个znode进行注册，监听znode的变化（包括znode本身内容及子节点的），其中一个系统update了znode，那么另一个系统能够收到通知，并作出相应处理

另一种心跳检测机制：检测系统和被检测系统之间并不直接关联起来，而是通过zk上某个节点关联，大大减少系统耦合。
另一种系统调度模式：某系统有控制台和推送系统两部分组成，控制台的职责是控制推送系统进行相应的推送工作。管理人员在控制台作的一些操作，实际上是修改了ZK上某些节点的状态，而ZK就把这些变化通知给他们注册Watcher的客户端，即推送系统，于是，作出相应的推送任务。
另一种工作汇报模式：一些类似于任务分发系统，子任务启动后，到zk来注册一个临时节点，并且定时将自己的进度进行汇报（将进度写回这个临时节点），这样任务管理者就能够实时知道任务进度。
总之，使用zookeeper来进行分布式通知和协调能够大大降低系统之间的耦合

分布式锁

分布式锁，这个主要得益于 ZooKeeper 为我们保证了数据的强一致性。锁服务可以分为两类，一个是保持独占，另一个是控制时序。

所谓保持独占，就是所有试图来获取这个锁的客户端，最终只有一个可以成功获得这把锁。通常的做法是把 zk 上的一个 znode 看作是一把锁，通过 create znode 的方式来实现。所有客户端都去创建 /distribute_lock 节点，最终成功创建的那个客户端也即拥有了这把锁。
控制时序，就是所有视图来获取这个锁的客户端，最终都是会被安排执行，只是有个全局时序了。做法和上面基本类似，只是这里 /distributelock 已经预先存在，客户端在它下面创建临时有序节点（这个可以通过节点的属性控制：CreateMode.EPHEMERALSEQUENTIAL 来指定）。Zk 的父节点（/distribute_lock）维持一份 sequence, 保证子节点创建的时序性，从而也形成了每个客户端的全局时序。

有了上面的基础之后，我们先来看一个简单的例证，说明zk的基本使用和4种不同的节点创建，使用zk客户端API非常简单，只需要在pom中导入下面的依赖，

     <dependency><groupId>com.101tec</groupId><artifactId>zkclient</artifactId><version>0.11</version></dependency>

首先，你需要启动本地的zk服务，这里为了演示，我直接在windows下面启动了，然后我们可以使用一个可视化连接工具来操作一下zk,可以在这个面板上创建节点、删除节点，从而对节点进行方便的管理，

下面首先通过代码来看看zk的API的使用，
1、创建一个持久节点，运行一下下面的程序，

public class Test1 {private static final String CONNECT_ADDRESS = "127.0.0.1:2181";private static final int SESSION_TIMEOUT = 5000;//原子计数器，阻塞主线程private static final CountDownLatch countDown = new CountDownLatch(1);public static void main(String[] args) {ZooKeeper zooKeeper = null;try {zooKeeper = new ZooKeeper(CONNECT_ADDRESS, SESSION_TIMEOUT, new Watcher() {public void process(WatchedEvent event) {//获取事件状态KeeperState keeperState = event.getState();//获取事件类型EventType eventType = event.getType();if(KeeperState.SyncConnected == keeperState){//表示事件类型是当前节点还不存在的时候if(EventType.None == eventType){countDown.countDown();System.out.println("zookeeper启动连接成功.....");}if(EventType.NodeCreated == eventType){System.out.println("zookeeper 接收到事件通知，正在创建节点.....");}}}});//节点没有创建完毕，主线程阻塞countDown.await();String path = "/temp";zooKeeper.exists(path, true);  //允许接收节点创建成功时的通知//CreateMode.EPHEMERAL   临时节点//CreateMode.PERSISTENT  持久节点//Ids.OPEN_ACL_UNSAFE     允许所有客户端以非验证的方式连接String nodeResult = zooKeeper.create(path, "acong".getBytes(), Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);System.out.println("节点创建成功:" + nodeResult);} catch (Exception e) {e.printStackTrace();} finally {if (zooKeeper != null) {try {zooKeeper.close();} catch (InterruptedException e) {e.printStackTrace();}}}}}

控制台可以看到，节点创建成功，我们再去面板看看，

节点成功创建，

下面我们再来创建一个临时节点，只需要更换一下节点类型，同时为了模拟效果，我们需要在节点创建完毕后让线程休眠一定时间，因为临时节点创建成功后一旦连接关闭就会消失，

然后运行一下，可以看到临时节点创建成功，然后当程序运行完毕，再去控制台看一下，

这时候temp下面的子节点已经消失，大家可以参照这样的形式再去试一下其他的两种顺序节点的创建，

下面来说说zookeeper的其他的一个很有用的场景，就是作为分布式服务的注册中心是如何来实现的，用过dubbo的同学应该知道,dubbo的服务注册中心使用的是zookeeper，dubbo服务端一旦启动，就会把服务端的服务地址和服务列表通过zookeeper进行注册，客户端就可以使用了，那么这个过程是怎么实现的呢？

当你了解了zookeeper的上面的持久节点和临时节点的基本原理后就会发现这个过程其实并不难，下面用一张图来描述一下这个过程，

按照这个思路，我们可以来梳理一下我们编写代码的思路，大致分为下面几个步骤，这里为了演示方便，我只使用了一个zookeeper的服务，

创建服务端，模拟dubbo的服务端，
服务端启动时，创建一个持久的父节点，
通过提前规划好的将注册到zookeeper上面的提供服务的服务器IP和端口号，通过创建临时节点的方式创建到上述的父节点下面，key就是服务名称，value就是IP+端口，
启动客户端，去上面创建好的父节点下面寻找所有的子节点的key对应的value，这个value的值就是服务端暴露的服务器地址，
拿到这个地址后，就可以通过拼接相关的服务名称，进行服务的具体调用了【这里，如果要拿到具体的服务名，也需要将具体的服务名称也注册到注册中心，这里不做演示了】
模拟某个服务IP宕机或者下线的情况，就是当生产环境中某个提供服务的服务器挂掉了，zookeeper的事件监听功能可以及时捕获到，从而可以立即通知客户端，重新进行服务的拉取和调用；

下面就用代码来具体实现一下这个过程，这里项目demo结构很简单，由于这里为了模拟出服务端和客户端通信的效果，采用了socket的方式，

先看第一个类，ServerHandler，代码如下，

public class ServerHandler implements Runnable {private Socket socket;public ServerHandler(Socket socket) {this.socket = socket;}public void run() {BufferedReader in = null;PrintWriter out = null;try {in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));out = new PrintWriter(this.socket.getOutputStream(), true);String body = null;while (true) {body = in.readLine();if (body == null)break;System.out.println("Receive : " + body);out.println("Hello, " + body);}} catch (Exception e) {if (in != null) {try {in.close();} catch (IOException e1) {e1.printStackTrace();}}if (out != null) {out.close();}if (this.socket != null) {try {this.socket.close();} catch (IOException e1) {e1.printStackTrace();}this.socket = null;}}}
}

简单说明一下，这个类的作用是，接收客户端发过来的消息，并响应 "hello：” + 客户端消息，

zookeeper服务端，ZkServerScoekt ，这个类做的事情就是上面步骤的前三步，代码基本上按照步骤中的描述进行书写，可以结合注释一起看，然后我们启动一下这个程序，

/*** ServerScoekt服务端* @author asus**/
public class ZkServerScoekt implements Runnable {//这里是写死的端口，实际应用中，是提前规划好，然后通过配置文件或者全局的常量池进行读取private static int port = 18081;//定义父节点，同上，最好通过配置文件进行读取和管理private static String parentPath = "/congge_service";public static void main(String[] args) throws IOException {ZkServerScoekt server = new ZkServerScoekt(port);Thread thread = new Thread(server);thread.start();}public ZkServerScoekt(int port) {this.port = port;}/*** 初始服务端连接信息，并创建注册到zookeeper上面的相关节点信息*/private void registServer() {//1、建立zk连接ZkClient zkClient = new ZkClient("127.0.0.1:2181", 6000, 1000);//2、先创建父节点if(!zkClient.exists(parentPath)){zkClient.createPersistent(parentPath);}//3、创建子节点String nodeName = parentPath + "/service_" + port;String nodeValue = "127.0.0.1:"+port;if(zkClient.exists(nodeName)){zkClient.delete(nodeName);}//创建临时子节点，用于服务发现zkClient.createEphemeral(nodeName,nodeValue);System.out.println("服务节点注册成功");}public void run() {ServerSocket serverSocket = null;try {serverSocket = new ServerSocket(port);registServer();System.out.println("Server start port:" + port);Socket socket = null;while (true) {socket = serverSocket.accept();new Thread(new ServerHandler(socket)).start();}} catch (Exception e) {e.printStackTrace();} finally {try {if (serverSocket != null) {serverSocket.close();}} catch (Exception e2) {}}}}

可以看到，节点创建成功，注意下面的service_18081是临时节点，这里并不会断掉，除非我们关闭程序，为了模拟多个注册的服务端地址，我们再换一个端口号重新启动一下，

再来控制台看看，第二个临时节点也创建成功了，

到这一步，我们的前三步就做完了，也就是说，我们在zookeeper注册中心上模拟注册了两个提供服务的URL，类似于提供了一个伪集群，那么客户端要使用zookeeper上面的服务，第一步是要找到服务提供者的服务器地址，也就是IP+端口，直接在父节点下面拉到上面注册的所有临时子节点，然后遍历获取每个临时子节点的value即可，下面看看具体的客户端代码，相信有了这么多描述之后，代码的结构基本上大家也了解了，

public class ZkServerClient {public static List<String> listServer = new ArrayList<String>();public static String parent = "/congge_service";public static void main(String[] args) {initServer();ZkServerClient client = new ZkServerClient();BufferedReader console = new BufferedReader(new InputStreamReader(System.in));while (true) {String name;try {name = console.readLine();if ("exit".equals(name)) {System.exit(0);}client.send(name);} catch (IOException e) {e.printStackTrace();}}}/*** 连接zookeeper并获取所有的可用的服务URL地址列表*/public static void initServer() {// listServer.add("127.0.0.1:18080");final ZkClient zkClient = new ZkClient("127.0.0.1:2181", 6000, 1000);//获取父节点下的所有子节点数据List<String> children = zkClient.getChildren(parent);//获取子节点，并存放在全局的list集合中getChilds(zkClient, children);// 监听事件，监听子节点的value变化，模拟节点的上下线，就可以通过该回调方法检测到zkClient.subscribeChildChanges(parent, new IZkChildListener() {public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {getChilds(zkClient, currentChilds);}});}/*** 获取子节点* @param zkClient* @param currentChilds*/private static void getChilds(ZkClient zkClient, List<String> currentChilds) {System.out.println("有服务器宕机，获取到新的服务器节点信息是:" + listServer.toString());listServer.clear();for (String p : currentChilds) {String pathValue = (String) zkClient.readData(parent + "/" + p);listServer.add(pathValue);}serverCount = listServer.size();System.out.println("从zk读取到最新的信息:" + listServer.toString());}// 请求次数private static int reqestCount = 1;// 服务数量private static int serverCount = 0;// 获取当前server信息public static String getServer() {// 实现负载均衡String serverName = listServer.get(reqestCount % serverCount);System.out.println("客戶端请求次数是:" + reqestCount+",对应服务器是:" + serverName);++reqestCount;return serverName;}public void send(String name) {String server = ZkServerClient.getServer();String[] cfg = server.split(":");Socket socket = null;BufferedReader in = null;PrintWriter out = null;try {socket = new Socket(cfg[0], Integer.parseInt(cfg[1]));in = new BufferedReader(new InputStreamReader(socket.getInputStream()));out = new PrintWriter(socket.getOutputStream(), true);out.println(name);while (true) {String resp = in.readLine();if (resp == null)break;else if (resp.length() > 0) {System.out.println("Receive : " + resp);break;}}} catch (Exception e) {e.printStackTrace();} finally {if (out != null) {out.close();}if (in != null) {try {in.close();} catch (IOException e) {e.printStackTrace();}}if (socket != null) {try {socket.close();} catch (IOException e) {e.printStackTrace();}}}}
}

在上面代码中有个地方需要注意一下，就是在dubbo底层，也采用了类似的负载均衡算法，假如提供服务的有多个，则注册到zookeeper上之后，客户端拿到URL列表会使用相应的负载均衡算法获取其中一个URL进行服务调用，

下面我们启动一下客户端，可以看到客户端从zookeeper上获取到了URL地址列表信息，

下面我们手动移除一个节点，最直接的方式就是关闭server端的一个端口，然后看一下效果，我们关闭1808这个服务，客户端大概等待了2秒左右，通过上面的节点事件订阅机制重新获取到了最新的节点信息，只剩下了18082这个节点URL

控制台也可以看出来，

当我们再次启动18081服务，客户端很快就重新检测到了，整个过程非常快，zookeeper底层帮助我们实现了无需人工干扰，

以上便是本篇整合的全部，希望对看到的小伙伴们有所帮助，大家可以沿着这个思路继续思考，当我们的服务URL提供出来了，如何将我们的服务也一起注册到注册中心上去，那样客户端就可以实现真正的RPC服务调用了，最后感谢观看！