Zookeeperdistributed协调service

/                      # 根node
├── app1                # application1 根node
│   ├── config          # application1 configurationnode
│   └── servers         # application1 servernode
│       ├── server1     # server1node
│       └── server2     # server2node
└── app2                # application2 根node

# 1.  under 载Zookeeper
$ wget https://dlcdn.apache.org/zookeeper/zookeeper-3.7.1/apache-zookeeper-3.7.1-bin.tar.gz

# 2. 解压Zookeeper
$ tar -xzvf apache-zookeeper-3.7.1-bin.tar.gz
$ mv apache-zookeeper-3.7.1-bin /opt/zookeeper

# 3. configurationenvironmentvariable
$ export ZOOKEEPER_HOME=/opt/zookeeper
$ export PATH=$PATH:$ZOOKEEPER_HOME/bin

# 4. creationdataTable of Contents and logTable of Contents
$ mkdir -p /opt/zookeeper/data /opt/zookeeper/logs

# 5. creationconfigurationfile
$ cp $ZOOKEEPER_HOME/conf/zoo_sample.cfg $ZOOKEEPER_HOME/conf/zoo.cfg
$ vim $ZOOKEEPER_HOME/conf/zoo.cfg

# modifyconfigurationfile in 容
dataDir=/opt/zookeeper/data
dataLogDir=/opt/zookeeper/logs
clientPort=2181
tickTime=2000
initLimit=10
syncLimit=5
server.1=localhost:2888:3888
server.2=localhost:2889:3889
server.3=localhost:2890:3890

# 6. creationmyidfile (单node模式可以跳过) 
$ echo "1" > /opt/zookeeper/data/myid

# 7. 启动Zookeeper
$ zkServer.sh start

# 8. verificationZookeeper启动
$ zkServer.sh status
# 应该看 to Mode: standalone (单node模式)  or Mode: leader/follower (cluster模式) 

# 9. 连接Zookeeper
$ zkCli.sh -server localhost:2181

# 连接Zookeeper
$ zkCli.sh -server localhost:2181

# creationnode
tryit-1:
[zk: localhost:2181(CONNECTED) 0] create /app1 "app1 data"
Created /app1

# creation子node
[zk: localhost:2181(CONNECTED) 1] create /app1/config "config data"
Created /app1/config

# creation临时node
[zk: localhost:2181(CONNECTED) 2] create -e /app1/temp "temp data"
Created /app1/temp

# creation顺序node
[zk: localhost:2181(CONNECTED) 3] create -s /app1/seq "seq data"
Created /app1/seq0000000000

# 查看nodedata
[zk: localhost:2181(CONNECTED) 4] get /app1
app1 data
cZxid = 0x100000002
ctime = Wed Aug 09 10:00:00 CST 2023
mZxid = 0x100000002
mtime = Wed Aug 09 10:00:00 CST 2023
pZxid = 0x100000005
cversion = 3
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 9
numChildren = 3

# modifynodedata
[zk: localhost:2181(CONNECTED) 5] set /app1 "updated app1 data"

# 查看子node
[zk: localhost:2181(CONNECTED) 6] ls /app1
[config, seq0000000000, temp]

# deletenode
[zk: localhost:2181(CONNECTED) 7] delete /app1/temp

# deletepackage含子node node
[zk: localhost:2181(CONNECTED) 8] deleteall /app1

# 退出Zookeeper Shell
[zk: localhost:2181(CONNECTED) 9] quit

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class ZookeeperExample {
    private static final String CONNECT_STRING = "localhost:2181";
    private static final int SESSION_TIMEOUT = 30000;
    private static final String PATH = "/app1";
    private static ZooKeeper zk;
    private static CountDownLatch connectedSignal = new CountDownLatch(1);

    // 连接监听器
    private static Watcher watcher = event -> {
        if (event.getState() == Watcher.Event.KeeperState.SyncConnected) {
            connectedSignal.countDown();
        }
    };

    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        // 1. 连接Zookeeper
        connect();

        try {
            // 2. creationnode
            createNode();

            // 3. 获取nodedata
            getData();

            // 4. modifynodedata
            setData();

            // 5. 查看子node
            getChildren();

            // 6. deletenode
            deleteNode();
        } finally {
            // 7. 关闭连接
            close();
        }
    }

    // 连接Zookeeper
    private static void connect() throws IOException, InterruptedException {
        zk = new ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, watcher);
        connectedSignal.await();
        System.out.println("Connected to Zookeeper");
    }

    // creationnode
    private static void createNode() throws KeeperException, InterruptedException {
        String path = zk.create(PATH, "app1 data".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("Created node: " + path);

        // creation子node
        String childPath = zk.create(PATH + "/config", "config data".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("Created child node: " + childPath);
    }

    // 获取nodedata
    private static void getData() throws KeeperException, InterruptedException {
        byte[] data = zk.getData(PATH, false, null);
        System.out.println("Data at " + PATH + ": " + new String(data));
    }

    // modifynodedata
    private static void setData() throws KeeperException, InterruptedException {
        Stat stat = zk.setData(PATH, "updated app1 data".getBytes(), -1);
        System.out.println("Updated data at " + PATH + ", dataVersion: " + stat.getDataVersion());
    }

    // 查看子node
    private static void getChildren() throws KeeperException, InterruptedException {
        List children = zk.getChildren(PATH, false);
        System.out.println("Children of " + PATH + ": " + children);
    }

    // deletenode
    private static void deleteNode() throws KeeperException, InterruptedException {
        // 先delete子node
        zk.delete(PATH + "/config", -1);
        System.out.println("Deleted child node: " + PATH + "/config");

        // 再delete父node
        zk.delete(PATH, -1);
        System.out.println("Deleted node: " + PATH);
    }

    // 关闭连接
    private static void close() throws InterruptedException {
        zk.close();
        System.out.println("Disconnected from Zookeeper");
    }
}

// 1.  in Zookeeperincreationconfigurationnode
$ zkCli.sh -server localhost:2181
[zk: localhost:2181(CONNECTED) 0] create /app/config "key1=value1;key2=value2"

// 2. 客户端监听configurationnode
byte[] data = zk.getData("/app/config", watcher, null);
String config = new String(data);
// 解析configuration
Map configMap = parseConfig(config);

// 3. 当configuration发生变化时, Zookeeper会notification客户端
@Override
public void process(WatchedEvent event) {
    if (event.getType() == Event.EventType.NodeDataChanged) {
        try {
            byte[] newData = zk.getData("/app/config", this, null);
            String newConfig = new String(newData);
            // updateconfiguration
            configMap = parseConfig(newConfig);
            System.out.println("Config updated: " + newConfig);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

public class DistributedLock {
    private static final String LOCK_PATH = "/lock";
    private ZooKeeper zk;
    private String lockNode;
    private String threadName;

    public DistributedLock(ZooKeeper zk, String threadName) {
        this.zk = zk;
        this.threadName = threadName;
    }

    // 获取lock
    public boolean lock() throws KeeperException, InterruptedException {
        // creation临时顺序node
        lockNode = zk.create(LOCK_PATH + "/lock_", threadName.getBytes(), 
                            ZooDefs.Ids.OPEN_ACL_UNSAFE, 
                            CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(threadName + " created node: " + lockNode);

        // 获取所 has locknode
        List lockNodes = zk.getChildren(LOCK_PATH, false);
        // sortlocknode
        Collections.sort(lockNodes);

        // check当 before node is 否 is 第一个
        String firstLockNode = LOCK_PATH + "/" + lockNodes.get(0);
        if (lockNode.equals(firstLockNode)) {
            System.out.println(threadName + " got the lock");
            return true;
        }

        // 监听 before 一个locknode
        String prevLockNode = LOCK_PATH + "/" + lockNodes.get(lockNodes.indexOf(lockNode.substring(LOCK_PATH.length() + 1)) - 1);
        final CountDownLatch latch = new CountDownLatch(1);
        Watcher watcher = event -> {
            if (event.getType() == Event.EventType.NodeDeleted) {
                latch.countDown();
            }
        };
        zk.getData(prevLockNode, watcher, null);
        
        // etc.待 before 一个locknode释放
        latch.await();
        System.out.println(threadName + " got the lock");
        return true;
    }

    // 释放lock
    public void unlock() throws KeeperException, InterruptedException {
        zk.delete(lockNode, -1);
        System.out.println(threadName + " released the lock");
    }
}