zookeeper分布式集群Curator的分布式long型计数器DistributedAtomicLong

不念不忘少年蓝@ 2023-06-03 05:51 25阅读 0赞

# zookeeper分布式集群Curator的分布式long型计数器DistributedAtomicLong #

同样作为分布式计数器，DistributedAtomicLong比SharedCount更简单易用。

import org.apache.curator.RetryPolicy;
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.framework.recipes.atomic.AtomicValue;
    import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
    import org.apache.curator.retry.ExponentialBackoffRetry;
    import org.apache.curator.retry.RetryNTimes;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.data.Stat;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    
    public class Main {
        private int COUNT = 10;
        private int TIME = 50;
    
        public static void main(String[] args) {
            //初始化log4j，zookeeper否则报错。
            //org.apache.log4j.BasicConfigurator.configure();
    
            try {
                Main app = new Main();
                app.zk();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        private void zk() throws Exception {
            //任意位置的SharedCount，只要使用相同的path，都可以得到这个计数值。
            String path = "/path/count";
    
            CuratorFramework client = initClient(path);
    
            //分布式集群中存在的某一处计数。
            //DistributedAtomicLong baseCount = new DistributedAtomicLong(client, path, new ExponentialBackoffRetry(1000, 3));
            //baseCount.trySet(0L);
    
            //开启COUNT个线程，模拟对分布式中SharedCount的不同节点的赋值操作。
            ExecutorService service = Executors.newFixedThreadPool(COUNT);
            for (int i = 0; i < COUNT; i++) {
                DistributedAtomicLong count = new DistributedAtomicLong(client, path, new RetryNTimes(10, 10));
    
                Runnable runnable = new Runnable() {
                    @Override
                    public void run() {
                        try {
                            TimeUnit.SECONDS.sleep((long) (Math.random() * TIME));
    
                            setValue(count);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                };
    
                service.submit(runnable);
            }
    
            service.shutdown();
            service.awaitTermination(TIME, TimeUnit.SECONDS);
        }
    
        private synchronized void setValue(DistributedAtomicLong count) throws Exception {
            System.out.println("-------------");
    
            System.out.println("当前值=" + count.get().preValue());
    
            long l = (long) (Math.random() * 1000);
            System.out.println("尝试赋值:" + l);
    
            AtomicValue<Long> result = count.trySet(l);
            if (result.succeeded()) {
                System.out.println(result.postValue() + "设置成功");
                System.out.println(result.preValue() + "->" + result.postValue());
            } else {
                System.out.println(result.postValue() + "设置失败");
                System.out.println("计数器仍是旧值:" + result.preValue());
            }
        }
    
        private CuratorFramework initClient(String path) throws Exception {
            CuratorFramework client = makeClient();
            client.start();
    
            boolean b = isPathExist(client, path);
    
            //如果不存在这个路径，stat为null，创建新的节点路径。
            if (!b) {
                String s = client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .forPath(path);
    
                System.out.println("创建 " + s);
            } else {
                System.out.println("已存在:" + path + ",不需重复创建");
            }
    
            return client;
        }
    
        //检测是否存在该路径。
        private boolean isPathExist(CuratorFramework client, String path) {
            boolean b = false;
    
            //检测是否存在该路径。
            try {
                Stat stat = client.checkExists().forPath(path);
                b = stat == null ? false : true;
            } catch (Exception e) {
                e.printStackTrace();
            }
    
            return b;
        }
    
        private CuratorFramework makeClient() {
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
            CuratorFramework client = CuratorFrameworkFactory.builder()
                    .connectString(getAddress())
                    .sessionTimeoutMs(10 * 1000)
                    .connectionTimeoutMs(20 * 1000)
                    .retryPolicy(retryPolicy)
                    .build();
    
            return client;
        }
    
        private String getAddress() {
            String ip = "127.0.0.1";
            return ip + ":2181," + ip + ":2182," + ip + ":2183";
        }
    }

输出：

创建 /path/count
    -------------
    当前值=3546358405553139250
    尝试赋值:385
    385设置成功
    3546358405553139250->385
    -------------
    当前值=385
    尝试赋值:161
    161设置成功
    385->161
    -------------
    当前值=161
    尝试赋值:672
    672设置成功
    161->672
    -------------
    当前值=672
    尝试赋值:299
    299设置成功
    672->299
    -------------
    当前值=299
    尝试赋值:98
    98设置成功
    299->98
    -------------
    当前值=98
    尝试赋值:53
    53设置成功
    98->53
    -------------
    当前值=53
    尝试赋值:549
    549设置成功
    53->549
    -------------
    当前值=549
    尝试赋值:869
    869设置成功
    549->869
    -------------
    当前值=869
    尝试赋值:222
    222设置成功
    869->222
    -------------
    当前值=222
    尝试赋值:522
    522设置成功
    222->522