ZooKeeper: An introduction to ZooKeeper

ZooKeeper is a distributed, open source distributed application collaboration service. ZooKeeper is designed to encapsulate complex and error-prone distributed consistency services into an efficient and reliable set of primitives that can be delivered to users in a series of easy-to-use interfaces. Hereinafter referred to as ZK.

Docker-compose is strongly recommended for development and testing, which is very convenient. Production requires multiple independent machines for cluster implementation, otherwise there would be no need to use ZK.

version: '3.1'

services:
  zoo1:
    image: zookeeper
    restart: always
    hostname: zoo1
    ports:
      - 2181: 2181
    environment:
      ZOO_MY_ID: 1
      ZOO_SERVERS: Server. 1 = 0.0.0.0:2888-3888; 2181 server.2=zoo2:2888:3888; 2181 server.3=zoo3:2888:3888; 2181

  zoo2:
    image: zookeeper
    restart: always
    hostname: zoo2
    ports:
      - 2182: 2181
    environment:
      ZOO_MY_ID: 2
      ZOO_SERVERS: server.1=zoo1:2888:3888; 2181 Server. 2 = 0.0.0.0:2888-3888; 2181 server.3=zoo3:2888:3888; 2181

  zoo3:
    image: zookeeper
    restart: always
    hostname: zoo3
    ports:
      - 2183: 2181
    environment:
      ZOO_MY_ID: 3
      ZOO_SERVERS: server.1=zoo1:2888:3888; 2181 server.2=zoo2:2888:3888; 2181 Server. 3 = 0.0.0.0:2888-3888; 2181
Copy the code

Basic test cases

package org.yao;

import com.google.common.collect.ImmutableList;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Op;
import org.apache.zookeeper.OpResult;
import org.apache.zookeeper.Transaction;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;

import static com.google.common.truth.Truth.assertThat;

public class ZooKeeperTests {
  private String pathPrefix = "/multi";
  private ZooKeeper zk;
  private CountDownLatch startLatch;
  private CountDownLatch closeLatch;
  private AsyncCallback.MultiCallback callback;

  private String path1 = pathPrefix + "1";
  private String path2 = pathPrefix + "2";
  private byte[] data1 = {0x1};
  private byte[] data2 = {0x2};

  @Before
  public void setUp(a) throws Exception {
    // Register the callback, and after the callback, stop waiting
    startLatch = new CountDownLatch(1);
    callback =
        (int rc, String path, Object ctx, List<OpResult> opResults) -> {
          assertThat(rc).isEqualTo(KeeperException.Code.OK.intValue());
          System.out.printf("delete multi executed");
          closeLatch.countDown();
        };
    zk = new ZooKeeper("localhost".2181.new DefaultWatcher());
    startLatch.await();
  }

  @After
  public void tearDown(a) throws Exception {
      // Clean up zk links
    closeLatch.await();
    zk.close();
  }

  @Test
  public void testMulti(a) throws Exception {
    closeLatch = new CountDownLatch(1);

    // Create two znodes
    Op createOp1 = Op.create(path1, data1, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    Op createOp2 = Op.create(path2, data2, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

    // Synchronous API
    zk.multi(ImmutableList.of(createOp1, createOp2));
    System.out.println("create multi executed");

    assertThat(zk.getData(path1, false.null)).isEqualTo(data1);
    assertThat(zk.getData(path2, false.null)).isEqualTo(data2);

    // Delete two znodes
    Op deleteOp1 = Op.delete(path1, -1);
    Op deleteOp2 = Op.delete(path2, -1);

    // Asynchronous API
    zk.multi(ImmutableList.of(deleteOp1, deleteOp2), callback, null);
  }

  @Test
  public void testTransaction(a) throws Exception {
    closeLatch = new CountDownLatch(1);

    // Create two znodes
    Transaction tx = zk.transaction();
    tx.create(path1, data1, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    tx.create(path2, data2, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

    // Synchronous API
    tx.commit();
    System.out.println("transaction committed");

    assertThat(zk.getData(path1, false.null)).isEqualTo(data1);
    assertThat(zk.getData(path2, false.null)).isEqualTo(data2);

    // Delete two znodes
    tx = zk.transaction();
    tx.delete(path1, -1);
    tx.delete(path2, -1);

    // Asynchronous API
    tx.commit(callback, null);
  }

  @Test
  public void testTransactionWithCheck(a) throws Exception {
    closeLatch = new CountDownLatch(0);

    {
      Transaction tx = zk.transaction();
      tx.create(path1, data1, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
      tx.create(path2, data2, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
      tx.check(path1, 0);
      tx.check(path2, 0);
      tx.commit();
    }

    {
      Transaction tx = zk.transaction();
      tx.check(path1, 0);
      tx.check(path2, 0);
      tx.delete(path1, 0);
      tx.delete(path2, 0); tx.commit(); }}/** * getChildren does not list descendants recursively. */
  @Test
  public void testGetChilren(a) throws Exception {
    closeLatch = new CountDownLatch(0);
    List<String> paths = zk.getChildren("/a".false);
    System.out.printf("child paths: %s\n", paths);
  }

  class DefaultWatcher implements Watcher {
    @Override
    public void process(WatchedEvent event) {
      if (event.getType() == Event.EventType.None
          && event.getState() == Event.KeeperState.SyncConnected) {
        System.out.println("zookeeper client connected"); startLatch.countDown(); }}}}Copy the code

Curator is a Java client library for ZooKeeper. Curator aims to simplify the use of ZK. Code that does not use a co-curator handles its own ConnectionLossException. And provides a lock, service discovery encapsulation of the more complete implementation, to avoid duplication of wheel.

package com.zew.learn;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.junit.Test;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

public class CuratorLockTests {
    private final String connectString = "10.0.4.162:2181";
    private final String lockPath = "/locks";
    private static final int QTY = 5;
    private static final int REPETITIONS = QTY * 10;


    @Test
    public void lockTest(a) {
        // Create a QTY thread pool to simulate a concurrent terminal scenario
        ExecutorService service = Executors.newFixedThreadPool(QTY);
        final FakeLimitedResource resource = new FakeLimitedResource();
        try {
            for (int i = 0; i < QTY; ++i) {
                final int index = i;
                Callable<Void> task = () -> {
                    // Create a client for a curator
                    CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, new ExponentialBackoffRetry(1000.3));
                    try {
                        client.start();

                        ExampleClientThatLocks example = new ExampleClientThatLocks(client, lockPath, resource, "Client " + index);
                        for (int j = 0; j < REPETITIONS; ++j) {
                            example.doWork(10, TimeUnit.SECONDS); }}catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    } catch (Exception e) {
                        e.printStackTrace();
                        // log or do something
                    } finally {
                        CloseableUtils.closeQuietly(client);
                    }
                    return null;
                };
                service.submit(task);
            }

            service.shutdown();
            service.awaitTermination(10, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        assertTrue(true);
    }

    static class FakeLimitedResource {
        private final AtomicBoolean inUse = new AtomicBoolean(false);
        // Simulate finite resource usage, with a random time interval
        public void use(a) throws InterruptedException {
            if(! inUse.compareAndSet(false.true)) {
                fail("Only one client can use the lock, so this means zK is letting multiple clients get the lock.");
                throw new IllegalStateException("Only one client can use this lock");

            }

            try {
                Thread.sleep((long) (3 * Math.random()));
            } finally {
                inUse.set(false); }}}/** * The class responsible for locking */
    static class ExampleClientThatLocks {
        private final InterProcessMutex lock;
        private final FakeLimitedResource resource;
        private final String clientName;

        public ExampleClientThatLocks(CuratorFramework client, String lockPath, FakeLimitedResource resource, String clientName) {
            this.resource = resource;
            this.clientName = clientName;
            // Create a lock, although the client is different, but lockPath can be used to lock the effect
            lock = new InterProcessMutex(client, lockPath);
        }
         /** * The core code of the actual lock is the following sentence *@paramTime Indicates the duration *@paramUnit Time unit *@throwsException Throws an Exception */
        public void doWork(long time, TimeUnit unit) throws Exception {
            // Try to acquire the lock, throw an exception if it has not been acquired for longer
            if(! lock.acquire(time, unit)) {throw new IllegalStateException(clientName + "Unable to obtain lock");
            }

            try {
                System.out.println(clientName + "Got the lock.");
                resource.use();
            } finally {
                System.out.println(clientName + "Release lock");
                lock.release(); // always release the lock in a finally block}}}}Copy the code

1. ZooKeeper combat and source code analysis
2. curator.apache.org/
3. Github.com/apache/cura…