专栏首页IT技术精选文摘跟着实例学习ZooKeeper的用法: 计数器

跟着实例学习ZooKeeper的用法: 计数器

这一篇文章我们将学习使用Curator来实现计数器。 顾名思义,计数器是用来计数的, 利用ZooKeeper可以实现一个集群共享的计数器。 只要使用相同的path就可以得到最新的计数器值, 这是由ZooKeeper的一致性保证的。Curator有两个计数器, 一个是用int来计数,一个用long来计数。

SharedCount

这个类使用int类型来计数。 主要涉及三个类。

  • SharedCount
  • SharedCountReader
  • SharedCountListener

SharedCount代表计数器, 可以为它增加一个SharedCountListener,当计数器改变时此Listener可以监听到改变的事件,而SharedCountReader可以读取到最新的值, 包括字面值和带版本信息的值VersionedValue。

例子代码:

package com.colobu.zkrecipe.counter;import java.io.IOException;import java.util.List;import java.util.Random;import java.util.concurrent.Callable;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.framework.recipes.shared.SharedCount;import org.apache.curator.framework.recipes.shared.SharedCountListener;import org.apache.curator.framework.recipes.shared.SharedCountReader;import org.apache.curator.framework.state.ConnectionState;import org.apache.curator.retry.ExponentialBackoffRetry;import org.apache.curator.test.TestingServer;import com.google.common.collect.Lists;public class SharedCounterExample implements SharedCountListener{    private static final int QTY = 5;    private static final String PATH = "/examples/counter";    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();
        }


    }    @Override
    public void stateChanged(CuratorFramework arg0, ConnectionState arg1) {
        System.out.println("State changed: " + arg1.toString());
    }    @Override
    public void countHasChanged(SharedCountReader sharedCount, int newCount) throws Exception {
        System.out.println("Counter's value is changed to " + newCount);        
    }

}

在这个例子中,我们使用baseCount来监听计数值(addListener方法)。 任意的SharedCount, 只要使用相同的path,都可以得到这个计数值。 然后我们使用5个线程为计数值增加一个10以内的随机数。

count.trySetCount(count.getVersionedValue(), count.getCount() + rand.nextInt(10))

这里我们使用trySetCount去设置计数器。 第一个参数提供当前的VersionedValue,如果期间其它client更新了此计数值, 你的更新可能不成功, 但是这时你的client更新了最新的值,所以失败了你可以尝试再更新一次。 而setCount是强制更新计数器的值。

注意计数器必须start,使用完之后必须调用close关闭它。

在这里再重复一遍前面讲到的, 强烈推荐你监控ConnectionStateListener, 尽管我们的有些例子没有监控它。 在本例中SharedCountListener扩展了ConnectionStateListener。 这一条针对所有的Curator recipes都适用,后面的文章中就不专门提示了。

DistributedAtomicLong

再看一个Long类型的计数器。 除了计数的范围比SharedCount大了之外, 它首先尝试使用乐观锁的方式设置计数器, 如果不成功(比如期间计数器已经被其它client更新了), 它使用InterProcessMutex方式来更新计数值。 还记得InterProcessMutex是什么吗? 它是我们前面跟着实例学习ZooKeeper的用法: 分布式锁 讲的分布式可重入锁。 这和上面的计数器的实现有显著的不同。

可以从它的内部实现DistributedAtomicValue.trySet中看出端倪。

    AtomicValue<byte[]>   trySet(MakeValue makeValue) throws Exception
    {
        MutableAtomicValue<byte[]>  result = new MutableAtomicValue<byte[]>(null, null, false);

        tryOptimistic(result, makeValue);        if ( !result.succeeded() && (mutex != null) )
        {
            tryWithMutex(result, makeValue);
        }        return result;
    }

此计数器有一系列的操作:

  • get(): 获取当前值
  • increment(): 加一
  • decrement(): 减一
  • add(): 增加特定的值
  • subtract(): 减去特定的值
  • trySet(): 尝试设置计数值
  • forceSet(): 强制设置计数值

必须检查返回结果的succeeded(), 它代表此操作是否成功。 如果操作成功, preValue()代表操作前的值, postValue()代表操作后的值。

我们下面的例子中使用5个线程对计数器进行加一操作,如果成功,将操作前后的值打印出来。

package com.colobu.zkrecipe.counter;import java.io.IOException;import java.util.List;import java.util.concurrent.Callable;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;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.curator.test.TestingServer;import com.google.common.collect.Lists;public class DistributedAtomicLongExample {    private static final int QTY = 5;    private static final String PATH = "/examples/counter";    public static void main(String[] args) throws IOException, Exception {        try (TestingServer server = new TestingServer()) {
            CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));
            client.start();

            List<DistributedAtomicLong> examples = Lists.newArrayList();
            ExecutorService service = Executors.newFixedThreadPool(QTY);            for (int i = 0; i < QTY; ++i) {                final DistributedAtomicLong count = new DistributedAtomicLong(client, PATH, new RetryNTimes(10, 10));

                examples.add(count);
                Callable<Void> task = new Callable<Void>() {                    @Override
                    public Void call() throws Exception {                        try {                            //Thread.sleep(rand.nextInt(1000));
                            AtomicValue<Long> value = count.increment();                            //AtomicValue<Long> value = count.decrement();
                            //AtomicValue<Long> value = count.add((long)rand.nextInt(20));
                            System.out.println("succeed: " + value.succeeded());                            if (value.succeeded())
                                System.out.println("Increment: from " + value.preValue() + " to " + value.postValue());
                        } catch (Exception e) {
                            e.printStackTrace();
                        }                        return null;
                    }
                };
                service.submit(task);
            }

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

    }

}

本文分享自微信公众号 - IT技术精选文摘(ITHK01),作者:ifeve

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2017-11-21

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Spring Boot使用过滤器和拦截器分别实现REST接口简易安全认证

    用户1263954
  • 跟着实例学习ZooKeeper的用法: 队列

    使用Curator也可以简化Ephemeral Node (临时节点)的操作。Curator也提供ZK Recipe的分布式队列实现。 利用ZK的 PERSIS...

    用户1263954
  • 协同过滤推荐算法Java代码实现

    什么是协同过滤 协同过滤是利用集体智慧的一个典型方法。要理解什么是协同过滤 (Collaborative Filtering, 简称 CF),首先想一个简单的问...

    用户1263954
  • 【tomcat】HTTPS访问配置 + restful调用远程HTTPS绕过验证

    由于是本地测试,请修改本地host文件C:\Windows\System32\drivers\etc\host

    肖哥哥
  • Eclipse连接Hadoop集群和WordCount实战

    本文将主要介绍Eclipse连接Hadoop集群和WordCount实践项目两大内容。

    挖掘大数据
  • Hadoop阅读笔记(一)——强大的MapReduce

    前言:来园子已经有8个月了,当初入园凭着满腔热血和一脑门子冲动,给自己起了个响亮的旗号“大数据 小世界”,顿时有了种世界都是我的,世界都在我手中的赶脚。可是.....

    JackieZheng
  • 单点登录SSO解决方案之SpringSecurity+JWT实现

      通过前面几天文章我们详细的介绍了SpringSecurity的使用,本文我们来看下,结合JWT来实现单点登录操作。

    用户4919348
  • HttpClient 发送Json

    shengjk1
  • [天池比赛] Multimodalities Recall

    https://tianchi.aliyun.com/competition/entrance/231786/information

    MachineLP
  • 性能工具之Ngrinder源码部署

    https://github.com/naver/ngrinder/releases

    高楼Zee

扫码关注云+社区

领取腾讯云代金券