Curator场景应用

Posted 2020-11-09 小~虎

分布式锁功能：

在分布式场景中，我们为了保证数据的一致性，经常在程序运行的某一个点，需要进行同步操作，(java提供synchronized或者Reentrantlock实现)，

使用curator基于zookeeper的特性提供的分布式锁来处理分布式场景的数据一致性，zookeeper本身的分布式是有写问题的，这里推荐使用Curator框架的

InterProcessMutex来实现。 

 1 package bjsxt.curator.lock;
 2 
 3 import java.text.SimpleDateFormat;
 4 import java.util.Date;
 5 import java.util.concurrent.CountDownLatch;
 6 
 7 import org.apache.curator.RetryPolicy;
 8 import org.apache.curator.framework.CuratorFramework;
 9 import org.apache.curator.framework.CuratorFrameworkFactory;
10 import org.apache.curator.framework.recipes.locks.InterProcessMutex;
11 import org.apache.curator.retry.ExponentialBackoffRetry;
12 
13 public class Lock2 {
14 
15     /** zookeeper地址 */
16     static final String CONNECT_ADDR = "192.168.2.2:2181";
17     /** session超时时间 */
18     static final int SESSION_OUTTIME = 5000;// ms
19 
20     static int count = 10;
21 
22     public static void genarNo() {
23         try {
24             count--;
25             System.out.println(count);
26         } finally {
27 
28         }
29     }
30 
31     public static void main(String[] args) throws Exception {
32 
33         // 1 重试策略：初试时间为1s 重试10次
34         RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
35         // 2 通过工厂创建连接
36         CuratorFramework cf = CuratorFrameworkFactory.builder()
37                 .connectString(CONNECT_ADDR).sessionTimeoutMs(SESSION_OUTTIME)
38                 .retryPolicy(retryPolicy)
39                 // .namespace("super")
40                 .build();
41         // 3 开启连接
42         cf.start();
43 
44         // 4 分布式锁
45         final InterProcessMutex lock = new InterProcessMutex(cf, "/super");
46         // final ReentrantLock reentrantLock = new ReentrantLock();
47         final CountDownLatch countdown = new CountDownLatch(1);
48 
49         for (int i = 0; i < 10; i++) {
50             new Thread(new Runnable() {
51                 @Override
52                 public void run() {
53                     try {
54                         countdown.await();
55                         // 加锁
56                         lock.acquire();
57                         // reentrantLock.lock();
58                         // -------------业务处理开始
59                         // genarNo();
60                         SimpleDateFormat sdf = new SimpleDateFormat(
61                                 "HH:mm:ss|SSS");
62                         System.out.println(sdf.format(new Date()));
63                         // System.out.println(System.currentTimeMillis());
64                         // -------------业务处理结束
65                     } catch (Exception e) {
66                         e.printStackTrace();
67                     } finally {
68                         try {
69                             // 释放
70                             lock.release();
71                             // reentrantLock.unlock();
72                         } catch (Exception e) {
73                             e.printStackTrace();
74                         }
75                     }
76                 }
77             }, "t" + i).start();
78         }
79         Thread.sleep(100);
80         countdown.countDown();
81 
82     }
83 }

 

分布式计数器功能

分布式计数器，在单JVM中，我们可以通过AtomicInteger这种经典的方式实现，但是在分布式的场景下，就需要利用Curator框架的DistributedAtomicInteger来实现

 1 package bjsxt.curator.atomicinteger;
 2 
 3 import org.apache.curator.RetryPolicy;
 4 import org.apache.curator.framework.CuratorFramework;
 5 import org.apache.curator.framework.CuratorFrameworkFactory;
 6 import org.apache.curator.framework.recipes.atomic.AtomicValue;
 7 import org.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;
 8 import org.apache.curator.retry.ExponentialBackoffRetry;
 9 import org.apache.curator.retry.RetryNTimes;
10 
11 public class CuratorAtomicInteger {
12 
13     /** zookeeper地址 */
14     static final String CONNECT_ADDR = "192.168.2.2:2181";
15     /** session超时时间 */
16     static final int SESSION_OUTTIME = 5000;// ms
17 
18     public static void main(String[] args) throws Exception {
19 
20         // 1 重试策略：初试时间为1s 重试10次
21         RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
22         // 2 通过工厂创建连接
23         CuratorFramework cf = CuratorFrameworkFactory.builder()
24                 .connectString(CONNECT_ADDR).sessionTimeoutMs(SESSION_OUTTIME)
25                 .retryPolicy(retryPolicy).build();
26         // 3 开启连接
27         cf.start();
28         // cf.delete().forPath("/super");
29 
30         // 4 使用DistributedAtomicInteger
31         DistributedAtomicInteger atomicIntger = new DistributedAtomicInteger(
32                 cf, "/super", new RetryNTimes(3, 1000));
33 
34         AtomicValue<Integer> value = atomicIntger.add(1);
35         System.out.println(value.succeeded());
36         System.out.println(value.postValue()); // 最新值
37         System.out.println(value.preValue()); // 原始值
38 
39     }
40 }

Curator框架，让一些很困难的问题，简单化了