zookeeper-client-examples

描述了一些ZooKeeper客戶端編碼相關(guān)的最佳實踐，并提供了可商用的樣例代碼，供大家研發(fā)的時候參考，提升大家接入ZooKeeper的效率。在生產(chǎn)環(huán)境上，ZooKeeper的地址信息往往都通過配置中心或者是k8s域名發(fā)現(xiàn)的方式獲得（如zookeeper-0.zookeeper:2181,zookeeper-1.zookeeper:2181,zookeeper-2.zookeeper:2181），這塊不是這篇文章描述的重點，以ZooKeeperConstant.SERVERS代替。平時開發(fā)中需要自己書寫ZooKeeper客戶端的時機場景可以說是少之又少，本文描述幾個常見的場景。本文中的例子均已上傳到github

使用ZooKeeper實現(xiàn)分布式Id生成

通過zk獲取不一樣的機器號，機器號取有序節(jié)點最后三位
id格式：

機器號 + 日期 + 小時 + 分鐘 + 秒 + 5位遞增號碼

以5位來計算，一秒可分近10w個id。

因為只需要在最開始的時候獲取機器號，沒有必要使用curator框架，使用ZooKeeper類即可，降低復(fù)雜度

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獲取不一樣的機器號，機器號取有序節(jié)點最后三位
     * id格式：
     * 機器號 + 日期 + 小時 + 分鐘 + 秒 + 5位遞增號碼
     * 一秒可分近10w個id
     * 需要對齊可以在每一位補零
     *
     * @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實現(xiàn)主備選舉

使用ZooKeeper做主備選舉，商用代碼和demo的主要差距就在能否正確處理SessionTimeout，如果不能正確處理SessionTimeout，主備選舉的代碼難以自愈。LeaderElectionService接收三個輸入?yún)?shù)：

• scene為場景，用來防止不同場景下主備選舉zkPath沖突
• serverId serverId用來區(qū)分主備不同實例，通常使用ip地址或hostname
• LeaderLatchListener 主備回調(diào)函數(shù)

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);
                }
            }
        }
    }

}

致謝

感謝 兵權(quán)的審稿。