Zookeeper实现分布式锁

package com.web.util;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs.Ids;

/**
 * 用于跨系统或者跨服务器之间的锁，不能用于同一个虚拟机
 * @author Administrator
 *
 */
public class LockUtil {
    /**
     * zk 客户端
     */
    private static CuratorFramework client = null;
    
    private static Logger logger = Logger.getLogger(LockUtil.class);
    
    /**
     * CountDownLatch这个类能够使一个线程等待其他线程完成各自的工作后再执行
     * 排他锁  阻塞：资源控制实现 
     */
    protected static CountDownLatch latch = new CountDownLatch(1);
    
    protected static CountDownLatch shardLocklatch = new CountDownLatch(1);
    
    /**
     * 谁当前调用获取锁的，实际生产中代表IP地址
     */
    private static String selfIdentity = null;
    
    private static String selfNodeName = null;
    
    public static synchronized void init(String connectString) {
        if (client != null)
            return;
        
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
            .connectString(connectString)
            .sessionTimeoutMs(10000)
            .retryPolicy(retryPolicy)
            .namespace("LockService")
            .build();
        client.start();
        
        // 创建锁目录
        try {
            if (client.checkExists().forPath("/ExclusiveLockDemo") == null) {
                client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.PERSISTENT)
                    .withACL(Ids.OPEN_ACL_UNSAFE)
                    .forPath("/ExclusiveLockDemo");
            }
            // 创建锁监听
            addChildWatcher("/ExclusiveLockDemo");
            if (client.checkExists().forPath("/ShardLockDemo") == null) {
                client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.PERSISTENT)
                    .withACL(Ids.OPEN_ACL_UNSAFE)
                    .forPath("/ShardLockDemo");
            }
        }
        catch (Exception e) {
            logger.error("ZK服务器连接不上");
            throw new RuntimeException("ZK服务器连接不上");
        }
    }
    /**
     * 获取排他锁
     * <功能详细描述>
     */
    public static synchronized void getExclusiveLock() {
        while (true) {
            try {
                client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).withACL(Ids.OPEN_ACL_UNSAFE).forPath(
                    "/ExclusiveLockDemo/lock");
                logger.info("成功获取到锁");
                return;// 如果节点创建成功，即说明获取锁成功
            }
            catch (Exception e) {
                logger.info("此次获取锁没有成功");
                try {
                    //如果没有获取到锁，需要重新设置同步资源值
                    if (latch.getCount() <= 0) {
                        latch = new CountDownLatch(1);
                    }
                    latch.await();
                }
                catch (InterruptedException e1) {
                    e1.printStackTrace();
                    logger.error("", e1);
                }
            }
        }
    }
    
    /**
     * 
     * @param type
     *            0为读锁，1为写锁
     * @param identity
     *            获取当前锁的所有者
     */
    public static boolean getShardLock(int type, String identity) {
        if (identity == null || "".equals(identity)) {
            throw new RuntimeException("identity不能为空");
        }
        if (identity.indexOf("-") != -1) {
            throw new RuntimeException("identity不能包含字符-");
        }
        if (type != 0 && type != 1) {
            throw new RuntimeException("type只能为0或者1");
        }
        String nodeName = null;
        if (type == 0) {
            nodeName = "R" + identity + "-";
        }
        else if (type == 1) {
            nodeName = "W" + identity + "-";
        }
        selfIdentity = nodeName;
        try {
            //if (client.checkExists().forPath("/ShardLockDemo/" + nodeName) == null)
            selfNodeName =
                client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).withACL(Ids.OPEN_ACL_UNSAFE).forPath(
                    "/ShardLockDemo/" + nodeName);
            logger.info("创建节点:" + selfNodeName);
            List<String> lockChildrens = client.getChildren().forPath("/ShardLockDemo");
            if (!canGetLock(lockChildrens, type, nodeName.substring(0, nodeName.length() - 1), false)) {
                shardLocklatch.await();
            }
            // return;// 获得锁成功就返回
        }
        catch (Exception e) {
            logger.info("出现异常", e);
            return false;
        }
        
        logger.info("成功获取锁");
        return true;
    }
    /**
     * java -jar lock.jar 0 192.168.1.1
     * <功能详细描述>
     * @param childrens
     * @param type
     * @param identity
     * @param reps
     * @return
     */
    private static boolean canGetLock(List<String> childrens, int type, String identity, boolean reps) {
        boolean res = false;
        if (childrens.size() <= 0)
            return true;
        
        try {
            String currentSeq = null;
            List<String> seqs = new ArrayList<String>();
            //List<String> identitys = new ArrayList<String>();
            Map<String, String> seqs_identitys = new HashMap<String, String>();
            for (String child : childrens) {
                String splits[] = child.split("-");
                seqs.add(splits[1]);
                //identitys.add(splits[0]);
                seqs_identitys.put(splits[1], splits[0]);
                if (identity.equals(splits[0]))
                    currentSeq = splits[1];
            }
            
            List<String> sortSeqs = new ArrayList<String>();
            sortSeqs.addAll(seqs);
            Collections.sort(sortSeqs);
            
            // 第一个节点，则无论是读锁还是写锁都可以获取
            if (currentSeq.equals(sortSeqs.get(0))) {
                res = true;
                logger.info("请求锁,因为是第一个请求锁的请求，所以获取成功");
                return res;
            }
            else {
                // 写锁，不是第一个就一定会失败
                if (type == 1) {
                    res = false;
                    //第一次请求取锁则设置监听，以后就不设置了，因为监听一直存在
                    //比如：前面有两个读锁，第三个为写锁，watch通知会多了一遍，当前面的读锁释放以后，写锁会去尝试，因为前面还有一个读锁，所以获取不到，然后就会重新注册watch，为导致通知两遍
                    if (reps == false)
                        addChildWatcher("/ShardLockDemo");
                    logger.info("请求写锁，因为前面有其它锁，所以获取锁失败");
                    return res;
                }
            }
            // int index =-1;
            //因为不是第一个节点，所以要判断在当前节点之前有没有写的节点存在,如果存在写锁则不能获取读锁
            boolean hasW = true;
            for (String seq : sortSeqs) {
                // ++index;
                if (seq.equals(currentSeq)) {
                    break;
                }
                if (!seqs_identitys.get(seq).startsWith("W"))
                    hasW = false;
            }
            if (type == 0 && hasW == false) {
                res = true;
            }
            else if (type == 0 && hasW == true) {
                res = false;
            }
            if (res == false) {
                // 添加监听,锁释放之后需要进行抢锁
                addChildWatcher("/ShardLockDemo");
                logger.info("因为没有获取到锁，添加锁的监听器");
            }
        }
        catch (Exception e) {
            e.printStackTrace();
        }
        return res;
    }
    /**
     * 能删除的节点尽量删除，不然会有延迟
     * <功能详细描述>
     * @return
     */
    public static boolean unlockForExclusive() {
        try {
            if (client.checkExists().forPath("/ExclusiveLockDemo/lock") != null) {
                client.delete().forPath("/ExclusiveLockDemo/lock");
            }
        }
        catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }
    
    public static boolean unlockForShardLock() {
        try {
            if (client.checkExists().forPath(selfNodeName) != null) {
                client.delete().forPath(selfNodeName);
            }
        }
        catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }
    /**
     * 添加子节点的监听
     * @param path
     * @throws Exception
     */
    public static void addChildWatcher(String path) throws Exception {
        @SuppressWarnings("resource")
        final PathChildrenCache cache = new PathChildrenCache(client, path, true);
        cache.start(StartMode.POST_INITIALIZED_EVENT);// ppt中需要讲StartMode
        // System.out.println(cache.getCurrentData().size());
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                if (event.getType().equals(PathChildrenCacheEvent.Type.INITIALIZED)) {
                    
                }
                else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)) {
                    
                }
                else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                    String path = event.getData().getPath();
                    System.out.println("收到监听" + path);
                    if (path.contains("ExclusiveLockDemo")) {
                        logger.info("排他锁,收到锁释放通知");
                        //CountDownLatch(1)，此时修改为0 latch.wait()释放
                        latch.countDown();
                    }
                    else if (path.contains("ShardLockDemo")) {
                        logger.info("共享锁,收到锁释放通知");
                        //收到自己的通知就不处理
                        if (path.contains(selfIdentity))
                            return;
                        List<String> lockChildrens = client.getChildren().forPath("/ShardLockDemo");
                        boolean isLock = false;
                        try {
                            if (selfIdentity.startsWith("R"))
                                isLock = canGetLock(lockChildrens, 0, selfIdentity.substring(0, selfIdentity.length() - 1), true);
                            else if (selfIdentity.startsWith("W"))
                                isLock = canGetLock(lockChildrens, 1, selfIdentity.substring(0, selfIdentity.length() - 1), true);
                        }
                        catch (Exception e) {
                            e.printStackTrace();
                        }
                        logger.info("收到锁释放监听后，重新尝试获取锁，结果为:" + isLock);
                        if (isLock) {
                            //获得锁
                            logger.info("获得锁，解除因为获取不到锁的阻塞");
                            shardLocklatch.countDown();
                        }
                    }
                }
                else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {
                    
                }
            }
        });
    }
}

package com.web;

import java.util.Calendar;
import java.util.Date;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.web.service.OrderRepository;
import com.web.service.OrderService;
import com.web.service.ProductRepository;
import com.web.service.pojo.Order;
import com.web.service.pojo.Product;
import com.web.util.LockUtil;

public class Test {
    private static ApplicationContext ctx;  
    
    static 
    {  
        ctx = new ClassPathXmlApplicationContext("applicationContext.xml");  
    }        
    
    /**
     * 测试超卖问题
     * <功能详细描述>
     */
    public static void testpurchase(){
        OrderService os = (OrderService)ctx.getBean("orderService"); 
        
        Order order = new Order();
        order.setProductId(1L);
        order.setCreateTime(new Date());
        order.setPnum(1);
        os.doOrder(order);
    }
    
    public static void testMapper(){
        ProductRepository mapper = (ProductRepository)ctx.getBean("productMapper"); 
        //测试id=1的用户查询，根据数据库中的情况，可以改成你自己的.
        System.out.println("得到用户id=1的用户信息");
        Product product = mapper.selectProductById(1L);
        System.out.println(product.getName()); 
        
        OrderRepository omapper = (OrderRepository)ctx.getBean("orderMapper"); 
        Order order = new Order();
        order.setProductId(1L);
        order.setCreateTime(new Date());
        order.setPnum(1);
        omapper.saveOrder(order);       
    }
    
    public static void testShardLog(int type,String identity){
        System.out.println("---------------开始获取锁"+identity);
        LockUtil.init("localhost:2181");
        LockUtil.getShardLock(type, identity);
        System.out.println("---------------获取锁结束"+identity);
    }
    public static void main(String[] args)  
    {  

        int type = Integer.parseInt(args[0]);//0读锁还是1写锁
        System.out.println("type="+type);//线程名称
        testShardLog(type,args[1]);
        //testShardLog(0,"f6");
//        try {
//            LockUtil.init("localhost:2181");
//            //LockUtil.addChildWatcher("/LockService");
//            LockUtil.getExclusiveLock();
//        } catch (Exception e1) {
//            // TODO Auto-generated catch block
//            e1.printStackTrace();
//        }
        
        //排他锁***************************超卖测试***************************
        
//        long nowtime = Calendar.getInstance().getTimeInMillis();
//        System.out.println("begin with "+nowtime);
//        testpurchase();
//        System.out.println("end with "+nowtime);
        
         //end排他锁***************************超卖测试***************************
        
        
    //    String []names = ctx.getBeanDefinitionNames();
//        for(String s: names){
//            System.out.println(s);
//        }
        
        try {
            Thread.sleep(20000);
            System.out.println("-------------开始--释放锁");
            LockUtil.unlockForShardLock();
            System.out.println("-------------成功--释放锁");
            Thread.sleep(30000);
            System.out.println("---------------结束退出");
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    } 
}