Zookeeper

        <!--zk-->
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.6</version>
        </dependency>

最基本的连接方式

@Test
    public void demo() throws IOException, InterruptedException, KeeperException {
        //连接单个server
//        ZooKeeper zk = new ZooKeeper("192.168.99.100:2181", 300000, new DemoWatcher());
        //集群连接
        ZooKeeper zk = new ZooKeeper("192.168.99.100:2181,192.168.99.100:2182,192.168.99.100:2183", 300000, new DemoWatcher());
        String node = "/app1";
        Stat stat = zk.exists(node, false);//检测/app1是否存在
        if (stat == null) {
            //创建节点
            String createResult = zk.create(node, "test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println("create: "+createResult);
        }
        //获取节点的值
        byte[] b = zk.getData(node, false, stat);
        System.out.println(new String(b));
        zk.close();
    }

    static class DemoWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            System.out.println("----------->");
            System.out.println("path:" + event.getPath());
            System.out.println("type:" + event.getType());
            System.out.println("stat:" + event.getState());
            System.out.println("<-----------");
        }
    }

利用开源项目连接，并且添加监听器

首先添加依赖

<!-- https://mvnrepository.com/artifact/com.101tec/zkclient -->
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.9</version>
</dependency>

实现

/**
 * Created by fengzp on 16/6/29.
 */
public class ZookeeperTest {

    private ZkClient zkClient;

    private String nodeName = "/zkListener";

    @Before
    public void before(){
        zkClient  = new ZkClient("192.168.99.100:2181,192.168.99.100:2182,192.168.99.100:2183");
    }

    @Test
    public void testZkListener() throws InterruptedException {
        //添加监听器
        zkClient.subscribeDataChanges(nodeName, new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {
                System.out.println("node data changed!");
                System.out.println("node=>" + s);
                System.out.println("data=>" + o);
                System.out.println("--------------");
            }

            @Override
            public void handleDataDeleted(String s) throws Exception {
                System.out.println("node data deleted!");
                System.out.println("s=>" + s);
                System.out.println("--------------");
            }
        });

        while (true){
            TimeUnit.SECONDS.sleep(5);
        }
    }

    @Test
    public void testUpdateData(){
        if (!zkClient.exists(nodeName)) {
            zkClient.createPersistent(nodeName);
        }
        zkClient.writeData(nodeName, "a");
        zkClient.writeData(nodeName, "b");
        zkClient.delete(nodeName);
        zkClient.delete(nodeName);//删除一个不存在的node，并不会报错
    }
}

1.2 zk节点的创建模式

zk节点有4种创建模式

//持久节点：节点创建后，会一直存在，不会因客户端会话失效而删除；
PERSISTENT(0, false, false),
//持久顺序节点：基本特性与持久节点一致，创建节点的过程中，zookeeper会在其名字后自动追加一个单调增长的数字后缀，作为新的节点名；
PERSISTENT_SEQUENTIAL(2, false, true),
//临时节点：客户端会话失效或连接关闭后，该节点会被自动删除，且不能再临时节点下面创建子节点，否则报如下错：org.apache.zookeeper.KeeperException$NoChildrenForEphemeralsException；
EPHEMERAL(1, true, false),
//临时顺序节点：基本特性与临时节点一致，创建节点的过程中，zookeeper会在其名字后自动追加一个单调增长的数字后缀，作为新的节点名；
EPHEMERAL_SEQUENTIAL(3, true, true);

可通过create方法来指定创建模式

zkClient.create(note, data, CreateMode.EPHEMERAL_SEQUENTIAL);

1.3 通过zk实现统一配置管理

目标：把公共的配置抽出来单独管理，而改变配置时，能快速的把一堆已经在线上运行的子应用，通通换掉相应的配置，而且还不用停机。

实现步骤：

1、公用配置不应该分散存放到各应用中，而是应该抽出来，统一存储到一个公用的位置（最容易想到的办法，放在db中，或统一的分布式cache server中，比如Redis，或其它类似的统一存储，比如ZooKeeper中）

2、对这些公用配置的添加、修改，应该有一个统一的配置管理中心应用来处理（这个也好办，做一个web应用来对这些配置做增、删、改、查即可），这里的操作就是对zk节点的操作

3、子应用创建一个zk的监听器，监听zk存放配置的节点的变化，每当节点的数据更新时，应用就相应的更新。

二、消除单点故障

2.1 思路

关键节点的单点故障(Single Point of Failure)在大型的架构中，往往是致命的。比如：SOA架构中，服务注册中心(Server Register)统一调度所有服务，如果这个节点挂了，基本上整个SOA架构也就崩溃了，另外hadoop 1.x/2.x中的namenode节点，这是hdfs的核心节点，如果namenode宕掉，hdfs也就废了。ZooKeeper的出现，很好的解决了这一难题，其核心原理如下：

1. 关键节点的运行实例（或服务器），可以跑多个，这些实例中的数据完全是相同的（即：对等设计），每个实例启动后，向ZK注册一个临时顺序节点，比如 /core-servers/server0000001, /core-servers/server0000002 ... ，最后的顺序号是由ZK自动递增的

2. 其它应用需要访问1中的核心服务器里，可以事先约定好，从ZK的这些临时节点中，挑选一个序号最小的节点，做为主服务器（即master）

3. 当master宕掉时，超过一定的时间阈值，临时节点将由ZK自动删除，这样原来序列最小的节点也就没了，客户端应用按2中的约定找最小节点的服务器时，自动会找到原来次最小的节点，继续充为master（老大挂了，老二顶上），即实现了故障转换。如果原来出问题的master恢复了，重新加入ZK，由于顺序号是一直递增，重新加入后，它将做为备胎待命。

2.2 测试代码

import org.I0Itec.zkclient.ZkClient;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Created by fengzp on 16/6/30.
 */
public class ZkSinglePointFailTest {

    @Test
    public void startServer1() throws InterruptedException {
        new MyServer("server1").start();
        while (true) {
            TimeUnit.SECONDS.sleep(5);
        }

    }

    @Test
    public void startServer2() throws InterruptedException {
        new MyServer("server2").start();
        while (true) {
            TimeUnit.SECONDS.sleep(5);
        }

    }

    @Test
    public void test() throws InterruptedException {
        MyClient clientServer = new MyClient();
        clientServer.run();

        //此时，手动停止coreServer1
        TimeUnit.SECONDS.sleep(60);

        //再次运行
        clientServer.run();

    }

    private String ServerNodeName = "/servers";

    public ZkClient getZkClient(){
        return new ZkClient("192.168.99.100:2181");
    }

    class MyClient{
        private String getServerData(){
            ZkClient zkClient = getZkClient();

            List<String> servers = zkClient.getChildren(ServerNodeName);
            if (servers.size() <= 0) {
                return null;
            }
            for (String s : servers) {
                System.out.println("server: "+s);
            }

            Object[] arr = servers.toArray();
            Arrays.sort(arr);

            String data = zkClient.readData(ServerNodeName + "/" + arr[0].toString());
            System.out.println("node：" + arr[0].toString() + ", data:" + data);
            return data;
        }

        public void run(){
            System.out.println("客户端应用运行中，正在调用：" + getServerData() + " 上的服务");
        }
    }

    class MyServer{
        private String hostName;

        public MyServer(String hostName){
            this.hostName = hostName;
        }

        public void start() {
            ZkClient zk = getZkClient();
            if (!zk.exists(ServerNodeName)){
                zk.createPersistent(ServerNodeName);
            }
            zk.createEphemeralSequential(ServerNodeName + "/server", hostName);
            System.out.println(hostName + " is running...");

        }
    }
}

View Code

首先调用startServer1方法启动server1, 然后调用startServer2方法启动server2, 因为代码里加入了死循环，所以server会一直在跑。

再调用test方法，正常应该会调用server1的服务，然后在中间我加入了60秒的睡眠，这时候手动停止server1，当再次运行时应会调用server2的服务。

输出:

三、ACL访问

3.1 acl机制(转)

zk做为分布式架构中的重要中间件，通常会在上面以节点的方式存储一些关键信息，默认情况下，所有应用都可以读写任何节点，在复杂的应用中，这不太安全，ZK通过ACL机制来解决访问权限问题，详见官网文档：http://zookeeper.apache.org/doc/r3.4.6/zookeeperProgrammers.html#sc_ZooKeeperAccessControl

总体来说，ZK的节点有5种操作权限：

CREATE、READ、WRITE、DELETE、ADMIN 也就是增、删、改、查、管理权限，这5种权限简写为crwda(即：每个单词的首字符缩写)

注：这5种权限中，delete是指对子节点的删除权限，其它4种权限指对自身节点的操作权限

身份的认证有4种方式：

world：默认方式，相当于全世界都能访问
auth：代表已经认证通过的用户(cli中可以通过addauth digest user:pwd 来添加当前上下文中的授权用户)
digest：即用户名:密码这种方式认证，这也是业务系统中最常用的
ip：使用Ip地址认证

Cli命令行下可以这样测试：

通过getAcl命令可以发现，刚创建的节点，默认是 world,anyone的认证方式，具有cdrwa所有权限

继续捣鼓：

先给/test增加了user1:+owfoSBn/am19roBPzR1/MfCblE的只读(r)权限控制，

说明：setAcl /test digest:用户名:密码:权限给节点设置ACL访问权限时，密码必须是加密后的内容，这里的+owfoSBn/am19roBPzR1/MfCblE=，对应的原文是12345 (至于这个密文怎么得来的，后面会讲到，这里先不管这个)，设置完Acl后，可以通过

getAcl /节点路径查看Acl设置

然后get /test时，提示认证无效，说明访问控制起作用了，接下来：

addauth digest user1:12345 给"上下文"增加了一个认证用户，即对应刚才setAcl的设置

然后再 get /test 就能取到数据了

最后 delete /test 成功了！原因是：根节点/默认是world:anyone:crdwa(即：全世界都能随便折腾)，所以也就是说任何人，都能对根节点/进行读、写、创建子节点、管理acl、以及删除子节点（再次映证了ACL中的delete权限应该理解为对子节点的delete权限）

刚才也提到了，setAcl /path digest这种方式，必须输入密码加密后的值，这在cli控制台上很不方便，所以下面这种方式更常用：

注意加框的部分，先用addauth digest user1:12345 增加一个认证用户，然后用 setAcl /test auth:user1:12345:r 设置权限，跟刚才的效果一样，但是密码这里输入的是明文，控制台模式下手动输入更方便。

3.2 通过java实现

import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.data.Stat;
import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
import org.junit.Test;

import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * Created by fengzp on 16/6/30.
 */
public class ZookeeperAclTest {

    private static final String zkAddress = "192.168.99.100:2181";
    private static final String testNode = "/test";
    private static final String readAuth = "read-user:123456";
    private static final String writeAuth = "write-user:123456";
    private static final String deleteAuth = "delete-user:123456";
    private static final String allAuth = "super-user:123456";
    private static final String adminAuth = "admin-user:123456";
    private static final String digest = "digest";

    @Test
    public void test() throws NoSuchAlgorithmException {
        initNode();

        System.out.println("---------------------");

        readTest();

        System.out.println("---------------------");

        writeTest();

        System.out.println("---------------------");

        changeACLTest();

        System.out.println("---------------------");

        deleteTest();

    }

    private void initNode() throws NoSuchAlgorithmException {
        ZkClient zkClient = new ZkClient(zkAddress);
        zkClient.addAuthInfo(digest, allAuth.getBytes());

        if (zkClient.exists(testNode)) {
            zkClient.delete(testNode);
            System.out.println("节点删除成功！");
        }

        List<ACL> acls = new ArrayList<>();
        acls.add(new ACL(ZooDefs.Perms.ALL, new Id(digest, DigestAuthenticationProvider.generateDigest(allAuth))));
        acls.add(new ACL(ZooDefs.Perms.READ, new Id(digest, DigestAuthenticationProvider.generateDigest(readAuth))));
        acls.add(new ACL(ZooDefs.Perms.WRITE, new Id(digest, DigestAuthenticationProvider.generateDigest(writeAuth))));
        acls.add(new ACL(ZooDefs.Perms.DELETE, new Id(digest, DigestAuthenticationProvider.generateDigest(deleteAuth))));
        acls.add(new ACL(ZooDefs.Perms.ADMIN, new Id(digest, DigestAuthenticationProvider.generateDigest(adminAuth))));
        zkClient.createPersistent(testNode, "test-data", acls);

        System.out.println(zkClient.readData(testNode).toString());
        System.out.println("节点创建成功！");
        zkClient.close();
    }

    private void readTest() {
        ZkClient zkClient = new ZkClient(zkAddress);

        try {
            System.out.println(zkClient.readData(testNode).toString());//没有认证信息，读取会出错
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }

        try {
            zkClient.addAuthInfo(digest, adminAuth.getBytes());
            System.out.println(zkClient.readData(testNode).toString());//admin权限与read权限不匹配，读取也会出错
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }

        try {
            zkClient.addAuthInfo(digest, readAuth.getBytes());
            System.out.println(zkClient.readData(testNode).toString());//只有read权限的认证信息，才能正常读取
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }

        zkClient.close();
    }

    private void writeTest() {
        ZkClient zkClient = new ZkClient(zkAddress);

        try {
            zkClient.writeData(testNode, "new-data");//没有认证信息，写入会失败
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }

        try {
            zkClient.addAuthInfo(digest, writeAuth.getBytes());
            zkClient.writeData(testNode, "new-data");//加入认证信息后,写入正常
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }

        try {
            zkClient.addAuthInfo(digest, readAuth.getBytes());
            System.out.println(zkClient.readData(testNode).toString());//读取新值验证
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }

        zkClient.close();
    }

    private void deleteTest() {
        ZkClient zkClient = new ZkClient(zkAddress);
        //zkClient.addAuthInfo(digest, deleteAuth.getBytes());
        try {
            //System.out.println(zkClient.readData(testNode));
            zkClient.delete(testNode);
            System.out.println("节点删除成功！");
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }
        zkClient.close();
    }

    private static void changeACLTest() {
        ZkClient zkClient = new ZkClient(zkAddress);
        //注：zkClient.setAcl方法查看源码可以发现，调用了readData、setAcl二个方法
        //所以要修改节点的ACL属性，必须同时具备read、admin二种权限
        zkClient.addAuthInfo(digest, adminAuth.getBytes());
        zkClient.addAuthInfo(digest, readAuth.getBytes());
        try {
            List<ACL> acls = new ArrayList<ACL>();
            acls.add(new ACL(ZooDefs.Perms.ALL, new Id(digest, DigestAuthenticationProvider.generateDigest(adminAuth))));
            zkClient.setAcl(testNode, acls);
            Map.Entry<List<ACL>, Stat> aclResult = zkClient.getAcl(testNode);
            System.out.println(aclResult.getKey());
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }
        zkClient.close();
    }
}

View Code

四、同步锁

4.1介绍(转)

目前分布式锁，比较成熟、主流的方案有基于redis及基于zookeeper的二种方案。

　　大体来讲，基于redis的分布式锁核心指令为SETNX，即如果目标key存在，写入缓存失败返回0，反之如果目标key不存在，写入缓存成功返回1，通过区分这二个不同的返回值，可以认为SETNX成功即为获得了锁。

　　redis分布式锁，看上去很简单，但其实要考虑周全，并不容易，网上有一篇文章讨论得很详细：http://blog.csdn.net/ugg/article/details/41894947/，有兴趣的可以阅读一下。

　　其主要问题在于某些异常情况下，锁的释放会有问题，比如SETNX成功，应用获得锁，这时出于某种原因，比如网络中断，或程序出异常退出，会导致锁无法及时释放，只能依赖于缓存的过期时间，但是过期时间这个值设置多大，也是一个纠结的问题，设置小了，应用处理逻辑很复杂的话，可能会导致锁提前释放，如果设置大了，又会导致锁不能及时释放，所以那篇文章中针对这些细节讨论了很多。

　　而基于zk的分布式锁，在锁的释放问题上处理起来要容易一些，其大体思路是利用zk的“临时顺序”节点，需要获取锁时，在某个约定节点下注册一个临时顺序节点，然后将所有临时节点按小从到大排序，如果自己注册的临时节点正好是最小的，表示获得了锁。(zk能保证临时节点序号始终递增，所以如果后面有其它应用也注册了临时节点，序号肯定比获取锁的应用更大）

　　当应用处理完成，或者处理过程中出现某种原因，导致与zk断开，超过时间阈值（可配置）后，zk server端会自动删除该临时节点，即：锁被释放。所有参与锁竞争的应用，只要监听父路径的子节点变化即可，有变化时（即：有应用断开或注册时），开始抢锁，抢完了大家都在一边等着，直到有新变化时，开始新一轮抢锁。

个人感觉：zk做分布式锁机制更完善，但zk抗并发的能力弱于redis，性能上略差，建议如果并发要求高，锁竞争激烈，可考虑用redis，如果抢锁的频度不高，用zk更适合。

以下是对于分布式锁所封装的一个抽象类

import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;

import java.util.Collections;
import java.util.List;

/**
 * Created by fengzp on 16/6/30.
 */
public abstract class AbstractZkLock {

    private int lockNumber = 1; //允许获取的锁数量(默认为1,即最小节点=自身时，认为获得锁)
    private ZkClient zk = null;
    private String rootNode = "/lock"; //根节点名称
    private String selfNode;
    private final String className = this.getClass().getSimpleName(); //当前实例的className
    private String selfNodeName;//自身注册的临时节点名
    private boolean handling = false;
    private static final String SPLIT = "/";
//    private String selfNodeFullName;

    /**
     * 通过Zk获取分布式锁
     */
    protected void getLock(int lockNumber) {
        setLockNumber(lockNumber);
        initBean();
        initNode();
        subscribe();
        register();
//        heartBeat();
        remainRunning();
    }

    protected void getLock() {
        getLock(1);
    }

    /**
     * 初始化结点
     */
    private void initNode() {

        String error;
        if (!rootNode.startsWith(SPLIT)) {
            error = "rootNode必须以" + SPLIT + "开头";
            System.out.println(error);
            throw new RuntimeException(error);
        }

        if (rootNode.endsWith(SPLIT)) {
            error = "不能以" + SPLIT + "结尾";
            System.out.println(error);
            throw new RuntimeException(error);
        }

        int start = 1;
        int index = rootNode.indexOf(SPLIT, start);
        String path;
        while (index != -1) {
            path = rootNode.substring(0, index);
            if (!zk.exists(path)) {
                zk.createPersistent(path);
            }
            start = index + 1;
            if (start >= rootNode.length()) {
                break;
            }
            index = rootNode.indexOf(SPLIT, start);
        }

        if (start < rootNode.length()) {
            if (!zk.exists(rootNode)) {
                zk.createPersistent(rootNode);
            }
        }

        selfNode = rootNode + SPLIT + className;

        if (!zk.exists(selfNode)) {
            zk.createPersistent(selfNode);
        }
    }

    /**
     * 向zk注册自身节点
     */
    private void register() {
        selfNodeName = zk.createEphemeralSequential(selfNode + SPLIT, StringUtils.EMPTY);
        if (!StringUtils.isEmpty(selfNodeName)) {
//            selfNodeFullName = selfNodeName;
            System.out.println("自身节点：" + selfNodeName + ",注册成功！");
            selfNodeName = selfNodeName.substring(selfNode.length() + 1);
        }
        checkMin();
    }

    /**
     * 订阅zk的节点变化
     */
    private void subscribe() {
        zk.subscribeChildChanges(selfNode, (parentPath, currentChilds) -> {
            checkMin();
        });
    }

    /**
     * 检测是否获得锁
     */
    private void checkMin() {
        List<String> list = zk.getChildren(selfNode);
        if (CollectionUtils.isEmpty(list)) {
            System.out.println(selfNode + " 无任何子节点!");
            lockFail();
            handling = false;
            return;
        }
        //按序号从小到大排
        Collections.sort(list);

        //如果自身ID在前N个锁中，则认为获取成功
        int max = Math.min(getLockNumber(), list.size());
        for (int i = 0; i < max; i++) {
            if (list.get(i).equals(selfNodeName)) {
                if (!handling) {
                    lockSuccess();
                    handling = true;
                    System.out.println("获得锁成功！");
                }
                return;
            }
        }

        int selfIndex = list.indexOf(selfNodeName);
        if (selfIndex > 0) {
            System.out.println("前面还有节点" + list.get(selfIndex - 1) + "，获取锁失败！");
        } else {
            System.out.println("获取锁失败！");
        }
        lockFail();

        handling = false;
    }

    /**
     * 获得锁成功的处理回调
     */
    protected abstract void lockSuccess();

    /**
     * 获得锁失败的处理回调
     */
    protected abstract void lockFail();

    /**
     * 初始化相关的Bean对象
     */
    protected abstract void initBean();


    protected void setZkClient(ZkClient zk) {
        this.zk = zk;
    }

    protected int getLockNumber() {
        return lockNumber;
    }

    protected void setLockNumber(int lockNumber) {
        this.lockNumber = lockNumber;
    }

    protected void setRootNode(String value) {
        this.rootNode = value;
    }

    /**
     * 防程序退出
     */
    private void remainRunning() {
        byte[] lock = new byte[0];
        synchronized (lock) {
            try {
                lock.wait();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("remainRunning出错：" + e.getMessage());
            }
        }
    }
}

View Code