ZK implementation of distributed global counter for cursor application scenario analysis

Curator Implement distributed global counters

In a stand-alone environment , How to realize thread safety self increment among multiple threads ( Count ) Implementation method and principle of , What about in a distributed environment ？ natural synchronized,lock,atomicInteger Based on Java Method cannot be satisfied , Because these can only be in the current JVM In the environment , In a distributed environment, there are multiple JVM Instance is a normal thing

Let's introduce Curator be based on Zookeeper Distributed counters implemented

Curator recipes The package implements DistributedAtomicInteger,DistributedAtomicLong Equally distributed atomic self increasing counter

Simple application

public class CuratorAtomicInteger {
    /** zookeeper Address  */
    static final String CONNECT_ADDR = "172.16.158.11:2181,"
            + "172.16.158.12:2181,"
            + "172.16.158.13:2181";
    /** session Timeout time  */
    static final int SESSION_OUTTIME = 5000;//ms

    public static void main(String[] args) throws Exception {

        //1  Retrying strategy ： The time for the first test is 1s  retry 10 Time 
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
        //2  Create connections through the factory 
        CuratorFramework cf = CuratorFrameworkFactory.builder()
                .connectString(CONNECT_ADDR)
                .sessionTimeoutMs(SESSION_OUTTIME)
                .retryPolicy(retryPolicy)
                .build();
        //3  Open the connection 
        cf.start();
        //cf.delete().forPath("/super");


        //4  Use DistributedAtomicInteger
        DistributedAtomicInteger atomicIntger =
                new DistributedAtomicInteger(cf, "/super", new RetryNTimes(3, 1000));

        AtomicValue<Integer> value = atomicIntger.add(1);
        System.out.println(value.succeeded());
        System.out.println(value.postValue());	// Latest value 
        System.out.println(value.preValue());	// Original value 

    }
}

The results returned to normal , No problem

Advanced applications （ Multi-threaded environment ）

So in the case of multithreading , Is it right ？ To be verified

public class CuratorAtomicInteger2 {
    static CountDownLatch countDownLatch = new CountDownLatch(10);

    public static void main(String[] args) throws Exception {
        CuratorFramework zkClient = getZkClient();
		// Specify the counter storage path   And retry policy 
        DistributedAtomicInteger distributedAtomicInteger = new DistributedAtomicInteger(zkClient, "/counter", new ExponentialBackoffRetry(1000, 3));
        // Multithread autoincrement 10*100 Time 
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                for (int j = 0; j < 100; j++) {
                    try {
                    	// call add Methods increase by themselves 
                        AtomicValue<Integer> result = distributedAtomicInteger.add(1);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                countDownLatch.countDown();

            }).start();
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // View results 
        System.out.println(" Multithread auto increment result " + distributedAtomicInteger.get().postValue());
    }

    private static CuratorFramework getZkClient() {
        String zkServerAddress = "172.16.158.11:2181,"
            + "172.16.158.12:2181,"
            + "172.16.158.13:2181";
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000);
        CuratorFramework zkClient = CuratorFrameworkFactory.builder()
                .connectString(zkServerAddress)
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(5000)
                .retryPolicy(retryPolicy)
                .build();
        zkClient.start();
        return zkClient;
    }

}

According to the principle that , Since the end of the increase , Should the result be 10*100？, After running for several times, Taoist friends will find that the actual values are not all 1000. Why? ？ This thing is fake ？
Let's take a look at AtomicValue result = distributedAtomicInteger.add(1) In this line of code ,add() Method source code

/**
     * Add delta to the current value and return the new value information. Remember to always
     * check {@link AtomicValue#succeeded()}.
     *
     * @param delta amount to add
     * @return value info
     * @throws Exception ZooKeeper errors
     */
    @Override
    public AtomicValue<Integer>    add(Integer delta) throws Exception
    {
        return worker(delta);
    }

It reads Remember to always check {@link AtomicValue#succeeded()}. in other words , The self - augmentation of this method is not necessarily successful , When initializing the distributed machine count object earlier , A retry policy was passed in , If concurrency and self increment occur in a distributed environment , Will try again and again , If the retry fails , The result returns failed