zookeeper应用

通过上一篇的学习，对zookeeper大致有了一些了解，但是想在实际开发与合适的业务场景中使用，还是需要依赖更多深入的学习，同时在项目中不断的实实践，发现问题并解决，才能对技术有更清晰与独特的见解。

本文从几个方面去学习如何使用zookeeper。

1、通过原生的api进行操作
2、通过zkClient进行操作
3、使用curator进行操作
4、各种应用场景的实现

当然以上内容主要来源于他人的成果，同时经过测试与理解，更加方面自己去理解其中的要义。

zookeeper API：

1、引入zookeeper依赖jar 
<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.10</version>
</dependency>

2、建立连接
public class ZookeeperConnection implements Watcher {
    private ZookeeperProperties properties;
    public static ZooKeeper zooKeeper;
    private CountDownLatch countDownLatch = new CountDownLatch(1);

    public ZookeeperConnection(ZookeeperProperties properties){
        this.properties = properties;
    }

    public void connect(){
        try {
            zooKeeper = new ZooKeeper(properties.getAddress(),properties.getTimeout(),this);
            countDownLatch.await();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("zookeeper connected!");
    }

    public void process(WatchedEvent event) {
        if(event.getState() == Event.KeeperState.SyncConnected){
            countDownLatch.countDown();
        }
    }
}

3、API
创建节点：
    zk.create("/node", data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT)
获取子节点：
    zk.getChildren("/node", true)
获取数据：
    zk.getData("/node", true, null)
设置数据： 
    zk.setData("/node","data".getBytes(),-1)
判断是否存在：
    zk.exists("/node",watch)
删除节点：
    zk.delete("/node",-1)
关于zk的api其实不多，但是我们需要知道的是在什么情况如何搭配使用，上面主要是通过同步的方式操作，当然我们在创建节点、设置数据、删除节点时都可以通过回调函数实现异步操作。同时需要注意的是，watch是一次性的，如果调用，下次如果需要继续监听指定节点，需要再次注册。

下面会对acl、watch进行一些说明。

zkClient使用：

1、引入zkClient依赖jar
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.9</version>
</dependency>
2、建立连接
    ZkClient zkClient = new ZkClient("10.8.109.60:2181", 10000, 10000, new SerializableSerializer())
3、API调用
  创建节点：
    zkClient.create("/user",user, CreateMode.PERSISTENT)
  删除节点：
    zkClient.delete("/node") //节点不存在返回false，节点存在子节点则异常
    zkClient.deleteRecursive("/pnode") //无论有无节点或子节点都能删除
  获取子节点：
    zkClient.getChildren("/")
  获取数据：
    zkClient.readData("/user", stat) //将状态会写入stat
  设置数据：  
    zkClient.writeDataReturnStat("/", data, -1)
  节点是否存在：
    zkClient.exists("/node")
  监听：
    IZkChildListener
        zkClient.subscribeChildChanges("/node", new IZkChildListener(){...}) //监听node节点的创建/删除、创建/删除子节点
    IZkDataListener
        zkClient.subscribeDataChanges("/", new IZkDataListener(){...}) //监听指定节点的数据改变

curator使用：

1、引入依赖
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>2.4.0</version>
</dependency>
2、创建连接
RetryPolicy retryPolicyOne = new ExponentialBackoffRetry(1000, 3) //重试间隔、重试次数
CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181", 5000, 5000, retryPolicyOne)
client.start();
或者
CuratorFramework client = CuratorFrameworkFactory
        .builder()
        .connectString("localhost:2181")
        .sessionTimeoutMs(5000)
        .connectionTimeoutMs(5000)
        .retryPolicy(retryPolicyThree)
        .build()
client.start()
3、API
  创建节点：
    client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/node", "".getBytes())
  删除节点： 
// 删除(无子节点的)节点
    client.delete().forPath("/a_node");
// 删除指定版本的(无子节点的)节点
    client.delete().withVersion(4).forPath("/a_node");
// 删除节点(如果有子节点，也一并删除)
    client.delete().deletingChildrenIfNeeded().forPath("/a_node");
// 删除指定版本的节点(如果有子节点，也一并删除)
    client.delete().deletingChildrenIfNeeded().withVersion(4).forPath("/a_node");
// 只要Session有效，那么就会对指定节点进行持续删除，知道删除成功; 这是一种保障机制
    client.delete().guaranteed().deletingChildrenIfNeeded().withVersion(4).forPath("/a_node");
  获取子节点：
    client.getChildren().forPath("/")
  获取数据：
    client.getData().storingStatIn(stat).forPath("/")
  设置数据：
    client.getData().storingStatIn(statOne).forPath("/")
  节点是否存在：
    client.checkExists().forPath("/")
 监听：
    //节点监听
    NodeCache cache = new NodeCache(client,"/node")
    cache.getListenable().addListener(new NodeCacheListener() { //节点创建、节点内容改变
        @Override
        public void nodeChanged() throws Exception {
            byte[] newData = cache.getCurrentData().getData();
        }
    });
    
    //子节点监听
    final PathChildrenCache cache = new PathChildrenCache(client,"/node", true);
    cache.start();
    // 当目标节点的子节点被创建、子节点被删除、子节点数据内容发生变化等时，会触发监听方法
    cache.getListenable().addListener(new PathChildrenCacheListener() {
        @Override
        public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
            byte[] data;
            switch (event.getType()){
                case CHILD_ADDED :
                    System.out.println("新增子节点!");
                    break;
                case  CHILD_REMOVED :
                    System.out.println("删除子节点!");
                    break;
                case CHILD_UPDATED :
                    System.out.println("修改子节点数据内容!");
                    break;
                default :;
            }
        }
    });
和zkClienet相比，zkClient监听的主要分为节点与数据，而curator则是针对本节点、子节点。

ACL与Watch:

1、在新建节点时，我们可以对该节点设置对应的ACL，保证在对节点的后续操作时都必须满足ACL的设定，那么ACL具体如何理解与设置？
其实对节点的访问控制主要是什么对象对该节点有什么样的操作，那么什么对象我们用Id表示，节点则是当前创建的node，而操作则有rwacd(READ\WRITE\ADMIN\CREATE\DELETE)。
而ACL的表达式一般格式为：scheme:id:perm，在zk客户端中我们可以这样来对节点设置acl。eg: world:anyone:wr
scheme:
    world：默认方式，相当于全世界都能访问，唯一id为anyone
    auth：代表已经认证通过的用户(cli中可以通过addauth digest user:pwd 来添加当前上下文中的授权用户)，没有id
    digest：即用户名:密码这种方式认证，这也是业务系统中最常用的，,setAcl <path> digest:<user>:<password(密文)>:<acl>
           第二密码是经过sha1及base64处理的密文
           echo -n <user>:<password> | openssl dgst -binary -sha1 | openssl base64
    host：根据地质认证
    ip：使用Ip地址认证
id:
    anyone
perm:
    CREATE: create
    READ: getData getChildren
    WRITE: setData
    DELETE: delete
    ADMIN: setAcl
代码示例：  
    //world
    List<ACL> acls = ZooDefs.Ids.OPEN_ACL_UNSAFE;
    //auth
    new ACL(ZooDefs.Perms.ALL,new Id("auth","username:password"));
    //digest
    ACL aclDigest = new ACL(ZooDefs.Perms.ALL, new Id("digest", DigestAuthenticationProvider.generateDigest("username:password")));
    //host
    new ACL(ZooDefs.Perms.ALL,new Id("host","scl.com"));
    //ip
    new ACL(ZooDefs.Perms.ALL,new Id("ip","192.168.1.1/25"));
CLI示例：
    auth:
        >addauth digest <username>:<password>
        >setAcl /node auth:<username>:wradc
    digest:
        >echo -n <user>:<password> | openssl dgst -binary -sha1 | openssl base64
        >setAcl <path> digest:<user>:<BASE64[SHA-1[password]]>:<acl>
    host:
        >setAcl <apth> host:<host>:<acl>  //支持后缀 没试过
    ip:
        >setAcl <path> ip:<ip/bits>:<acl> //支持ip段，*匹配 不管怎么设置，不是Authentication is not valid就是alc is valid

watch主要是在创建节点、获取数据、判断节点是否存在是触发，针对不同的行为或者是事件被触发时所作出的响应，在zookeeper中有以下事件：
    EventType.NodeCreated
    EventType.NodeDeleted
    EventType.NodeDataChanged
    EventType.NodeChildrenChaged
这些主要是围绕节点与数据变化时对应的事件，但是要注意，NodeDeleted会影响getChildren设置的watcher，详情可以看这个网站。
-

zookeeper应用场景：

在curator-recipes中其实对以下场景基本都有实现，主要是需要了解其实现原理。

1. 命名服务 由于zookeeper中的节点（路径）的唯一性，我们可以创建唯一的服务命名；同时在分布式环境中，还可以借助节点数据的版本创建有序的分布式id。 我们可以通过数据版本dataVersion实现： @Component public class IdGenerator { private ZookeeperConnection connection; private static final String NODE_ID = "/namespace/id"; @Autowired public IdGenerator(ZookeeperConnection connection){ this.connection = connection; initNode(); } private void initNode() { try { if( connection.zk().exists(NODE_ID,false)==null){ if(connection.zk().exists("/namespace",false)==null){ connection.zk().create("/namespace",null,ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); } connection.zk().create(NODE_ID,null,ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT); } } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } } public int id(){ try { Stat stat = connection.zk().setData(NODE_ID,null, -1); if(stat!=null){ return stat.getVersion(); } } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } return -1; } } 或者通过持久化有序节点实现： @Component public class IdSequence { private ZookeeperConnection connection; private ZkClient zkClient; private static final String PARENT_NODE = "/namespace"; private static final String NODE = "/idseq"; @Autowired public IdSequence(ZookeeperConnection connection){ this.connection = connection; init(); } private void init() { connection.initZkClient(); this.zkClient = connection.getZkClient(); if(!zkClient.exists(PARENT_NODE)){ zkClient.createPersistent(PARENT_NODE); } } public String id(){ String node = zkClient.createPersistentSequential(PARENT_NODE + NODE, null); return node.substring((PARENT_NODE+NODE).length()); } }
2. 配置管理 在集群环境中，各个子系统都会有多个相同的实例，如果其相关的配置有改动，则每个系统都要统一修改，通过zookeeper，将配置写入某个节点下，每个子系统通过watcher监听节点的变化，从而能够及时响应配置的修改，而只用关注一个配置中心即可。 @Component public class ConfigManager implements Watcher,ApplicationEventPublisherAware { private String configNode = "/config"; @Autowired private ZookeeperConnection connection; private ApplicationEventPublisher applicationEventPublisher; public String watchConfig(){ try { byte[] data = connection.zk().getData(configNode, this, null); return new String(data); } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } return null; } public void setData(String config){ try { connection.zk().setData(configNode,config.getBytes(),-1); } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } } public void process(WatchedEvent event) { String config = watchConfig();//持续监听 applicationEventPublisher.publishEvent(new ConfigChangeEvent(config)); } @Override public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { this.applicationEventPublisher = applicationEventPublisher; } } @Data public class ConfigChangeEvent extends ApplicationEvent { private String config; /** * Create a new ApplicationEvent. * * @param source the object on which the event initially occurred (never {@code null}) */ public ConfigChangeEvent(Object source) { super(source); this.config = Objects.toString(source,null); } } @Component public class ConfigChangeListener implements ApplicationListener<ConfigChangeEvent> { @Autowired private ConfigRegistry configRegistry; public void onApplicationEvent(ConfigChangeEvent event) { configRegistry.run(event.getConfig()); } } @Component public class ConfigRegistry { private List<Server> servers = new ArrayList<>(); public void regist(Server... ss){ Arrays.stream(ss).forEach(s->{servers.add(s);}); } public void run(String config){ servers.stream().forEach(s->{ s.refresh(config); }); } } public class SimpleServer implements Server{ private String name; public SimpleServer(String name){ this.name = name; } public SimpleServer(){ this.name = UUID.randomUUID().toString(); } @Override public String getName() { return this.name; } @Override public void refresh(String config) { System.out.println(String.format("server :%s config is changed :%s",getName(),config)); } } @Test public void test() throws InterruptedException { Server s1 = new SimpleServer("server1"); Server s2 = new SimpleServer("server2"); Server s3 = new SimpleServer("server3"); configRegistry.regist(s1,s2,s3); configManager.watchConfig(); TimeUnit.SECONDS.sleep(Long.MAX_VALUE); } 原理基本上就是创建一个/config节点，多个服务关注该节点，即监听其数据变化，通过getData置入监听，但注意watch的一次性，需要往复监听，一段该节点数据发生变化，那么注册的所有服务将会通过发布监听事件将新的配置传递到各个子服务上。
3. 负载均衡 在传统的单应用中，往往会因为用户的激增，导致无法一次性处理较多的请求，这时候可以部署多个完全一样的应用，通过负载均衡将各个请求分发到不同系统中取，一般会用ngixn、LVS完成，当然 zookeeper也可以实现了。 -
4. 分布式锁 前面用到redis做分布式锁，主要是为了保证分布式环境中，数据的一致性，zookeeper可以通过节点的唯一性实现排它锁，通过有序节点，实现共享锁。借助curator更快地实现。 public void lock(String id){ try { this.lockPath = zk.create(PARENT_NODE.concat(NODE), id.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); List<String> children = zk.getChildren(PARENT_NODE, true); if(!CollectionUtils.isEmpty(children)){ children.sort(Comparator.comparing(String::valueOf)); if(!PARENT_NODE.concat("/").concat(children.get(0)).equals(lockPath)){ int preIndex = Collections.binarySearch(children,lockPath.substring(PARENT_NODE.length()+1)); System.out.println("id:"+id+" 锁节点："+lockPath); CountDownLatch countDownLatch = new CountDownLatch(1); zk.exists(PARENT_NODE.concat("/").concat(children.get(preIndex-1)), new Watcher() { @Override public void process(WatchedEvent event) { if(event.getType() == Event.EventType.NodeDeleted){ countDownLatch.countDown(); // System.out.println("删除锁节点："+event.getPath()); } } }); countDownLatch.await(); } System.out.println("获取锁："+lockPath+" id:"+id); } } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } } public void unlock(){ if(!StringUtils.isEmpty(lockPath)){ try { System.out.println("释放锁:"+lockPath); zk.delete(lockPath,-1); } catch (InterruptedException e) { e.printStackTrace(); } catch (KeeperException e) { e.printStackTrace(); } } } 测试： public void lock() throws InterruptedException { int count = 10; CountDownLatch countDownLatch = new CountDownLatch(count); List<String> result = new ArrayList<>(); for(int i = 0; i<count; i++){ String name = "app"+i; Thread app = new Thread(()->{ ZkLock lock = new ZkLock(connection); lock.lock(name);//阻塞 result.add(name); lock.unlock(); countDownLatch.countDown(); },name); app.start(); } countDownLatch.await(); System.out.println(result); }
5. 发布订阅 其实就是监听一个节点，但由于watch是一次性的，需要循环监听。可以通过zkClient完成 @Data public class Publisher { private static String NODE = "/info"; private String channel; private ZkClient zkClient; public Publisher(String channel,ZkClient zkClient){ this.channel = channel; this.zkClient = zkClient; if(!zkClient.exists("/".concat(channel))){ zkClient.createPersistent("/".concat(channel)); } } public void push(Object data) { zkClient.createPersistentSequential("/".concat(channel).concat(NODE),data); } public void subscribe(Subscibe subscibe) { zkClient.subscribeChildChanges("/".concat(channel),subscibe); zkClient.subscribeDataChanges("/".concat(channel),subscibe); } public void unsubscribe(Subscibe subscibe){ zkClient.unsubscribeDataChanges("/".concat(channel),subscibe); zkClient.unsubscribeChildChanges("/".concat(channel),subscibe); } } @Data public class Subscibe implements IZkChildListener ,IZkDataListener { private String channel; private ZkClient zkClient; private Callback callback; public Subscibe(String channel,ZkClient zkClient){ this.channel = channel; this.zkClient = zkClient; } @Override public void handleChildChange(String node, List<String> list) throws Exception { List<String> children = zkClient.getChildren("/".concat(channel)); if(children!=null && children.size()>0){ children.sort(Comparator.comparing(String::valueOf)); for(String child : children){ Object data = zkClient.readData("/".concat(channel).concat("/").concat(child)); if(callback!=null){ callback.call(data); } zkClient.delete("/".concat(channel).concat("/").concat(child)); } }else{ } zkClient.subscribeChildChanges("/".concat(channel),this); } @Override public void handleDataChange(String dataPath, Object data) throws Exception { System.out.println("数据修改: "+ dataPath +":" +data); } @Override public void handleDataDeleted(String dataPath) throws Exception { System.out.println("节点数据删除："+dataPath); } public interface Callback{ void call(Object data); } } @Data @Component public class PSCenterManager { private List<Publisher> publishers = new ArrayList<>(); private ZkClient zkClient; @Autowired public PSCenterManager(ZookeeperConnection connection){ this.zkClient = connection.getZkClient(); } public void addPublisher(Publisher ... _publishers){ publishers.addAll(Arrays.asList(_publishers)); } /** * 发布消息 * @param data */ public void push(Object data){ for(Publisher publisher : publishers){ publisher.push(data); } } /** * 订阅节点 * @param subscibe */ public void registry(Subscibe subscibe){ for(Publisher publisher : publishers){ if(subscibe.getChannel().equals(publisher.getChannel())){ publisher.subscribe(subscibe); } } } public void unregistry(Subscibe subscibe){ for(Publisher publisher : publishers){ if(subscibe.getChannel().equals(publisher.getChannel())){ publisher.unsubscribe(subscibe); } } } } 测试：在/channel节点下新增节点都能及时发现 public class PubSubTest extends AbstractTest { @Autowired private PSCenterManager centerManager; @Test public void test() throws InterruptedException { String channel = "channel"; ZkClient zkClient = centerManager.getZkClient(); zkClient.setZkSerializer(new ZkSerializer(){ @Override public byte[] serialize(Object data) throws ZkMarshallingError { return data==null?null:data.toString().getBytes(); } @Override public Object deserialize(byte[] bytes) throws ZkMarshallingError { return new String(bytes); } });// Publisher publisher = new Publisher(channel,zkClient); Subscibe subscibe = new Subscibe(channel,zkClient); subscibe.setCallback(new Subscibe.Callback() { @Override public void call(Object data) { System.out.println("数据:"+data); } }); centerManager.addPublisher(publisher); centerManager.registry(subscibe); for(int i=0;i<10;i++){ centerManager.push("data"+i); TimeUnit.SECONDS.sleep(1); } while (true){ } } }
6. 消息队列 类似于mq，生产者生产消息，消费者消费消息，其实核心的是如何构建一个分布式队列。通过某个节点来维护消息的通道，而具体的消息作为该节点的值。分布式消息队列要注意在进行数据存取时如何保证数据的一致性。 -
7. 集群管理 -

通过zookeeper实现各种场景的应用确实不易，还需深入理解zookeeper，虽然zk的实现非常复杂，但是暴露的api还是比较简单，而且上面说到的场景在其他的jar包中也有实现，以上可能有些地方因为理解或者测试原因可能存在一定的问题，还是需要在不断实践过程中深入理解与实践才是学习的根本。

代码：https://github.com/suspring/springboot-zookeeper