利用curator实现的zookeeper分布式锁服务

╰半橙微兮° 2022-08-08 04:30 5阅读 0赞

Curator是Netflix开源的一套ZooKeeper客户端框架，用它来操作zookeeper更加简单方便，按Curator官方所比喻的，guava to JAVA， Curator to Zookeeper，Curator采用了fluent风格的代码，非常简洁。     
    主要看下，使用curator操作zookeeper的一些基础例子： 
    主要的功能： 
    1，在zk上添加，或更新数据 
    2，删除zk节点上数据 
    3，读取某个节点上的数据 
    4，上传一些本地文件到zk节点上 
    5，检查zookeeper上是否存在某个节点路径 
    核心代码如下： 
     package com.qin.curator.zk;
    import java.io.File;
    import java.util.List;
    import org.apache.commons.io.FileUtils;
    import org.apache.curator.CuratorZookeeperClient;
    import org.apache.curator.RetryPolicy;
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
    import org.apache.curator.framework.api.CreateBuilder;
    import org.apache.curator.retry.ExponentialBackoffRetry;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZKUtil;
    import framework.CrudExamples;
    /**
     * @author qindongliang
     * curator操作zookeeper的
     * 基本例子
     * **/
    public class CuratorTools {
    	static CuratorFramework zkclient=null;
    	static String nameSpace="php";
    	static {
    		  String zkhost="192.168.46.22:2181";//zk的host
    		  RetryPolicy rp=new ExponentialBackoffRetry(1000, 3);//重试机制
    		  Builder builder = CuratorFrameworkFactory.builder().connectString(zkhost)
    				  .connectionTimeoutMs(5000)
    				  .sessionTimeoutMs(5000)
    				  .retryPolicy(rp);
    		  builder.namespace(nameSpace);
    		  CuratorFramework zclient = builder.build();
    		  zkclient=zclient;
    		  zkclient.start();// 放在这前面执行
    		  zkclient.newNamespaceAwareEnsurePath(nameSpace);
    	}
    	public static void main(String[] args)throws Exception {
    		CuratorTools ct=new  CuratorTools();
    		//ct.getListChildren("/zk/bb");
    		//ct.upload("/jianli/123.txt", "D:\\123.txt");
    		//ct.createrOrUpdate("/zk/cc334/zzz","c");
    		//ct.delete("/qinb/bb");
    		//ct.checkExist("/zk");
    		ct.read("/jianli/123.txt");
    		zkclient.close();
    	}
    	/**
    	 * 创建或更新一个节点
    	 * 
    	 * @param path 路径
    	 * @param content 内容
    	 * **/
    	public void createrOrUpdate(String path,String content)throws Exception{
    		zkclient.newNamespaceAwareEnsurePath(path).ensure(zkclient.getZookeeperClient());
    	    zkclient.setData().forPath(path,content.getBytes());	
    	    System.out.println("添加成功！！！");
    	}
    	/**
    	 * 删除zk节点
    	 * @param path 删除节点的路径
    	 * 
    	 * **/
    	public void delete(String path)throws Exception{
    		zkclient.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
    		System.out.println("删除成功!");
    	}
    	/**
    	 * 判断路径是否存在
    	 * @param path
    	 * **/
    	public void checkExist(String path)throws Exception{
    		if(zkclient.checkExists().forPath(path)==null){
    			System.out.println("路径不存在!");
    		}else{
    			System.out.println("路径已经存在!");
    		}
    	}
    	/**
    	 * 读取的路径
    	 * @param path
    	 * **/
    	public void read(String path)throws Exception{
    		String data=new String(zkclient.getData().forPath(path),"gbk");
    		System.out.println("读取的数据:" data);
    	}
    	/**
    	 * @param path 路径
    	 * 获取某个节点下的所有子文件
    	 * */
    	public void getListChildren(String path)throws Exception{
    		List<String> paths=zkclient.getChildren().forPath(path);
    		for(String p:paths){
    			System.out.println(p);
    		}
    	}
    	/**
    	 * @param zkPath zk上的路径
    	 * @param localpath 本地上的文件路径
    	 * 
    	 * **/
    	public void upload(String zkPath,String localpath)throws Exception{
    		createrOrUpdate(zkPath, "");//创建路径
    		byte[] bs=FileUtils.readFileToByteArray(new File(localpath));
    		zkclient.setData().forPath(zkPath, bs);
    		System.out.println("上传文件成功！");
    	}
    }
     import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    import com.netflix.curator.RetryPolicy;
    import com.netflix.curator.framework.CuratorFramework;
    import com.netflix.curator.framework.CuratorFrameworkFactory;
    import com.netflix.curator.framework.recipes.locks.InterProcessMutex;
    import com.netflix.curator.retry.ExponentialBackoffRetry;
    public class TestCuratorLock {
     /**
      * @param args
      * @throws InterruptedException
      */
     public static void main(String[] args) throws InterruptedException {
      // TODO Auto-generated method stub
      CountDownLatch latch = new CountDownLatch(5);
      String zookeeperConnectionString = "localhost:2181,localhost:2182,localhost:2183";
      RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
      CuratorFramework client = CuratorFrameworkFactory.newClient(
        zookeeperConnectionString, retryPolicy);
      client.start();
      System.out.println("客户端启动。。。。");
      ExecutorService exec = Executors.newCachedThreadPool();
      for (int i = 0; i < 5; i++) {
       exec.submit(new MyLock("client" + i, client, latch));
      }
      exec.shutdown();
      latch.await();
      System.out.println("所有任务执行完毕");
      client.close();
      System.out.println("客户端关闭。。。。");
     }
     static class MyLock implements Runnable {
      private String name;
      private CuratorFramework client;
      private CountDownLatch latch;
      public MyLock(String name, CuratorFramework client, CountDownLatch latch) {
       this.name = name;
       this.client = client;
       this.latch = latch;
      }
      public String getName() {
       return name;
      }
      public void setName(String name) {
       this.name = name;
      }
      @Override
      public void run() {
       // TODO Auto-generated method stub
       InterProcessMutex lock = new InterProcessMutex(client,
         "/test_group");
       try {
        if (lock.acquire(120, TimeUnit.SECONDS)) {
         try {
          // do some work inside of the critical section here
          System.out.println("----------" + this.name
            + "获得资源----------");
          System.out.println("----------" + this.name
            + "正在处理资源----------");
          Thread.sleep(10 * 1000);
          System.out.println("----------" + this.name
            + "资源使用完毕----------");
          latch.countDown();
         } finally {
          lock.release();
          System.out.println("----------" + this.name
            + "释放----------");
         }
        }
       } catch (Exception e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
       }
       }
     }
     }使用curator实现zookeeper锁服务的示例分享

import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    import com.netflix.curator.RetryPolicy;
    import com.netflix.curator.framework.CuratorFramework;
    import com.netflix.curator.framework.CuratorFrameworkFactory;
    import com.netflix.curator.framework.recipes.locks.InterProcessMutex;
    import com.netflix.curator.retry.ExponentialBackoffRetry;
    public class TestCuratorLock {
     /**
      * @param args
      * @throws InterruptedException
      */
     public static void main(String[] args) throws InterruptedException {
      // TODO Auto-generated method stub
      CountDownLatch latch = new CountDownLatch(5);
      String zookeeperConnectionString = "localhost:2181,localhost:2182,localhost:2183";
      RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
      CuratorFramework client = CuratorFrameworkFactory.newClient(
        zookeeperConnectionString, retryPolicy);
      client.start();
      System.out.println("客户端启动。。。。");
      ExecutorService exec = Executors.newCachedThreadPool();
      for (int i = 0; i < 5; i++) {
       exec.submit(new MyLock("client" + i, client, latch));
      }
      exec.shutdown();
      latch.await();
      System.out.println("所有任务执行完毕");
      client.close();
      System.out.println("客户端关闭。。。。");
     }
     static class MyLock implements Runnable {
      private String name;
      private CuratorFramework client;
      private CountDownLatch latch;
      public MyLock(String name, CuratorFramework client, CountDownLatch latch) {
       this.name = name;
       this.client = client;
       this.latch = latch;
      }
      public String getName() {
       return name;
      }
      public void setName(String name) {
       this.name = name;
      }
      @Override
      public void run() {
       // TODO Auto-generated method stub
       InterProcessMutex lock = new InterProcessMutex(client,
         "/test_group");
       try {
        if (lock.acquire(120, TimeUnit.SECONDS)) {
         try {
          // do some work inside of the critical section here
          System.out.println("----------" + this.name
            + "获得资源----------");
          System.out.println("----------" + this.name
            + "正在处理资源----------");
          Thread.sleep(10 * 1000);
          System.out.println("----------" + this.name
            + "资源使用完毕----------");
          latch.countDown();
         } finally {
          lock.release();
          System.out.println("----------" + this.name
            + "释放----------");
         }
        }
       } catch (Exception e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
       }
       }
     }
     }