Zookeeper-Zookeeper应用场景实战

@Test
public void createTest() throws KeeperException, InterruptedException {
String path = zooKeeper.create(ZK_NODE, "data".getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
 log.info("created path: {}",path);
 }

@Test
 public void createAsycTest() throws InterruptedException {
 zooKeeper.create(ZK_NODE, "data".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
        CreateMode.PERSISTENT,
        (rc, path, ctx, name) -> log.info("rc {},path {},ctx {},name
{}",rc,path,ctx,name),"context");
    TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
 }

@Test
 public void setTest() throws KeeperException, InterruptedException {

 Stat stat = new Stat();
 byte[] data = zooKeeper.getData(ZK_NODE, false, stat);
 log.info("修改前: {}",new String(data));
 zooKeeper.setData(ZK_NODE, "changed!".getBytes(), stat.getVersion());
 byte[] dataAfter = zooKeeper.getData(ZK_NODE, false, stat);
 log.info("修改后: {}",new String(dataAfter));
 }

<!-- zookeeper client -->
 <dependency>
     <groupId>org.apache.zookeeper</groupId>
     <artifactId>zookeeper</artifactId>
     <version>3.8.0</version>
 </dependency>

 <!--curator-->
 <dependency>
     <groupId>org.apache.curator</groupId>
         <artifactId>curator-recipes</artifactId>
             <version>5.1.0</version>
             <exclusions>
                 <exclusion>
                     <groupId>org.apache.zookeeper</groupId>
                     <artifactId>zookeeper</artifactId>
                 </exclusion>
             </exclusions>
  </dependency>

创建一个客户端实例

// 重试策略
 RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3)
 //创建客户端实例
CuratorFramework client = CuratorFrameworkFactory.newClient(zookeeperConnectionString,
retryPolicy);

 //启动客户端
 client.start();

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
 CuratorFramework client = CuratorFrameworkFactory.builder()
 .connectString("192.168.128.129:2181")
 .sessionTimeoutMs(5000) // 会话超时时间
 .connectionTimeoutMs(5000) // 连接超时时间
 .retryPolicy(retryPolicy)
 .namespace("base") // 包含隔离名称
 .build();
 client.start();

Curator Caches:

node cache:

public NodeCache(CuratorFramework client,
 String path)
 Parameters:
 client - the client
 path - path to cache

public void addListener(NodeCacheListener listener)
 Add a change listener
 Parameters:
 listener - the listener
 @Slf4j
 public class NodeCacheTest extends AbstractCuratorTest{

 public static final String NODE_CACHE="/node-cache";

 @Test
 public void testNodeCacheTest() throws Exception {

 createIfNeed(NODE_CACHE);
 NodeCache nodeCache = new NodeCache(curatorFramework, NODE_CACHE);
 nodeCache.getListenable().addListener(new NodeCacheListener() {
 @Override
 public void nodeChanged() throws Exception {
 log.info("{} path nodeChanged: ",NODE_CACHE);
 printNodeData();
 }
 });

 nodeCache.start();
 }


public void printNodeData() throws Exception {
 byte[] bytes = curatorFramework.getData().forPath(NODE_CACHE);
 log.info("data: {}",new String(bytes));
 }
 }

path cache:

public PathChildrenCache(CuratorFramework client,
 String path,
 boolean cacheData)
 Parameters:
 client - the client
 path - path to watch
 cacheData - if true, node contents are cached in addition to the stat

public void addListener(PathChildrenCacheListener listener)
 Add a change listener
 Parameters:
 listener - the listener
 @Slf4j
 public class PathCacheTest extends AbstractCuratorTest{

 public static final String PATH="/path-cache";

 @Test
 public void testPathCache() throws Exception {

 createIfNeed(PATH);
 PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework,
PATH, true);
 pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
 @Override
 public void childEvent(CuratorFramework client, PathChildrenCacheEvent
event) throws Exception {
 log.info("event: {}",event);
 }
 });

 // 如果设置为true则在首次启动时就会缓存节点内容到Cache中
 pathChildrenCache.start(true);
 }
 }

tree cache:

public TreeCache(CuratorFramework client,
 String path,
 boolean cacheData)
 Parameters:
 client - the client
 path - path to watch
 cacheData - if true, node contents are cached in addition to the stat

public void addListener(TreeCacheListener listener)
 Add a change listener
 Parameters:
 listener - the listener
 @Slf4j
 public class TreeCacheTest extends AbstractCuratorTest{

 public static final String TREE_CACHE="/tree-path";

 @Test
 public void testTreeCache() throws Exception {
 createIfNeed(TREE_CACHE);
 TreeCache treeCache = new TreeCache(curatorFramework, TREE_CACHE);
 treeCache.getListenable().addListener(new TreeCacheListener() {
 @Override
 public void childEvent(CuratorFramework client, TreeCacheEvent event)
throws Exception {
 log.info(" tree cache: {}",event);
 }
 });
 treeCache.start();
 }
 }

2. Zookeeper在分布式命名服务中的实战

2.1 分布式API目录

2.2 分布式节点的命名

@Slf4j
 public class IDMaker extends CuratorBaseOperations {

 private String createSeqNode(String pathPefix) throws Exception {
 CuratorFramework curatorFramework = getCuratorFramework();
 //创建一个临时顺序节点
 String destPath = curatorFramework.create()
 .creatingParentsIfNeeded()
 .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
 .forPath(pathPefix);
 return destPath;
 }

 public String makeId(String path) throws Exception {
 String str = createSeqNode(path);
 if(null != str){
 //获取末尾的序号
 int index = str.lastIndexOf(path);
 if(index>=0){
 index+=path.length();
 return index<=str.length() ? str.substring(index):"";
 }
 }
 return str;
 }
 }

 @Test
 public void testMarkId() throws Exception {
 IDMaker idMaker = new IDMaker();
 idMaker.init();
 String pathPrefix = "/idmarker/id-";

 for(int i=0;i<5;i++){
 new Thread(()->{
 for (int j=0;j<10;j++){
 String id = null;
 try {
 id = idMaker.makeId(pathPrefix);
 log.info("{}线程第{}个创建的id为
{}",Thread.currentThread().getName(),
 j,id);
 } catch (Exception e) {
 e.printStackTrace();
 }
 }
 },"thread"+i).start();
 }

 Thread.sleep(Integer.MAX_VALUE);

 }

基于Zookeeper实现SnowFlakeID算法

public class SnowflakeIdGenerator {

 /**
 * 单例
 */
 public static SnowflakeIdGenerator instance =
 new SnowflakeIdGenerator();


 /**
 * 初始化单例
 *
 * @param workerId 节点Id,最大8091
 * @return the 单例
 */
 public synchronized void init(long workerId) {
 if (workerId > MAX_WORKER_ID) {
 // zk分配的workerId过大
 throw new IllegalArgumentException("woker Id wrong: " + workerId);
 }
 instance.workerId = workerId;
 }

 private SnowflakeIdGenerator() {

 }


 /**
 * 开始使用该算法的时间为: 2017-01-01 00:00:00
 */
 private static final long START_TIME = 1483200000000L;

 /**
 * worker id 的bit数，最多支持8192个节点
 */
 private static final int WORKER_ID_BITS = 13;

 /**
 * 序列号，支持单节点最高每毫秒的最大ID数1024
 */
 private final static int SEQUENCE_BITS = 10;

 /**
 * 最大的 worker id ，8091
 * -1 的补码（二进制全1）右移13位, 然后取反
 */
 private final static long MAX_WORKER_ID = ~(-1L << WORKER_ID_BITS);

 /**
 * 最大的序列号，1023
 * -1 的补码（二进制全1）右移10位, 然后取反
 */
 private final static long MAX_SEQUENCE = ~(-1L << SEQUENCE_BITS);

 /**
 * worker 节点编号的移位
 */
 private final static long WORKER_ID_SHIFT = SEQUENCE_BITS;

 /**
 * 时间戳的移位
 */
 private final static long TIMESTAMP_LEFT_SHIFT = WORKER_ID_BITS + SEQUENCE_BITS;

 /**
 * 该项目的worker 节点 id
 */
 private long workerId;

 /**
 * 上次生成ID的时间戳
 */
 private long lastTimestamp = -1L;

 /**
 * 当前毫秒生成的序列
 */
 private long sequence = 0L;

 /**
 * Next id long.
 *
 * @return the nextId
 */
 public Long nextId() {
 return generateId();
 }

 /**
 * 生成唯一id的具体实现
 */
 private synchronized long generateId() {
 long current = System.currentTimeMillis();

 if (current < lastTimestamp) {
 // 如果当前时间小于上一次ID生成的时间戳，说明系统时钟回退过，出现问题返回-1
 return -1;
 }

 if (current == lastTimestamp) {
 // 如果当前生成id的时间还是上次的时间，那么对sequence序列号进行+1
 sequence = (sequence + 1) & MAX_SEQUENCE;

 if (sequence == MAX_SEQUENCE) {
 // 当前毫秒生成的序列数已经大于最大值，那么阻塞到下一个毫秒再获取新的时间戳
 current = this.nextMs(lastTimestamp);
 }
 } else {
 // 当前的时间戳已经是下一个毫秒
 sequence = 0L;
 }

 // 更新上次生成id的时间戳
 lastTimestamp = current;

 // 进行移位操作生成int64的唯一ID

 //时间戳右移动23位
 long time = (current - START_TIME) << TIMESTAMP_LEFT_SHIFT;

 //workerId 右移动10位
 long workerId = this.workerId << WORKER_ID_SHIFT;

 return time | workerId | sequence;
 }

 /**
 * 阻塞到下一个毫秒
 */
 private long nextMs(long timeStamp) {
 long current = System.currentTimeMillis();
 while (current <= timeStamp) {
 current = System.currentTimeMillis();
 }
 return current;
 }
 }

/**
 * 入队
 * @param data
 * @throws Exception
 */
 public void enqueue(String data) throws Exception {
 // 创建临时有序子节点
 zk.create(QUEUE_ROOT + "/queue-", data.getBytes(StandardCharsets.UTF_8),
 ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
 }

 /**
 * 出队
 * @return
 * @throws Exception
 */
 public String dequeue() throws Exception {
 while (true) {
 List<String> children = zk.getChildren(QUEUE_ROOT, false);
 if (children.isEmpty()) {
 return null;
 }

 Collections.sort(children);

 for (String child : children) {
 String childPath = QUEUE_ROOT + "/" + child;
 try {
 byte[] data = zk.getData(childPath, false, null);
 zk.delete(childPath, -1);
 return new String(data, StandardCharsets.UTF_8);
 } catch (KeeperException.NoNodeException e) {
 // 节点已被其他消费者删除，尝试下一个节点
 }
 }
 }
 }

public class CuratorDistributedQueueDemo {
 private static final String QUEUE_ROOT = "/curator_distributed_queue";

 public static void main(String[] args) throws Exception {
 CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181",
 new ExponentialBackoffRetry(1000, 3));
 client.start();

 // 定义队列序列化和反序列化
 QueueSerializer<String> serializer = new QueueSerializer<String>() {
 @Override
 public byte[] serialize(String item) {
 return item.getBytes();
 }

 @Override
 public String deserialize(byte[] bytes) {
 return new String(bytes);
 }
 };

 // 定义队列消费者
 QueueConsumer<String> consumer = new QueueConsumer<String>() {
 @Override
 public void consumeMessage(String message) throws Exception {
 System.out.println("消费消息: " + message);
 }

 @Override
 public void stateChanged(CuratorFramework curatorFramework, ConnectionState
 connectionState) {

 }
 };

 // 创建分布式队列
 DistributedQueue<String> queue = QueueBuilder.builder(client, consumer,
serializer, QUEUE_ROOT)
 .buildQueue();
 queue.start();

 // 生产消息
 for (int i = 0; i < 5; i++) {
 String message = "Task-" + i;
 System.out.println("生产消息: " + message);
 queue.put(message);
 Thread.sleep(1000);
 }

 Thread.sleep(10000);
queue.close();
 client.close();
 }
 }

// 创建分布式队列
 QueueBuilder<String> builder = QueueBuilder.builder(client, consumer, serializer,
"/order");
 //指定了一个锁节点路径/orderlock,用于实现分布式锁，以保证队列操作的原子性和顺序性。
 queue = builder.lockPath("/orderlock").buildQueue();
 //启动队列,这时队列开始监听ZooKeeper中/order节点下的消息。
 queue.start();