zookeeper-client-examples

ZooKeeperZooKeeperZooKeeperk8szookeeper-0.zookeeper:2181,zookeeper-1.zookeeper:2181,zookeeper-2.zookeeper:2181``ZooKeeperConstant.SERVERS``ZooKeepergithub

ZooKeeperId

zk id

 +  +  +  +  + 5

510wid

curator``ZooKeeper

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;

import java.time.LocalDateTime;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @author hezhangjian
 */
@Slf4j
public class ZkIdGenerator {

    private final String path = "/zk-id";

    private final AtomicInteger atomicInteger = new AtomicInteger();

    private final AtomicReference<String> machinePrefix = new AtomicReference<>("");

    private static final String[] AUX_ARRAY = {"", "0", "00", "000", "0000", "00000"};

    /**
     * zk
     * id
     *  +  +  +  +  + 5
     * 10wid
     * 
     *
     * @return
     */
    public Optional<String> genId() {
        if (machinePrefix.get().equals("")) {
            acquireMachinePrefix();
        }
        if (machinePrefix.get().length() == 0) {
            // get id failed
            return Optional.empty();
        }
        final LocalDateTime now = LocalDateTime.now();
        int aux = atomicInteger.getAndAccumulate(1, ((left, right) -> {
            int val = left + right;
            return val > 99999 ? 1 : val;
        }));
        String time = conv2Str(now.getDayOfYear(), 3) + conv2Str(now.getHour(), 2) + conv2Str(now.getMinute(), 2) + conv2Str(now.getSecond(), 2);
        String suffix = conv2Str(aux, 5);
        return Optional.of(machinePrefix.get() + time + suffix);
    }

    private synchronized void acquireMachinePrefix() {
        if (machinePrefix.get().length() > 0) {
            return;
        }
        try {
            ZooKeeper zooKeeper = new ZooKeeper(ZooKeeperConstant.SERVERS, 30_000, null);
            final String s = zooKeeper.create(path, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
            if (s.length() > 3) {
                machinePrefix.compareAndSet("", s.substring(s.length() - 3));
            }
        } catch (Exception e) {
            log.error("connect to zookeeper failed, exception is ", e);
        }
    }

    private static String conv2Str(int value, int length) {
        if (length > 5) {
            throw new IllegalArgumentException("length should be less than 5");
        }
        String str = String.valueOf(value);
        return AUX_ARRAY[length - str.length()] + str;
    }

}

ZooKeeper

ZooKeeperdemoSessionTimeout****SessionTimeoutLeaderElectionService

• scene``zkPath
• serverId serverIdip``hostname
• LeaderLatchListener

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * @author hezhangjian
 */
@Slf4j
public class LeaderElectionService {

    private final ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("zookeeper-init").build();

    private final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1, threadFactory);

    private final CuratorFramework framework;

    private final LeaderLatch leaderLatch;

    private final String zkPath;

    public LeaderElectionService(String scene, String serverId, LeaderLatchListener listener) {
        this.framework = CuratorFrameworkFactory.newClient(ZooKeeperConstant.SERVERS, new ExponentialBackoffRetry(1000, 3));
        this.zkPath = String.format("/election/%s", scene);
        this.leaderLatch = new LeaderLatch(framework, zkPath, serverId);
        leaderLatch.addListener(listener);
        executorService.execute(this::init);
    }

    private void init() {
        initStep1();
        initStep2();
        initStep3();
        executorService.shutdown();
    }

    private void initStep1() {
        while (true) {
            try {
                this.framework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(zkPath);
                break;
            } catch (Exception e) {
                log.error("create parent path exception is ", e);
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    private void initStep2() {
        while (true) {
            try {
                this.framework.start();
                break;
            } catch (Exception e) {
                log.error("create parent path exception is ", e);
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    private void initStep3() {
        while (true) {
            try {
                this.leaderLatch.start();
                break;
            } catch (Exception e) {
                log.error("create parent path exception is ", e);
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    public void close() {
        if (leaderLatch != null) {
            try {
                leaderLatch.close();
            } catch (Exception e) {
                log.error("leader latch close exception ", e);
            }
        }
        if (framework != null) {
            try {
                framework.close();
            } catch (Exception e) {
                log.error("frame close exception ", e);
            }
        }
    }

    static class ConnListener implements ConnectionStateListener {

        private final String path;

        private final String serverId;

        public ConnListener(String path, String serverId) {
            this.path = path;
            this.serverId = serverId;
        }


        @Override
        public void stateChanged(CuratorFramework client, ConnectionState newState) {
            if (newState != ConnectionState.LOST) {
                return;
            }
            while (true) {
                try {
                    client.getZookeeperClient().blockUntilConnectedOrTimedOut();
                    client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, serverId.getBytes(StandardCharsets.UTF_8));
                    break;
                } catch (Exception e) {
                    log.error("rebuild exception ", e);
                }
            }
        }
    }

}