ZooKeeper實(shí)現(xiàn)分布式鎖

1.什么是分布式鎖

一般的鎖:一般我們說(shuō)的鎖是但進(jìn)程多線(xiàn)程的鎖,在多線(xiàn)程并發(fā)編程中,用于線(xiàn)程之間的數(shù)據(jù)同步,保護(hù)共享資源的訪(fǎng)問(wèn)

分布式鎖:分布式鎖指的是在分布式環(huán)境下,保護(hù)跨進(jìn)程,跨主機(jī),跨網(wǎng)絡(luò)的共享資源,實(shí)現(xiàn)互斥訪(fǎng)問(wèn),保證一致性

2.分布式鎖的架構(gòu)圖

20160320214246073.png

3.分布式鎖的算法流程

20160320221558695.png
package com.jike.lock;
 
import java.util.concurrent.TimeUnit;
 
public interface DistributedLock {
    
    /*
     * 獲取鎖,如果沒(méi)有得到就等待
     */
    public void acquire() throws Exception;
 
    /*
     * 獲取鎖,直到超時(shí)
     */
    public boolean acquire(long time, TimeUnit unit) throws Exception;
 
    /*
     * 釋放鎖
     */
    public void release() throws Exception;
 
 
}
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
 
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
 
public class BaseDistributedLock {
    
    private final ZkClientExt client;
    private final String  path;
    
    //zookeeper中l(wèi)ocker節(jié)點(diǎn)的路徑
    private final String  basePath;
    private final String  lockName;
    private static final Integer  MAX_RETRY_COUNT = 10;
        
    public BaseDistributedLock(ZkClientExt client, String path, String lockName){
 
        this.client = client;
        this.basePath = path;
        this.path = path.concat("/").concat(lockName);      
        this.lockName = lockName;
        
    }
    
    private void deleteOurPath(String ourPath) throws Exception{
        client.delete(ourPath);
    }
    
    private String createLockNode(ZkClient client,  String path) throws Exception{
        
        return client.createEphemeralSequential(path, null);
    }
    
    private boolean waitToLock(long startMillis, Long millisToWait, String ourPath) throws Exception{
        
        boolean  haveTheLock = false;
        boolean  doDelete = false;
        
        try
        {
 
            while ( !haveTheLock )
            {
                //獲取lock節(jié)點(diǎn)下的所有節(jié)點(diǎn)
                List<String> children = getSortedChildren();
                String sequenceNodeName = ourPath.substring(basePath.length()+1);
 
                //獲取當(dāng)前節(jié)點(diǎn)的在所有節(jié)點(diǎn)列表中的位置
                int  ourIndex = children.indexOf(sequenceNodeName);
                //節(jié)點(diǎn)位置小于0,說(shuō)明沒(méi)有找到節(jié)點(diǎn)
                if ( ourIndex<0 ){
                    throw new ZkNoNodeException("節(jié)點(diǎn)沒(méi)有找到: " + sequenceNodeName);
                }
                
                //節(jié)點(diǎn)位置大于0說(shuō)明還有其他節(jié)點(diǎn)在當(dāng)前的節(jié)點(diǎn)前面,就需要等待其他的節(jié)點(diǎn)都釋放
                boolean isGetTheLock = ourIndex == 0;
                String  pathToWatch = isGetTheLock ? null : children.get(ourIndex - 1);
 
                if ( isGetTheLock ){
                    
                    haveTheLock = true;
                    
                }else{
                    /**
                     * 獲取當(dāng)前節(jié)點(diǎn)的次小的節(jié)點(diǎn),并監(jiān)聽(tīng)節(jié)點(diǎn)的變化
                     */
                    String  previousSequencePath = basePath .concat( "/" ) .concat( pathToWatch );
                    final CountDownLatch    latch = new CountDownLatch(1);
                    final IZkDataListener previousListener = new IZkDataListener() {
                        
                        public void handleDataDeleted(String dataPath) throws Exception {
                            latch.countDown();          
                        }
                        
                        public void handleDataChange(String dataPath, Object data) throws Exception {
                            // ignore                                   
                        }
                    };
 
                    try 
                    {                  
                        //如果節(jié)點(diǎn)不存在會(huì)出現(xiàn)異常
                        client.subscribeDataChanges(previousSequencePath, previousListener);
                        
                        if ( millisToWait != null )
                        {
                            millisToWait -= (System.currentTimeMillis() - startMillis);
                            startMillis = System.currentTimeMillis();
                            if ( millisToWait <= 0 )
                            {
                                doDelete = true;    // timed out - delete our node
                                break;
                            }
 
                            latch.await(millisToWait, TimeUnit.MICROSECONDS);
                        }
                        else
                        {
                            latch.await();
                        }
                    }
                    catch ( ZkNoNodeException e ) 
                    {
                        //ignore
                    }finally{
                        client.unsubscribeDataChanges(previousSequencePath, previousListener);
                    }
 
                }
            }
        }
        catch ( Exception e )
        {
            //發(fā)生異常需要?jiǎng)h除節(jié)點(diǎn)
            doDelete = true;
            throw e;
        }
        finally
        {
            //如果需要?jiǎng)h除節(jié)點(diǎn)
            if ( doDelete )
            {
                deleteOurPath(ourPath);
            }
        }
        return haveTheLock;
    }
    
    private String getLockNodeNumber(String str, String lockName)
    {
        int index = str.lastIndexOf(lockName);
        if ( index >= 0 )
        {
            index += lockName.length();
            return index <= str.length() ? str.substring(index) : "";
        }
        return str;
    }
    
    List<String> getSortedChildren() throws Exception
    {
        try{
            
            List<String> children = client.getChildren(basePath);
            Collections.sort
            (
                children,
                new Comparator<String>()
                {
                    public int compare(String lhs, String rhs)
                    {
                        return getLockNodeNumber(lhs, lockName).compareTo(getLockNodeNumber(rhs, lockName));
                    }
                }
            );
            return children;
            
        }catch(ZkNoNodeException e){
            
            client.createPersistent(basePath, true);
            return getSortedChildren();
            
        }
    }
    
    protected void releaseLock(String lockPath) throws Exception{
        deleteOurPath(lockPath);    
        
    }
    
    /**
     * 嘗試獲取鎖
     * @param time
     * @param unit
     * @return
     * @throws Exception
     */
    protected String attemptLock(long time, TimeUnit unit) throws Exception{
        
        final long      startMillis = System.currentTimeMillis();
        final Long      millisToWait = (unit != null) ? unit.toMillis(time) : null;
 
        String          ourPath = null;
        boolean         hasTheLock = false;
        boolean         isDone = false;
        int             retryCount = 0;
        
        //網(wǎng)絡(luò)閃斷需要重試一試
        while ( !isDone )
        {
            isDone = true;
 
            try
            {
                ourPath = createLockNode(client, path);
                hasTheLock = waitToLock(startMillis, millisToWait, ourPath);
            }
            catch ( ZkNoNodeException e )
            {
                if ( retryCount++ < MAX_RETRY_COUNT )
                {
                    isDone = false;
                }
                else
                {
                    throw e;
                }
            }
        }
        if ( hasTheLock )
        {
            return ourPath;
        }
 
        return null;
    }
    
    
}
 
import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer;
 
public class TestDistributedLock {
    
    public static void main(String[] args) {
        
        final ZkClientExt zkClientExt1 = new ZkClientExt("192.168.1.105:2181", 5000, 5000, new BytesPushThroughSerializer());
        final SimpleDistributedLockMutex mutex1 = new SimpleDistributedLockMutex(zkClientExt1, "/Mutex");
        
        final ZkClientExt zkClientExt2 = new ZkClientExt("192.168.1.105:2181", 5000, 5000, new BytesPushThroughSerializer());
        final SimpleDistributedLockMutex mutex2 = new SimpleDistributedLockMutex(zkClientExt2, "/Mutex");
        
        try {
            mutex1.acquire();
            System.out.println("Client1 locked");
            Thread client2Thd = new Thread(new Runnable() {
                
                public void run() {
                    try {
                        mutex2.acquire();
                        System.out.println("Client2 locked");
                        mutex2.release();
                        System.out.println("Client2 released lock");
                        
                    } catch (Exception e) {
                        e.printStackTrace();
                    }               
                }
            });
            client2Thd.start();
            Thread.sleep(5000);
            mutex1.release();           
            System.out.println("Client1 released lock");
            
            client2Thd.join();
            
        } catch (Exception e) {
 
            e.printStackTrace();
        }
        
    }
 
}

原文鏈接:https://blog.csdn.net/ZuoAnYinXiang/article/details/50938430

最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀(guān)點(diǎn),簡(jiǎn)書(shū)系信息發(fā)布平臺(tái),僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容