        <!--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

查看全文

相关阅读:
为什么会需要消息队列(MQ)？
RBAC用户角色权限设计方案
 转：jquery 父、子页面之间页面元素的获取，方法的调用
 LeetCode Wiggle Subsequence
LeetCode Longest Arithmetic Sequence
LeetCode Continuous Subarray Sum
LeetCode Maximum Length of Repeated Subarray
LeetCode Is Subsequence
LeetCode Integer Break
LeetCode Largest Sum of Averages

原文地址：https://www.cnblogs.com/andyfengzp/p/5629347.html