分布式鎖的三種實現(xiàn)方式
作者：BarryWang
www.cnblogs.com/barrywxx/p/11644803.html
分布式鎖三種實現(xiàn)方式：
1. 基于數(shù)據(jù)庫實現(xiàn)分布式鎖；
2. 基于緩存（Redis等）實現(xiàn)分布式鎖；
3. 基于Zookeeper實現(xiàn)分布式鎖；
一， 基于數(shù)據(jù)庫實現(xiàn)分布式鎖
?
1. 悲觀鎖
利用select … where … for update 排他鎖
注意: 其他附加功能與實現(xiàn)一基本一致，這里需要注意的是“where name=lock ”，name字段必須要走索引，否則會鎖表。有些情況下，比如表不大，mysql優(yōu)化器會不走這個索引，導(dǎo)致鎖表問題。
?
2. 樂觀鎖
所謂樂觀鎖與前邊最大區(qū)別在于基于CAS思想，是不具有互斥性，不會產(chǎn)生鎖等待而消耗資源，操作過程中認為不存在并發(fā)沖突，只有update version失敗后才能覺察到。我們的搶購、秒殺就是用了這種實現(xiàn)以防止超賣。
通過增加遞增的版本號字段實現(xiàn)樂觀鎖
?
二， 基于緩存（Redis等）實現(xiàn)分布式鎖
?
1. 使用命令介紹：
（1）SETNX
SETNX key val：當且僅當key不存在時，set一個key為val的字符串，返回1；若key存在，則什么都不做，返回0。
（2）expire
expire key timeout：為key設(shè)置一個超時時間，單位為second，超過這個時間鎖會自動釋放，避免死鎖。
（3）delete
delete key：刪除key
在使用Redis實現(xiàn)分布式鎖的時候，主要就會使用到這三個命令。
?
2. 實現(xiàn)思想：
（1）獲取鎖的時候，使用setnx加鎖，并使用expire命令為鎖添加一個超時時間，超過該時間則自動釋放鎖，鎖的value值為一個隨機生成的UUID，通過此在釋放鎖的時候進行判斷。
（2）獲取鎖的時候還設(shè)置一個獲取的超時時間，若超過這個時間則放棄獲取鎖。
（3）釋放鎖的時候，通過UUID判斷是不是該鎖，若是該鎖，則執(zhí)行delete進行鎖釋放。
?
3. 分布式鎖的簡單實現(xiàn)代碼：
/*** 分布式鎖的簡單實現(xiàn)代碼 4 */public class DistributedLock {private final JedisPool jedisPool;public DistributedLock(JedisPool jedisPool) {this.jedisPool = jedisPool;}/*** 加鎖* @param lockName 鎖的key* @param acquireTimeout 獲取超時時間* @param timeout 鎖的超時時間* @return 鎖標識*/public String lockWithTimeout(String lockName, long acquireTimeout, long timeout) {Jedis conn = null;String retIdentifier = null;try {// 獲取連接conn = jedisPool.getResource();// 隨機生成一個valueString identifier = UUID.randomUUID().toString();// 鎖名，即key值String lockKey = "lock:" + lockName;// 超時時間，上鎖后超過此時間則自動釋放鎖int lockExpire = (int) (timeout / 1000);// 獲取鎖的超時時間，超過這個時間則放棄獲取鎖long end = System.currentTimeMillis() + acquireTimeout;while (System.currentTimeMillis() < end) {if (conn.setnx(lockKey, identifier) == 1) {conn.expire(lockKey, lockExpire);// 返回value值，用于釋放鎖時間確認retIdentifier = identifier;return retIdentifier;}// 返回-1代表key沒有設(shè)置超時時間，為key設(shè)置一個超時時間if (conn.ttl(lockKey) == -1) {conn.expire(lockKey, lockExpire);}try {Thread.sleep(10);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}} catch (JedisException e) {e.printStackTrace();} finally {if (conn != null) {conn.close();}}return retIdentifier;}/*** 釋放鎖* @param lockName 鎖的key* @param identifier 釋放鎖的標識* @return*/public boolean releaseLock(String lockName, String identifier) {Jedis conn = null;String lockKey = "lock:" + lockName;boolean retFlag = false;try {conn = jedisPool.getResource();while (true) {// 監(jiān)視lock，準備開始事務(wù)conn.watch(lockKey);// 通過前面返回的value值判斷是不是該鎖，若是該鎖，則刪除，釋放鎖if (identifier.equals(conn.get(lockKey))) {Transaction transaction = conn.multi();transaction.del(lockKey);Listif (results == null) {continue;}retFlag = true;}conn.unwatch();break;}} catch (JedisException e) {e.printStackTrace();} finally {if (conn != null) {conn.close();}}return retFlag;}}
?
4.?測試剛才實現(xiàn)的分布式鎖
例子中使用50個線程模擬秒殺一個商品，使用–運算符來實現(xiàn)商品減少，從結(jié)果有序性就可以看出是否為加鎖狀態(tài)。
模擬秒殺服務(wù)，在其中配置了jedis線程池，在初始化的時候傳給分布式鎖，供其使用。
public class Service {private static JedisPool pool = null;private DistributedLock lock = new DistributedLock(pool);int n = 500;static {JedisPoolConfig config = new JedisPoolConfig();// 設(shè)置最大連接數(shù)config.setMaxTotal(200);// 設(shè)置最大空閑數(shù)config.setMaxIdle(8);// 設(shè)置最大等待時間config.setMaxWaitMillis(1000 * 100);// 在borrow一個jedis實例時，是否需要驗證，若為true，則所有jedis實例均是可用的config.setTestOnBorrow(true);pool = new JedisPool(config, "127.0.0.1", 6379, 3000);}public void seckill() {// 返回鎖的value值，供釋放鎖時候進行判斷String identifier = lock.lockWithTimeout("resource", 5000, 1000);System.out.println(Thread.currentThread().getName() + "獲得了鎖");System.out.println(--n);lock.releaseLock("resource", identifier);}}
?
模擬線程進行秒殺服務(wù);
public class ThreadA extends Thread {private Service service;public ThreadA(Service service) {this.service = service;}public void run() {service.seckill();}}public class Test {public static void main(String[] args) {Service service = new Service();for (int i = 0; i < 50; i++) {ThreadA threadA = new ThreadA(service);threadA.start();}}}
結(jié)果如下，結(jié)果為有序的：
若注釋掉使用鎖的部分：
public void seckill() {// 返回鎖的value值，供釋放鎖時候進行判斷//String indentifier = lock.lockWithTimeout("resource", 5000, 1000);System.out.println(Thread.currentThread().getName() + "獲得了鎖");System.out.println(--n);//lock.releaseLock("resource", indentifier);}
從結(jié)果可以看出，有一些是異步進行的：
?
三， 基于Zookeeper實現(xiàn)分布式鎖
?
ZooKeeper是一個為分布式應(yīng)用提供一致性服務(wù)的開源組件，它內(nèi)部是一個分層的文件系統(tǒng)目錄樹結(jié)構(gòu)，規(guī)定同一個目錄下只能有一個唯一文件名。基于ZooKeeper實現(xiàn)分布式鎖的步驟如下：
（1）創(chuàng)建一個目錄mylock；
（2）線程A想獲取鎖就在mylock目錄下創(chuàng)建臨時順序節(jié)點；
（3）獲取mylock目錄下所有的子節(jié)點，然后獲取比自己小的兄弟節(jié)點，如果不存在，則說明當前線程順序號最小，獲得鎖；
（4）線程B獲取所有節(jié)點，判斷自己不是最小節(jié)點，設(shè)置監(jiān)聽比自己次小的節(jié)點；
（5）線程A處理完，刪除自己的節(jié)點，線程B監(jiān)聽到變更事件，判斷自己是不是最小的節(jié)點，如果是則獲得鎖。
這里推薦一個Apache的開源庫Curator，它是一個ZooKeeper客戶端，Curator提供的InterProcessMutex是分布式鎖的實現(xiàn)，acquire方法用于獲取鎖，release方法用于釋放鎖。
實現(xiàn)源碼如下：
import lombok.extern.slf4j.Slf4j;import org.apache.commons.lang.StringUtils;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.retry.RetryNTimes;import org.apache.zookeeper.CreateMode;import org.apache.zookeeper.data.Stat;import org.springframework.beans.factory.annotation.Value;import org.springframework.context.annotation.Bean;import org.springframework.stereotype.Component;/*** 分布式鎖Zookeeper實現(xiàn)**/4jpublic class ZkLock implements DistributionLock {private String zkAddress = "zk_adress";private static final String root = "package root";private CuratorFramework zkClient;private final String LOCK_PREFIX = "/lock_";public DistributionLock initZkLock() {if (StringUtils.isBlank(root)) {throw new RuntimeException("zookeeper 'root' can't be null");}zkClient = CuratorFrameworkFactory.builder().connectString(zkAddress).retryPolicy(new RetryNTimes(2000, 20000)).namespace(root).build();zkClient.start();return this;}public boolean tryLock(String lockName) {lockName = LOCK_PREFIX+lockName;boolean locked = true;try {Stat stat = zkClient.checkExists().forPath(lockName);if (stat == null) {log.info("tryLock:{}", lockName);stat = zkClient.checkExists().forPath(lockName);if (stat == null) {zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(lockName, "1".getBytes());} else {log.warn("double-check stat.version:{}", stat.getAversion());locked = false;}} else {log.warn("check stat.version:{}", stat.getAversion());locked = false;}} catch (Exception e) {locked = false;}return locked;}public boolean tryLock(String key, long timeout) {return false;}public void release(String lockName) {lockName = LOCK_PREFIX+lockName;try {zkClient.delete().guaranteed().deletingChildrenIfNeeded().forPath(lockName);log.info("release:{}", lockName);} catch (Exception e) {log.error("刪除", e);}}public void setZkAddress(String zkAddress) {this.zkAddress = zkAddress;}}
優(yōu)點：具備高可用、可重入、阻塞鎖特性，可解決失效死鎖問題。
缺點：因為需要頻繁的創(chuàng)建和刪除節(jié)點，性能上不如Redis方式。
?
四，對比
數(shù)據(jù)庫分布式鎖實現(xiàn)
缺點：
1.db操作性能較差，并且有鎖表的風險
2.非阻塞操作失敗后，需要輪詢，占用cpu資源;
3.長時間不commit或者長時間輪詢，可能會占用較多連接資源
Redis(緩存)分布式鎖實現(xiàn)
缺點：
1.鎖刪除失敗 過期時間不好控制
2.非阻塞，操作失敗后，需要輪詢，占用cpu資源;
ZK分布式鎖實現(xiàn)
缺點：性能不如redis實現(xiàn)，主要原因是寫操作（獲取鎖釋放鎖）都需要在Leader上執(zhí)行，然后同步到follower。
總之：ZooKeeper有較好的性能和可靠性。
?
從理解的難易程度角度（從低到高）數(shù)據(jù)庫 > 緩存 > Zookeeper
從實現(xiàn)的復(fù)雜性角度（從低到高）Zookeeper >= 緩存 > 數(shù)據(jù)庫
從性能角度（從高到低）緩存 > Zookeeper >= 數(shù)據(jù)庫
從可靠性角度（從高到低）Zookeeper > 緩存 > 數(shù)據(jù)庫
特別推薦一個分享架構(gòu)+算法的優(yōu)質(zhì)內(nèi)容，還沒關(guān)注的小伙伴，可以長按關(guān)注一下：
長按訂閱更多精彩▼
如有收獲，點個在看，誠摯感謝
免責聲明：本文內(nèi)容由21ic獲得授權(quán)后發(fā)布，版權(quán)歸原作者所有，本平臺僅提供信息存儲服務(wù)。文章僅代表作者個人觀點，不代表本平臺立場，如有問題，請聯(lián)系我們，謝謝！
下載文檔
