码迷，mamicode.com
首页 > 其他好文 > 详细

基于zookeeper简单实现分布式锁

时间：2014-11-19 14:09:43      阅读：171      评论：0      收藏：0      [点我收藏+]

标签：des   style   blog   io   ar   os   sp   java   for   

这里利用zookeeper的EPHEMERAL_SEQUENTIAL类型节点及watcher机制，来简单实现分布式锁。
主要思想：
1、开启10个线程，在disLocks节点下各自创建名为sub的EPHEMERAL_SEQUENTIAL节点；
2、获取disLocks节点下所有子节点，排序，如果自己的节点编号最小，则获取锁；
3、否则watch排在自己前面的节点，监听到其删除后，进入第2步（重新检测排序是防止监听的节点发生连接失效，导致的节点删除情况）；
4、删除自身sub节点，释放连接；

package zookeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.util.List;
import java.io.IOException;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;

public class DistributedLock implements Watcher{
    private int threadId;
    private ZooKeeper zk = null;
    private String selfPath;
    private String waitPath;
    private String LOG_PREFIX_OF_THREAD;
    private static final int SESSION_TIMEOUT = 10000;
    private static final String GROUP_PATH = "/disLocks";
    private static final String SUB_PATH = "/disLocks/sub";
    private static final String CONNECTION_STRING = "192.168.*.*:2181";
    
    private static final int THREAD_NUM = 10; 
    //确保连接zk成功；
    private CountDownLatch connectedSemaphore = new CountDownLatch(1);
    //确保所有线程运行结束；
    private static final CountDownLatch threadSemaphore = new CountDownLatch(THREAD_NUM);
    private static final Logger LOG = LoggerFactory.getLogger(AllZooKeeperWatcher.class);
    public DistributedLock(int id) {
        this.threadId = id;
        LOG_PREFIX_OF_THREAD = "【第"+threadId+"个线程】";
    }
    public static void main(String[] args) {
        for(int i=0; i < THREAD_NUM; i++){
            final int threadId = i+1;
            new Thread(){
                @Override
                public void run() {
                    try{
                        DistributedLock dc = new DistributedLock(threadId);
                        dc.createConnection(CONNECTION_STRING, SESSION_TIMEOUT);
                        //GROUP_PATH不存在的话，由一个线程创建即可；
                        synchronized (threadSemaphore){
                            dc.createPath(GROUP_PATH, "该节点由线程" + threadId + "创建", true);
                        }
                        dc.getLock();
                    } catch (Exception e){
                        LOG.error("【第"+threadId+"个线程】 抛出的异常：");
                        e.printStackTrace();
                    }
                }
            }.start();
        }
        try {
            threadSemaphore.await();
            LOG.info("所有线程运行结束!");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    /**
     * 获取锁
     * @return
     */
    private void getLock() throws KeeperException, InterruptedException {
        selfPath = zk.create(SUB_PATH,null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        LOG.info(LOG_PREFIX_OF_THREAD+"创建锁路径:"+selfPath);
        if(checkMinPath()){
            getLockSuccess();
        }
    }
    /**
     * 创建节点
     * @param path 节点path
     * @param data 初始数据内容
     * @return
     */
    public boolean createPath( String path, String data, boolean needWatch) throws KeeperException, InterruptedException {
        if(zk.exists(path, needWatch)==null){
            LOG.info( LOG_PREFIX_OF_THREAD + "节点创建成功, Path: "
                    + this.zk.create( path,
                    data.getBytes(),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.PERSISTENT )
                    + ", content: " + data );
        }
        return true;
    }
    /**
     * 创建ZK连接
     * @param connectString	 ZK服务器地址列表
     * @param sessionTimeout Session超时时间
     */
    public void createConnection( String connectString, int sessionTimeout ) throws IOException, InterruptedException {
            zk = new ZooKeeper( connectString, sessionTimeout, this);
            connectedSemaphore.await();
    }
    /**
     * 获取锁成功
    */
    public void getLockSuccess() throws KeeperException, InterruptedException {
        if(zk.exists(this.selfPath,false) == null){
            LOG.error(LOG_PREFIX_OF_THREAD+"本节点已不在了...");
            return;
        }
        LOG.info(LOG_PREFIX_OF_THREAD + "获取锁成功，赶紧干活！");
        Thread.sleep(2000);
        LOG.info(LOG_PREFIX_OF_THREAD + "删除本节点："+selfPath);
        zk.delete(this.selfPath, -1);
        releaseConnection();
        threadSemaphore.countDown();
    }
    /**
     * 关闭ZK连接
     */
    public void releaseConnection() {
        if ( this.zk !=null ) {
            try {
                this.zk.close();
            } catch ( InterruptedException e ) {}
        }
        LOG.info(LOG_PREFIX_OF_THREAD + "释放连接");
    }
    /**
     * 检查自己是不是最小的节点
     * @return
     */
    public boolean checkMinPath() throws KeeperException, InterruptedException {
         List<String> subNodes = zk.getChildren(GROUP_PATH, false);
         Collections.sort(subNodes);
         int index = subNodes.indexOf( selfPath.substring(GROUP_PATH.length()+1));
         switch (index){
             case -1:{
                 LOG.error(LOG_PREFIX_OF_THREAD+"本节点已不在了..."+selfPath);
                 return false;
             }
             case 0:{
                 LOG.info(LOG_PREFIX_OF_THREAD+"子节点中，我果然是老大"+selfPath);
                 return true;
             }
             default:{
                 this.waitPath = GROUP_PATH +"/"+ subNodes.get(index - 1);
                 LOG.info(LOG_PREFIX_OF_THREAD+"获取子节点中，排在我前面的"+waitPath);
                 try{
                     zk.getData(waitPath, true, new Stat());
                     return false;
                 }catch(KeeperException e){
                     if(zk.exists(waitPath,false) == null){
                         LOG.info(LOG_PREFIX_OF_THREAD+"子节点中，排在我前面的"+waitPath+"已失踪，幸福来得太突然?");
                         return checkMinPath();
                     }else{
                         throw e;
                     }
                 }
             }
                 
         }
     
    }
    @Override
    public void process(WatchedEvent event) {
        if(event == null){
            return;
        }
        Event.KeeperState keeperState = event.getState();
        Event.EventType eventType = event.getType();
        if ( Event.KeeperState.SyncConnected == keeperState) {
            if ( Event.EventType.None == eventType ) {
                LOG.info( LOG_PREFIX_OF_THREAD + "成功连接上ZK服务器" );
                connectedSemaphore.countDown();
            }else if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
                LOG.info(LOG_PREFIX_OF_THREAD + "收到情报，排我前面的家伙已挂，我是不是可以出山了？");
                try {
                    if(checkMinPath()){
                        getLockSuccess();
                    }
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }else if ( Event.KeeperState.Disconnected == keeperState ) {
            LOG.info( LOG_PREFIX_OF_THREAD + "与ZK服务器断开连接" );
        } else if ( Event.KeeperState.AuthFailed == keeperState ) {
            LOG.info( LOG_PREFIX_OF_THREAD + "权限检查失败" );
        } else if ( Event.KeeperState.Expired == keeperState ) {
            LOG.info( LOG_PREFIX_OF_THREAD + "会话失效" );
        }
    }
}

log配置文件：
# DEFAULT 
log4j.rootLogger=INFO,CONSOLE

#
# Log INFO level and above messages to the console
#
log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
log4j.appender.CONSOLE.Threshold=INFO
log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} - %m%n


log4j.appender.COMMONSTAT=org.apache.log4j.DailyRollingFileAppender
log4j.appender.COMMONSTAT.Threshold=INFO
log4j.appender.COMMONSTAT.File=/home/zookeeper/zookeeper-test-agent/logs/test.log
log4j.appender.COMMONSTAT.DatePattern='.'yyyy-MM-dd

log4j.appender.COMMONSTAT.layout=org.apache.log4j.PatternLayout
log4j.appender.COMMONSTAT.layout.ConversionPattern=[%d{yyyy-MM-dd HH:mm:ss}] - %m%n

log4j.logger.org.displaytag=WARN
log4j.logger.org.apache.zookeeper=ERROR
log4j.logger.org.springframework=WARN
log4j.logger.org.I0Itec=WARN
log4j.logger.commonStat=INFO,COMMONSTAT

运行结果：
2014-11-19 11:34:10,894 - 【第9个线程】成功连接上ZK服务器
2014-11-19 11:34:10,895 - 【第8个线程】成功连接上ZK服务器
2014-11-19 11:34:10,894 - 【第1个线程】成功连接上ZK服务器
2014-11-19 11:34:10,894 - 【第7个线程】成功连接上ZK服务器
2014-11-19 11:34:10,894 - 【第4个线程】成功连接上ZK服务器
2014-11-19 11:34:10,895 - 【第5个线程】成功连接上ZK服务器
2014-11-19 11:34:10,896 - 【第2个线程】成功连接上ZK服务器
2014-11-19 11:34:10,894 - 【第10个线程】成功连接上ZK服务器
2014-11-19 11:34:10,894 - 【第3个线程】成功连接上ZK服务器
2014-11-19 11:34:10,895 - 【第6个线程】成功连接上ZK服务器
2014-11-19 11:34:10,910 - 【第9个线程】节点创建成功, Path: /disLocks, content: 该节点由线程9创建
2014-11-19 11:34:10,912 - 【第9个线程】创建锁路径:/disLocks/sub0000000000
2014-11-19 11:34:10,917 - 【第6个线程】创建锁路径:/disLocks/sub0000000001
2014-11-19 11:34:10,917 - 【第9个线程】子节点中，我果然是老大/disLocks/sub0000000000
2014-11-19 11:34:10,921 - 【第3个线程】创建锁路径:/disLocks/sub0000000002
2014-11-19 11:34:10,922 - 【第6个线程】获取子节点中，排在我前面的/disLocks/sub0000000000
2014-11-19 11:34:10,923 - 【第9个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:10,924 - 【第10个线程】创建锁路径:/disLocks/sub0000000003
2014-11-19 11:34:10,924 - 【第3个线程】获取子节点中，排在我前面的/disLocks/sub0000000001
2014-11-19 11:34:10,928 - 【第10个线程】获取子节点中，排在我前面的/disLocks/sub0000000002
2014-11-19 11:34:10,929 - 【第1个线程】创建锁路径:/disLocks/sub0000000004
2014-11-19 11:34:10,932 - 【第5个线程】创建锁路径:/disLocks/sub0000000005
2014-11-19 11:34:10,935 - 【第1个线程】获取子节点中，排在我前面的/disLocks/sub0000000003
2014-11-19 11:34:10,936 - 【第2个线程】创建锁路径:/disLocks/sub0000000006
2014-11-19 11:34:10,936 - 【第5个线程】获取子节点中，排在我前面的/disLocks/sub0000000004
2014-11-19 11:34:10,940 - 【第4个线程】创建锁路径:/disLocks/sub0000000007
2014-11-19 11:34:10,941 - 【第2个线程】获取子节点中，排在我前面的/disLocks/sub0000000005
2014-11-19 11:34:10,943 - 【第8个线程】创建锁路径:/disLocks/sub0000000008
2014-11-19 11:34:10,944 - 【第4个线程】获取子节点中，排在我前面的/disLocks/sub0000000006
2014-11-19 11:34:10,945 - 【第7个线程】创建锁路径:/disLocks/sub0000000009
2014-11-19 11:34:10,946 - 【第8个线程】获取子节点中，排在我前面的/disLocks/sub0000000007
2014-11-19 11:34:10,947 - 【第7个线程】获取子节点中，排在我前面的/disLocks/sub0000000008
2014-11-19 11:34:12,923 - 【第9个线程】删除本节点：/disLocks/sub0000000000
2014-11-19 11:34:12,926 - 【第6个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:12,928 - 【第6个线程】子节点中，我果然是老大/disLocks/sub0000000001
2014-11-19 11:34:12,930 - 【第9个线程】释放连接
2014-11-19 11:34:12,930 - 【第6个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:14,930 - 【第6个线程】删除本节点：/disLocks/sub0000000001
2014-11-19 11:34:14,937 - 【第3个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:14,941 - 【第3个线程】子节点中，我果然是老大/disLocks/sub0000000002
2014-11-19 11:34:14,943 - 【第6个线程】释放连接
2014-11-19 11:34:14,946 - 【第3个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:16,946 - 【第3个线程】删除本节点：/disLocks/sub0000000002
2014-11-19 11:34:16,949 - 【第10个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:16,951 - 【第10个线程】子节点中，我果然是老大/disLocks/sub0000000003
2014-11-19 11:34:16,953 - 【第3个线程】释放连接
2014-11-19 11:34:16,953 - 【第10个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:18,953 - 【第10个线程】删除本节点：/disLocks/sub0000000003
2014-11-19 11:34:18,957 - 【第1个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:18,960 - 【第10个线程】释放连接
2014-11-19 11:34:18,961 - 【第1个线程】子节点中，我果然是老大/disLocks/sub0000000004
2014-11-19 11:34:18,964 - 【第1个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:20,964 - 【第1个线程】删除本节点：/disLocks/sub0000000004
2014-11-19 11:34:20,967 - 【第5个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:20,969 - 【第5个线程】子节点中，我果然是老大/disLocks/sub0000000005
2014-11-19 11:34:20,971 - 【第1个线程】释放连接
2014-11-19 11:34:20,971 - 【第5个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:22,971 - 【第5个线程】删除本节点：/disLocks/sub0000000005
2014-11-19 11:34:22,974 - 【第2个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:22,978 - 【第2个线程】子节点中，我果然是老大/disLocks/sub0000000006
2014-11-19 11:34:22,979 - 【第5个线程】释放连接
2014-11-19 11:34:22,981 - 【第2个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:24,981 - 【第2个线程】删除本节点：/disLocks/sub0000000006
2014-11-19 11:34:24,985 - 【第4个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:24,989 - 【第2个线程】释放连接
2014-11-19 11:34:24,989 - 【第4个线程】子节点中，我果然是老大/disLocks/sub0000000007
2014-11-19 11:34:24,995 - 【第4个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:26,995 - 【第4个线程】删除本节点：/disLocks/sub0000000007
2014-11-19 11:34:26,998 - 【第8个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:27,000 - 【第8个线程】子节点中，我果然是老大/disLocks/sub0000000008
2014-11-19 11:34:27,004 - 【第8个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:27,004 - 【第4个线程】释放连接
2014-11-19 11:34:29,004 - 【第8个线程】删除本节点：/disLocks/sub0000000008
2014-11-19 11:34:29,007 - 【第7个线程】收到情报，排我前面的家伙已挂，我是不是可以出山了？
2014-11-19 11:34:29,009 - 【第7个线程】子节点中，我果然是老大/disLocks/sub0000000009
2014-11-19 11:34:29,010 - 【第8个线程】释放连接
2014-11-19 11:34:29,011 - 【第7个线程】获取锁成功，赶紧干活！
2014-11-19 11:34:31,011 - 【第7个线程】删除本节点：/disLocks/sub0000000009
2014-11-19 11:34:31,017 - 【第7个线程】释放连接
2014-11-19 11:34:31,017 - 所有线程运行结束!



基于zookeeper简单实现分布式锁

标签：des   style   blog   io   ar   os   sp   java   for   

原文地址：http://blog.csdn.net/desilting/article/details/41280869
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！