Java 类org.apache.curator.framework.recipes.shared.SharedCount 实例源码

public static void printAll(CuratorFramework client, String path,
        String prefix) {
    client.start();
    List<String> children;
    try {
        children = client.getChildren().forPath(path);
        for (String child : children) {
            if (child.startsWith(prefix)) {
                SharedCount c = new SharedCount(client, path + child, 0);
                c.start();
                System.out.println(path + child + "\t" + c.getCount());
                c.close();
            }
        }
        client.close();
    } catch (Exception e) {
        e.printStackTrace();

    }
}

@SuppressWarnings("resource")
private int getIncrementForPrefix(String prefix) {
    String path = String.format(COUNTER_PATH, prefix);
    SharedCount sharedCount = new SharedCount(curator.get(), path, 1);
    try {
        sharedCount.start();
        for (int count = sharedCount.getCount(); true; count = sharedCount.getCount()) {
            if (sharedCount.trySetCount(count + 1)) {
                return count;
            }
        }
    } catch (Exception e) {
        throw FabricException.launderThrowable(e);
    } finally {
        IOUtils.safeClose(sharedCount);
    }
}

private void incrSharedCount(SharedCount sharedCount) throws Exception {
  while (true) {
    // Loop until we successfully increment the counter
    VersionedValue<Integer> versionedValue = sharedCount.getVersionedValue();
    if (sharedCount.trySetCount(versionedValue, versionedValue.getValue() + 1)) {
      break;
    }
  }
}

private void incrSharedCount(SharedCount sharedCount) throws Exception {
  while (true) {
    // Loop until we successfully increment the counter
    VersionedValue<Integer> versionedValue = sharedCount.getVersionedValue();
    if (sharedCount.trySetCount(versionedValue, versionedValue.getValue() + 1)) {
      break;
    }
  }
}

private void incrSharedCount(SharedCount sharedCount) throws Exception {
  while (true) {
    // Loop until we successfully increment the counter
    VersionedValue<Integer> versionedValue = sharedCount.getVersionedValue();
    if (sharedCount.trySetCount(versionedValue, versionedValue.getValue() + 1)) {
      break;
    }
  }
}

private void incrSharedCount(SharedCount sharedCount) throws Exception {
  while (true) {
    // Loop until we successfully increment the counter
    VersionedValue<Integer> versionedValue = sharedCount.getVersionedValue();
    if (sharedCount.trySetCount(versionedValue, versionedValue.getValue() + 1)) {
      break;
    }
  }
}

/**
 * Creates a {@link ZooKeeperSharedCount} to store a shared count between multiple instances.
 *
 * @param path to the shared count in ZooKeeper
 * @param seedCount for the shared count
 * @return a shared count
 */
public ZooKeeperSharedCount createSharedCount(String path, int seedCount) {
    return new ZooKeeperSharedCount(
        new SharedCount(
            facade,
            path,
            seedCount));
}

public static SharedCount createAndStartCounter(String zkHosts,
        String counterPath) {
    SharedCount c = createCounter(zkHosts, counterPath);
    try {
        c.start();
        return c;
    } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException("Failed to start the counter: "
                + counterPath + "\n" + e.getMessage());
    }
}

public static SharedCount createAndStartCounter(CuratorFramework client,
        String counterPath) {
    SharedCount c = new SharedCount(client, counterPath, 0);
    try {
        c.start();
        return c;
    } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException("Failed to start the counter: "
                + counterPath + "\n" + e.getMessage());
    }
}

public static int getCounter(CuratorFramework client, String counterPath) {
    SharedCount c = new SharedCount(client, counterPath, 0);
    try {
        c.start();
        int val = c.getCount();
        c.close();
        return val;

    } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException("Failed to add the counter: "
                + counterPath + "\n" + e.getMessage());
    }
}

public static void addCounter(SharedCount c, int increment) {
    int oldVal = c.getCount();
    try {
        c.setCount(oldVal + increment);
    } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException("Failed to increment the counter: " + c);
    }
}

public static void addCounter(CuratorFramework client, String counterPath,
        int increment) {
    SharedCount c = new SharedCount(client, counterPath, 0);
    try {
        c.start();
        int oldVal = c.getCount();
        c.setCount(oldVal + increment);
        c.close();
    } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException("Failed to add the counter: "
                + counterPath + "\n" + e.getMessage());
    }
}

public static void setCounter(CuratorFramework client, String counterPath,
        int value) {
    SharedCount c = new SharedCount(client, counterPath, 0);
    try {
        c.start();
        c.setCount(value);
        c.close();
    } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException("Failed to add the counter: "
                + counterPath + "\n" + e.getMessage());
    }
}

private void incrSharedCount(SharedCount sharedCount) throws Exception {
  while (true) {
    // Loop until we successfully increment the counter
    VersionedValue<Integer> versionedValue = sharedCount.getVersionedValue();
    if (sharedCount.trySetCount(versionedValue, versionedValue.getValue() + 1)) {
      break;
    }
  }
}

private void incrSharedCount(SharedCount sharedCount) throws Exception {
  while (true) {
    // Loop until we successfully increment the counter
    VersionedValue<Integer> versionedValue = sharedCount.getVersionedValue();
    if (sharedCount.trySetCount(versionedValue, versionedValue.getValue() + 1)) {
      break;
    }
  }
}

public ZooKeeperSharedCount(SharedCount sharedCount) {
    this.sharedCount = Preconditions.checkNotNull(sharedCount);
}

public static SharedCount createCounter(String hosts, String counterPath) {
    CuratorFramework client = createClient(hosts);
    return new SharedCount(client, counterPath, 0);
}

public static int getCounter(SharedCount c) {
    return c.getCount();
}

@Test
public void testThreadedLeaseIncrease() throws Exception
{
    final Timing timing = new Timing();
    CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
    try
    {
        client.start();

        final SharedCount count = new SharedCount(client, "/foo/count", 1);
        count.start();

        final InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, "/test", count);

        ExecutorService service = Executors.newCachedThreadPool();

        final CountDownLatch latch1 = new CountDownLatch(1);
        final CountDownLatch latch2 = new CountDownLatch(1);
        Future<Object> future1 = service.submit
            (
                new Callable<Object>()
                {
                    @Override
                    public Object call() throws Exception
                    {
                        Lease lease = semaphore.acquire(timing.seconds(), TimeUnit.SECONDS);
                        Assert.assertNotNull(lease);
                        latch1.countDown();
                        lease = semaphore.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS);
                        Assert.assertNotNull(lease);
                        latch2.countDown();
                        return null;
                    }
                }
            );
        Future<Object> future2 = service.submit
            (
                new Callable<Object>()
                {
                    @Override
                    public Object call() throws Exception
                    {
                        Assert.assertTrue(latch1.await(timing.forWaiting().seconds(), TimeUnit.SECONDS));
                        timing.sleepABit(); // make sure second acquire is waiting
                        Assert.assertTrue(count.trySetCount(2));
                        //Make sure second acquire takes less than full waiting time:
                        timing.sleepABit();
                        Assert.assertTrue(latch2.await(0, TimeUnit.SECONDS));
                        return null;
                    }
                }
            );

        future1.get();
        future2.get();

        count.close();
    }
    finally
    {
        TestCleanState.closeAndTestClean(client);
    }
}

public static void main(String[] args) throws IOException, Exception {
    final Random rand = new Random();
    SharedCounterExample example = new SharedCounterExample();
    try (TestingServer server = new TestingServer()) {
        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));
        client.start();

        SharedCount baseCount = new SharedCount(client, PATH, 0);
        baseCount.addListener(example);
        baseCount.start();

        List<SharedCount> examples = Lists.newArrayList();
        ExecutorService service = Executors.newFixedThreadPool(QTY);
        for (int i = 0; i < QTY; ++i) {
            final SharedCount count = new SharedCount(client, PATH, 0);
            examples.add(count);
            Callable<Void> task = new Callable<Void>() {
                @Override
                public Void call() throws Exception {
                    count.start();
                    Thread.sleep(rand.nextInt(10000));
                    System.out.println("Increment:" + count.trySetCount(count.getVersionedValue(), count.getCount() + rand.nextInt(10)));
                    return null;
                }
            };
            service.submit(task);
        }



        service.shutdown();
        service.awaitTermination(10, TimeUnit.MINUTES);

        for (int i = 0; i < QTY; ++i) {
            examples.get(i).close();
        }
        baseCount.close();
    }


}

/**
 * Creates a {@link ZooKeeperCheckpointIDCounter} instance.
 *
 * @param client      Curator ZooKeeper client
 * @param counterPath ZooKeeper path for the counter. It's sufficient to have a path per-job.
 */
public ZooKeeperCheckpointIDCounter(CuratorFramework client, String counterPath) {
    this.client = checkNotNull(client, "Curator client");
    this.counterPath = checkNotNull(counterPath, "Counter path");
    this.sharedCount = new SharedCount(client, counterPath, 1);
}