Zookeeper

# 查看jdk版本
java -version

# 下载 zookeeper
# 下载对应版本安装包
wget http://mirrors.bfsu.edu.cn/apache/zookeeper/zookeeper-3.6.2/apache-zookeeper-3.6.2-bin.tar.gz
# 解压到指定安装目录
tar -zxvf apache-zookeeper-3.6.2-bin.tar.gz -C /usr/local/zookeeper
# 创建数据和日志目录
cd /usr/local/zookeeper && mv apache-zookeeper-3.6.2-bin apache-zookeeper-3.6.2
mkdir -pv data logs
# 设置自己的配置文件
vi /usr/local/zookeeper/apache-zookeeper-3.6.2/conf/zoo.cfg

# 心跳间隔 毫秒
tickTime=1000

# 启动时leader连接follower，超过多少次心跳间隔，follower连接超时
initLimit=10
# leader与follower 通信超时 最长心跳间隔次数
syncLimit=5

# 客户端连接的端口
clientPort=2181
# 客户端连接的最大数量
#maxClientCnxns=60

# 数据文件目录
dataDir=/usr/local/zookeeper/apache-zookeeper-3.6.2/data
# 日志目录
dataLogDir=/usr/local/zookeeper/apache-zookeeper-3.6.2/logs

# 最小会话超时时间
# minSessionTimeout=2000
# 最大会话超时时间
# maxSessionTimeout=20000

# 清除任务间隔（以小时为单位）
# 设置为0以禁用自动清除功能
autopurge.purgeInterval=5
# 最多保存20个文件 日志文件、快照
autopurge.snapRetainCount=20 

#开启四字命令
#4lw.commands.whitelist=*

# 集群信息
# server.A=B:C:D
# A 代表记号服务器，在dataDir目录下myid文件下记录
# B 服务器ip地址
# C 服务器与集群中的leader服务器交换信息的端口
# D 选举leader所用的端口
# server.1=127.0.0.1:3181:4181

启动zk : bin/zkServer.sh start
查看ZK服务状态: bin/zkServer.sh status
停止ZK服务: bin/zkServer.sh stop
重启ZK服务: bin/zkServer.sh restart
连接服务器 : bin/zkCli.sh -server 127.0.0.1:2181

# 创建每个服务对应的目录 这里使用2181，2182，2183三个端口，data是数据目录，conf是配置文件目录，logs是日志目录。
mkdir -pv cluster/{2181/{conf,data,logs},2182/{conf,data,logs},2183/{conf,data,logs}}

# 指定服务机器名称， zookeeper集群中的每台机器都知道其他的机器的信息，便于在集群出现问题后，重新选举leader等。
echo 1 > cluster/2181/data/myid
echo 2 > cluster/2182/data/myid
echo 3 > cluster/2183/data/myid

#为每个服务配置指定的配置文件，主要修改端口号，数据、日志目录。
# 进入2181服务目录，修改其配置文件
vi cluster/2181/conf/zoo.cfg

# 更改端口号，数据、日志目录，并配置集群机器信息
clientPort=2181
dataDir=/usr/local/zookeeper/apache-zookeeper-3.6.2/cluster/2181/data
dataLogDir=/usr/local/zookeeper/apache-zookeeper-3.6.2/cluster/2181/logs

server.1=127.0.0.1:3181:4181
server.2=127.0.0.1:3182:4182
server.3=127.0.0.1:3183:4183

# 复制配置文件到其他服务目录
cp cluster/2181/conf/zoo.cfg cluster/2182/conf/zoo.cfg
cp cluster/2181/conf/zoo.cfg cluster/2183/conf/zoo.cfg

# 修改其他服务配置文件 
vi cluster/2182/conf/zoo.cfg
vi cluster/2182/conf/zoo.cfg

# vim 快速替换 %s/2181/2182/g

./bin/zkServer.sh start /usr/local/zookeeper/apache-zookeeper-3.6.2/cluster/2181/conf/zoo.cfg
./bin/zkServer.sh start /usr/local/zookeeper/apache-zookeeper-3.6.2/cluster/2182/conf/zoo.cfg
./bin/zkServer.sh start /usr/local/zookeeper/apache-zookeeper-3.6.2/cluster/2183/conf/zoo.cfg

./bin/zkServer.sh status /usr/local/zookeeper/apache-zookeeper-3.6.2/cluster/2181/conf/zoo.cfg
./bin/zkServer.sh status /usr/local/zookeeper/apache-zookeeper-3.6.2/cluster/2182/conf/zoo.cfg
./bin/zkServer.sh status /usr/local/zookeeper/apache-zookeeper-3.6.2/cluster/2183/conf/zoo.cfg

# 创建一个节点 -e为临时节点 -s为顺序节点 -c为容器节点 -t为ttl节点
create /file "file data"

# 获取当前节点数据
get /file

# 设置节点数据
set /file "data"

# 创建子节点
create /file/test "data"

# 获取当前节点状态
stat /file

# 删除一个节点
delete /file
# 删除指定版本的节点
delete -v 1 /file

# 删除节点以及其包含的子节点
deleteall /file

# 设置节点配置
# -n 子节点个数 -b 节点数据长度
# 超出配额并不会报错，而是会在日志（*.out）中出现警告
setquota /data -n 3
setquota /data -b 100

# 查看配额
listquota /data

# 删除配额
delquota /data

# 列出节点下面的所有子节点
ls -R /

# 强制同步所有的更新操作
sync /data

package cn.sivan.test;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.util.concurrent.CountDownLatch;

/**
 * zookeeper的连接和数据库的连接不同，数据库通过DriverManager的getConnect方法就可以直接获取到连接。
 * 但zookeeper在获取连接的过程中，使用了Future。也就意味着，new之后所拿到的仅仅是一个zookeeper对象，而这个对象可能还没有连接到zookeeper服务。
 */
public class Connect implements Watcher {

    private CountDownLatch countDownLatch = new CountDownLatch(1);

    @Override
    public void process(WatchedEvent watchedEvent) {

        if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
            //连接创建成功，唤醒等待线程。
            countDownLatch.countDown();
        }
    }

    /**
     * 连接zookeeper
     * @param connStr address
     * @param timeout 超时时间
     */
    public ZooKeeper connect(String connStr, int timeout) {

        try {
            //创建zookeeper
            ZooKeeper zooKeeper = new ZooKeeper(connStr, timeout, this);

            //当前线程 等待连接连接成功的通知
            countDownLatch.await();
            return zooKeeper;
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return null;
    }
}

package cn.sivan.test;

import org.apache.zookeeper.AddWatchMode;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
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 java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class ZookeeperSimple {

    public static void main(String[] args) throws Exception {
        String path = "/chodble";
        String address = "127.0.0.1:2181";
        int timeout = 3000;

        Connect zkConnect = new Connect();
        ZooKeeper zk = zkConnect.connect(address, timeout);

        //判断节点是否存在
        if (null == zk.exists(path, true)) {

            //设置认证方式
            Id ADMIN_IDS = new Id("digest", DigestAuthenticationProvider.generateDigest("root:root"));
            Id USER_IDS = new Id("digest", DigestAuthenticationProvider.generateDigest("user:user"));
            Id ANYONE_ID_UNSAFE = ZooDefs.Ids.ANYONE_ID_UNSAFE;

            /**
             *设置ACL权限
             *Create  允许对子节点Create 操作
             *Read    允许对本节点GetChildren 和GetData 操作
             *Write   允许对本节点SetData 操作
             *Delete  允许对子节点Delete 操作(本节点也可以删除)
             *Admin   允许对本节点setAcl 操作
             *ALL = READ | WRITE | CREATE | DELETE | ADMIN;
             */
            List<ACL> aclList = new ArrayList<>();
            aclList.add(new ACL(ZooDefs.Perms.ADMIN, ADMIN_IDS));
            //aclList.add(new ACL(ZooDefs.Perms.CREATE | ZooDefs.Perms.READ | ZooDefs.Perms.WRITE | ZooDefs.Perms.DELETE, USER_IDS));
            aclList.add(new ACL(ZooDefs.Perms.READ, ANYONE_ID_UNSAFE));

            /**
             * 创建根节点
             * 不支持递归创建，即无法在父节点不存在的情况下创建一个子节点
             * 一个节点已经存在了，那么创建同名节点的时候，会抛出NodeExistException异常。如果是顺序节点,那么永远不会抛出NodeExistException异常
             * 临时节点不能有子节点
             *
             * CreateMode
             * PERSISTENT : 持久节点
             * PERSISTENT_SEQUENTIAL : 持久顺序节点
             * EPHEMERAL : 临时节点
             * EPHEMERAL_SEQUENTIAL : 临时顺序节点
             * CONTAINER : 容器节点
             * PERSISTENT_WITH_TTL : 持久TTL节点
             * PERSISTENT_SEQUENTIAL_WITH_TTL : 持久顺序TTL节点
             */
            zk.create(path, "宁川".getBytes(), aclList, CreateMode.PERSISTENT);

            //设置Acl
            zk.setACL(path, Collections.singletonList(new ACL(ZooDefs.Perms.CREATE | ZooDefs.Perms.READ | ZooDefs.Perms.WRITE | ZooDefs.Perms.DELETE, USER_IDS)), -1);
        }

        //进行认证
        zk.addAuthInfo("digest", "user:user".getBytes());

        //查看ACL权限
        Stat stat = null;
        if ((stat = zk.exists(path, true)) != null) {
            System.out.println("查看节点权限：" + zk.getACL(path, stat));
        }

        //获取子节点数量
        System.out.println("获取子节点数量：" + zk.getAllChildrenNumber(path));

        //获取所有的节点
        System.out.println("所有的节点:" + zk.getChildren("/", false));

        //获取节点的值，并设置监听
        System.out.println("获取节点的值：" + new String(zk.getData(path, (event -> {
            System.out.println("getWatch：" + event.toString());
        }), stat)));

        //添加监听
        zk.addWatch(path, (event) -> {
            System.out.println("addWatch：" + event.toString());
        }, AddWatchMode.PERSISTENT);

        //设置节点数据
        // 1 -自动维护
        System.out.println("设置节点数据：" + zk.setData(path, "123".getBytes(), -1));

        /**
         * 异步创建一个 临时顺序节点，ACL为 ip:127.0.0.1:c
         */
        zk.create("/node",
                "123".getBytes(),
                Collections.singletonList(new ACL(ZooDefs.Perms.CREATE, new Id("ip", "127.0.0.1"))),
                CreateMode.EPHEMERAL_SEQUENTIAL,
                //new AsyncCallback.StringCallback()
                (rc, path1, ctx, name) -> {
                    System.out.println("rc:" + rc);
                    System.out.println("path:" + path1);
                    System.out.println("ctx:" + ctx);
                    System.out.println("name:" + name);
                }, "传给服务端的内容,会在异步回调时传回来");
        //等待执行结果
        Thread.sleep(2000);

        //删除节点
        if ((stat = zk.exists(path, true)) != null) {
            List<String> subPaths = zk.getChildren(path, false);
            if (subPaths.isEmpty()) {
                zk.delete(path, stat.getVersion());
            } else {
                for (String subPath : subPaths) {
                    zk.delete(path + "/" + subPath, -1);
                }
            }
        }
    }
}

public enum CreateMode {

    /**
     * 持久节点
     * The znode will not be automatically deleted upon client's disconnect.
     */
    PERSISTENT(0, false, false, false, false),

    /**
     * 持久顺序节点
     * The znode will not be automatically deleted upon client's disconnect,
     * and its name will be appended with a monotonically increasing number.
     */
    PERSISTENT_SEQUENTIAL(2, false, true, false, false),

    /**
     * 临时节点
     * The znode will be deleted upon the client's disconnect.
     */
    EPHEMERAL(1, true, false, false, false),

    /**
     * 临时顺序节点
     * The znode will be deleted upon the client's disconnect, and its name
     * will be appended with a monotonically increasing number.
     */
    EPHEMERAL_SEQUENTIAL(3, true, true, false, false),

    /**
     * 容器节点
     * The znode will be a container node. Container
     * nodes are special purpose nodes useful for recipes such as leader, lock,
     * etc. When the last child of a container is deleted, the container becomes
     * a candidate to be deleted by the server at some point in the future.
     * Given this property, you should be prepared to get
     * {@link org.apache.zookeeper.KeeperException.NoNodeException}
     * when creating children inside of this container node.
     */
    CONTAINER(4, false, false, true, false),

    /**
     * 持久TTL节点
     * The znode will not be automatically deleted upon client's disconnect.
     * However if the znode has not been modified within the given TTL, it
     * will be deleted once it has no children.
     */
    PERSISTENT_WITH_TTL(5, false, false, false, true),

    /**
     * 持久顺序TTL节点
     * The znode will not be automatically deleted upon client's disconnect,
     * and its name will be appended with a monotonically increasing number.
     * However if the znode has not been modified within the given TTL, it
     * will be deleted once it has no children.
     */
    PERSISTENT_SEQUENTIAL_WITH_TTL(6, false, true, false, true);
}

public interface Ids {

        /**
         * world:anyone:adrwa
         * This Id represents anyone.
         */
        Id ANYONE_ID_UNSAFE = new Id("world", "anyone");

        /**
         * 认证后可操作
         * This Id is only usable to set ACLs. It will get substituted with the
         * Id's the client authenticated with.
         */
        Id AUTH_IDS = new Id("auth", "");

        /**
         * 完全开放的ACL，任何连接的客户端都可以操作该属性znode
         * This is a completely open ACL .
         */
        @SuppressFBWarnings(value = "MS_MUTABLE_COLLECTION", justification = "Cannot break API")
        ArrayList<ACL> OPEN_ACL_UNSAFE = new ArrayList<ACL>(Collections.singletonList(new ACL(Perms.ALL, ANYONE_ID_UNSAFE)));

        /**
         * 只有创建者才有ACL权限
         * This ACL gives the creators authentication id's all permissions.
         */
        @SuppressFBWarnings(value = "MS_MUTABLE_COLLECTION", justification = "Cannot break API")
        ArrayList<ACL> CREATOR_ALL_ACL = new ArrayList<ACL>(Collections.singletonList(new ACL(Perms.ALL, AUTH_IDS)));

        /**
         * 只读ACL
         * This ACL gives the world the ability to read.
         */
        @SuppressFBWarnings(value = "MS_MUTABLE_COLLECTION", justification = "Cannot break API")
        ArrayList<ACL> READ_ACL_UNSAFE = new ArrayList<ACL>(Collections.singletonList(new ACL(Perms.READ, ANYONE_ID_UNSAFE)));
}

nodedatachanged # 节点数据改变
nodecreate # 节点创建事件
nodedelete #节点删除事件
nodechildrenchanged # 子节点改变事件
package cn.sivan.test;

import org.apache.zookeeper.AddWatchMode;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
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 java.util.Collections;

public class WatchTest {
    public static void main(String[] args) throws Exception {

        String path = "/chodble2";
        String pathValue = "root-value";

        String subPath = path + "/node";
        String subPathValue = "sbu-value";

        String address = "127.0.0.1:2181";
        int timeout = 3000;

        Connect zkConnect = new Connect();
        ZooKeeper zk = zkConnect.connect(address, timeout);

        //判断节点是否存在
        Stat stat = null;
        if ((stat = zk.exists(path, true)) == null) {

            //创建节点
            zk.create(path, pathValue.getBytes(), Collections.singletonList(new ACL(ZooDefs.Perms.ALL, ZooDefs.Ids.ANYONE_ID_UNSAFE)), CreateMode.PERSISTENT);
        }

        //进行认证
        zk.addAuthInfo("digest", "user:user".getBytes());

        //设置节点监听
        zk.addWatch(path, event -> {
            System.out.println("addWatch-1：" + event);
        }, AddWatchMode.PERSISTENT);

        //获取值
        zk.getData(path, true, new Stat());

        //设置值
        zk.setData(path, "new-data".getBytes(), -1);

        //测试通知
        for (int i = 0; i < 10; i++) {
            //重新设置值
            zk.setData(path, ("new-data" + i).getBytes(), -1);
        }

        //设置权限
        zk.setACL(path, Collections.singletonList(new ACL(ZooDefs.Perms.ALL, new Id("digest", DigestAuthenticationProvider.generateDigest("user:user")))), -1);

        //创建子节点
        zk.create(subPath, subPathValue.getBytes(), Collections.singletonList(new ACL(ZooDefs.Perms.ALL, ZooDefs.Ids.AUTH_IDS)), CreateMode.PERSISTENT);

        //获取子节点的值
        zk.getData(subPath, true, new Stat());

        //设置子节点的值
        zk.setData(subPath, "sub-new-data".getBytes(), -1);

        //设置子节点权限
        zk.setACL(subPath, Collections.singletonList(new ACL(ZooDefs.Perms.ALL, new Id("digest", DigestAuthenticationProvider.generateDigest("user:user")))), -1);

        //获取子节点的个数
        zk.getAllChildrenNumber(path);

        //列出所有的子节点
        zk.getChildren(path, true);

        //查看节点ACl
        if ((stat = zk.exists(path, true)) != null) {
            System.out.println("查看节点权限：" + zk.getACL(path, stat));
        }

        //删除子节点
        zk.delete(subPath, -1);

        //删除节点
        zk.delete(path, -1);
    }
}